chezbetty.get_days_on_shame() - Code Metrics - Inspection of "Add days on wall of shame" - um-cseg/chez-betty - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 9e58f4...6e6bb8 )

by Pat

created 2016-03-24 17:00 UTC

chezbetty.get_days_on_shame() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	5
dl	0
loc	33
rs	8.0894

from decimal import Decimal
import itertools
import qrcode
import qrcode.image.svg
import stripe
import traceback

from pyramid.renderers import render
from pyramid.threadlocal import get_current_registry

try:
    import lxml.etree as ET
except ImportError:
    import xml.etree.ElementTree as ET

import pytz

import arrow
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase

def _send_email(msg, FROM):
    settings = get_current_registry().settings

    if 'debugging' in settings and bool(int(settings['debugging'])):
        if 'debugging_send_email' not in settings or settings['debugging_send_email'] != 'true':
            print("Mail suppressed due to debug settings")
            return

    if 'smtp.host' in settings:
        sm = smtplib.SMTP(host=settings['smtp.host'])
    else:
        sm = smtplib.SMTP()
        sm.connect()
    if 'smtp.username' in settings:
        sm.login(settings['smtp.username'], settings['smtp.password'])

    def gen_sendto(msg):
        send_to = []
        if 'To' in msg:
            send_to += msg['To'].split(', ')
        if 'CC' in msg:
            send_to += msg['CC'].split(', ')
        if 'BCC' in msg:
            send_to += msg['BCC'].split(', ')
        return send_to


    if 'debugging' in settings and bool(int(settings['debugging'])):
        if 'debugging_send_email' in settings and settings['debugging_send_email'] == 'true':
            try:
                msg.replace_header('To', settings['debugging_send_email_to'])
                print("DEBUG: e-mail TO overidden to {}".format(msg['To']))
            except KeyError: pass
            try:
                msg.replace_header('CC', settings['debugging_send_email_to'])
                print("DEBUG: e-mail CC overidden to {}".format(msg['CC']))
            except KeyError: pass
            try:
                msg.replace_header('BCC', settings['debugging_send_email_to'])
                print("DEBUG: e-mail BCC overidden to {}".format(msg['BCC']))
            except KeyError: pass
            send_to = gen_sendto(msg)
            sm.sendmail(FROM, send_to, msg.as_string())
    else:
        send_to = gen_sendto(msg)
        sm.sendmail(FROM, send_to, msg.as_string())
    sm.quit()


def send_email(TO, SUBJECT, body,

        FROM='[email protected]',
        encoding='html',
        ):
    msg = MIMEMultipart()
    msg['Subject'] = SUBJECT
    msg['From'] = FROM
    msg['To'] = TO
    if encoding == 'text':
        msg.attach(MIMEText(body))
    else:
        msg.attach(MIMEText(body, encoding))
    print(msg.as_string())
    _send_email(msg, FROM)


def send_bcc_email(BCC, SUBJECT, body,

        FROM='[email protected]',
        encoding='html',
        ):
    msg = MIMEMultipart()
    msg['Subject'] = SUBJECT
    msg['From'] = FROM
    msg['BCC'] = BCC
    if encoding == 'text':
        msg.attach(MIMEText(body))
    else:
        msg.attach(MIMEText(body, encoding))
    print(msg.as_string())
    _send_email(msg, FROM)


def user_password_reset(user):
    password = user.random_password()
    send_email(TO=user.uniqname+'@umich.edu',
               SUBJECT='Chez Betty Login',
               body=render('templates/admin/email_password.jinja2', {'user': user, 'password': password}))

def notify_new_top_wall_of_shame(user):
    # First, double-check to make sure this user is in enough debt to be on the
    # wall of shame
    # TODO: This threshold should be a configurable parameter
    if user.balance > -5:
        print("Suppressed new wall of shame e-mail b/c user is under the limit")
        return
    send_email(
            TO=user.uniqname+'@umich.edu',
            SUBJECT='[Chez Betty] You are on top of the Wall of Shame! :(',
            body=render('templates/admin/email_new_top_wall_of_shame.jinja2',
                {'user': user})
            )

def notify_pool_out_of_credit(owner, pool):
    send_email(
            TO=owner.uniqname+'@umich.edu',
            SUBJECT='[Chez Betty] Credit limit reached for {} pool'.format(pool.name),
            body=render('templates/admin/email_pool_credit_limit.jinja2',
                {'pool': pool, 'owner': owner})
            )


def new_user_email(user):
    send_email(
            TO=user.uniqname+'@umich.edu',
            SUBJECT='[Chez Betty] Welcome to Chez Betty!',
            body=render('templates/admin/email_new_user.jinja2',
                {'user': user})
            )


def string_to_qrcode(s):
    factory = qrcode.image.svg.SvgPathImage
    img = qrcode.make(s, image_factory=factory, box_size=14,
        version=4,
        border=0)
    img.save('/dev/null')   # This is needed, I swear.
    return ET.tostring(img._img).decode('utf-8')

class InvalidGroupPeriod(Exception):
    pass

def group(rows, period='day'):

    def fix_timezone(i):
        return i.timestamp.to('America/Detroit')
    def group_month(i):
        dt = fix_timezone(i)
        return dt.replace(day=1)
    def group_year(i):
        dt = fix_timezone(i)
        return dt.replace(month=1,day=1)

    if period == 'day':
        group_function = lambda i: fix_timezone(i).date()
    elif period == 'month':
        group_function = group_month
    elif period == 'year':
        group_function = group_year
    elif period == 'month_each':
        group_function = lambda i: fix_timezone(i).month
    elif period == 'day_each':
        group_function = lambda i: fix_timezone(i).day
    elif period == 'weekday_each':
        group_function = lambda i: fix_timezone(i).weekday()
    elif period == 'hour_each':
        group_function = lambda i: fix_timezone(i).hour
    else:
        raise(InvalidGroupPeriod(period))


    if 'each' in period:
        # If we are grouping in a very finite set of bins (like days of the
        # week), then use a hash map instead of a list as a return
        sums = {}
        for row in rows:
            item_period = group_function(row)
            if item_period not in sums:
                sums[item_period] = 0
            sums[item_period] += row.summable

    else:
        # If we are grouping in a long list of things (days over some range)
        # then a list is better.
        sums = []
        for (item_period, items) in itertools.groupby(rows, group_function):
            total = 0
            for item in items:
                total += item.summable
            sums.append((item_period, total))

    return sums


def get_days_on_shame(user, rows):
    directions = {
        'purchase':   -1, # user balance goes down by amount
        'cashdeposit': 1, # user balance goes up by amount
        'ccdeposit':   1,
        'btcdeposit':  1,
        'adjustment':  1
    }

    if user.balance >= 0:
        return 0
    balance = user.balance
    for r in reversed(rows):
        amount = r[0]
        trtype = r[1]
        to_uid = r[2]
        fr_uid = r[3]
        timest = r[4]
        t = timest.timestamp*1000

        # We get the user/pool id from whether we care about where the
        # money came from or went
        if directions[trtype] == -1:
            userid = fr_uid
        else:
            userid = to_uid

        balance = balance - (directions[trtype]*amount)

        if balance > -5:
            break

    return (arrow.now() - timest).days

# Returns an array of tuples where the first item in the tuple is a millisecond
# timestamp and the second item is the total number of things so far.
def timeseries_cumulative(rows):
    total = 0
    out = []

    for r in rows:
        if len(r) == 1:
            total += 1
        else:
            total += r[1]
        t = r[0].timestamp*1000
        out.append((t, total))

    return out

# [(milliseconds, debt, bank_balance, debt/# users in debt), ...]
def timeseries_balance_total_daily(rows):

    # Is debt going away or coming in
    directions = {
        'purchase':   -1, # user balance goes down by amount
        'cashdeposit': 1, # user balance goes up by amount
        'ccdeposit':   1,
        'btcdeposit':  1,
        'adjustment':  1
    }

    user_balances = {}
    total_debt = 0
    total_balance = 0
    users_in_debt = 0

    out = []

    for r in rows:
        amount = r[0]
        trtype = r[1]
        to_uid = r[2]
        fr_uid = r[3]
        timest = r[4]
        t = timest.timestamp*1000

        # We get the user/pool id from whether we care about where the
        # money came from or went
        if directions[trtype] == -1:
            userid = fr_uid
        else:
            userid = to_uid

        # Calculate the new balance so we can compare it to the old
        old_balance = user_balances.get(userid, 0)
        new_balance = old_balance + (directions[trtype]*amount)
        user_balances[userid] = new_balance

        # Look for swings from in debt to not in debt and vice-versa.
        # This is how we update the running totals of debt and bank holdings.
        if old_balance < 0 and new_balance >= 0:
            # Was in debt, now not
            total_debt -= -1*old_balance
            total_balance += new_balance
            users_in_debt -= 1
        elif old_balance >= 0 and new_balance < 0:
            # Wasn't in debt, now is
            total_balance -= old_balance
            total_debt += -1*new_balance
            users_in_debt += 1
        elif new_balance < 0:
            # still in debt
            total_debt += -1*directions[trtype]*amount
        else:
            # hey, in the black!
            total_balance += directions[trtype]*amount

        # Add to output array
        debt_per_user = 0 if users_in_debt == 0 else round((total_debt*100)/users_in_debt)
        out.append((t, round(total_debt*100), round(total_balance*100), debt_per_user))

    return out

def post_stripe_payment(
        datalayer, # Need to pass as argument to avoid circular import, ugh
        request,
        token,
        amount,
        total_cents,
        account_making_payment,
        account_depositing_into,
        ):
    # See http://stripe.com/docs/tutorials/charges
    stripe.api_key = request.registry.settings['stripe.secret_key']

    charge = (amount + Decimal('0.3')) / Decimal('0.971')
    fee = charge - amount
    if total_cents != int(round((amount + fee)*100)):
        print("Stripe total mismatch. total_cents {} != {}".format(
            total_cents, int(round((amount + fee)*100))))
        request.session.flash('Unexpected error processing transaction. Card NOT charged.', 'error')
        return False

    if amount <= 0.0:
        request.session.flash(
                _('Deposit amount must be greater than $0.00. Card NOT charged.'),
                'error'
                )
        return False

    try:
        charge = stripe.Charge.create(
                amount = total_cents,
                currency="usd",
                source=token,
                description=account_making_payment.uniqname+'@umich.edu'
                )

    except stripe.CardError as e:
        traceback.print_exc()
        request.session.flash('Card error processing transaction. Card NOT charged.', 'error')
        return False
    except stripe.StripeError as e:
        traceback.print_exc()
        request.session.flash('Unexpected error processing transaction. Card NOT charged.', 'error')
        request.session.flash('Please e-mail [email protected] so we can correct this error', 'error')
        return False

    try:
        deposit = datalayer.cc_deposit(
                account_making_payment,
                account_depositing_into,
                amount,
                charge['id'],
                charge['source']['last4'])

        request.session.flash('Deposit added successfully.', 'success')

    except Exception as e:
        traceback.print_exc()
        request.session.flash('A unknown error has occured.', 'error')
        request.session.flash('Your card HAS been charged, but your account HAS NOT been credited.', 'error')
        request.session.flash('Please e-mail [email protected] so we can correct this error', 'error')

    return True



1			from decimal import Decimal
2			import itertools
3			import qrcode
4			import qrcode.image.svg
5			import stripe
6			import traceback
7
8			from pyramid.renderers import render
9			from pyramid.threadlocal import get_current_registry
10
11			try:
12			import lxml.etree as ET
13			except ImportError:
14			import xml.etree.ElementTree as ET
15
16			import pytz
17
18			import arrow
19			import smtplib
20			from email.mime.text import MIMEText
21			from email.mime.multipart import MIMEMultipart
22			from email.mime.base import MIMEBase
23
24			def _send_email(msg, FROM):
25			settings = get_current_registry().settings
26
27			if 'debugging' in settings and bool(int(settings['debugging'])):
28			if 'debugging_send_email' not in settings or settings['debugging_send_email'] != 'true':
29			print("Mail suppressed due to debug settings")
30			return
31
32			if 'smtp.host' in settings:
33			sm = smtplib.SMTP(host=settings['smtp.host'])
34			else:
35			sm = smtplib.SMTP()
36			sm.connect()
37			if 'smtp.username' in settings:
38			sm.login(settings['smtp.username'], settings['smtp.password'])
39
40			def gen_sendto(msg):
41			send_to = []
42			if 'To' in msg:
43			send_to += msg['To'].split(', ')
44			if 'CC' in msg:
45			send_to += msg['CC'].split(', ')
46			if 'BCC' in msg:
47			send_to += msg['BCC'].split(', ')
48			return send_to
49
50
51			if 'debugging' in settings and bool(int(settings['debugging'])):
52			if 'debugging_send_email' in settings and settings['debugging_send_email'] == 'true':
53			try:
54			msg.replace_header('To', settings['debugging_send_email_to'])
55			print("DEBUG: e-mail TO overidden to {}".format(msg['To']))
56			except KeyError: pass
57			try:
58			msg.replace_header('CC', settings['debugging_send_email_to'])
59			print("DEBUG: e-mail CC overidden to {}".format(msg['CC']))
60			except KeyError: pass
61			try:
62			msg.replace_header('BCC', settings['debugging_send_email_to'])
63			print("DEBUG: e-mail BCC overidden to {}".format(msg['BCC']))
64			except KeyError: pass
65			send_to = gen_sendto(msg)
66			sm.sendmail(FROM, send_to, msg.as_string())
67			else:
68			send_to = gen_sendto(msg)
69			sm.sendmail(FROM, send_to, msg.as_string())
70			sm.quit()
71
72
73			def send_email(TO, SUBJECT, body,
			0 ignored issues – show Duplication introduced 2015-11-24 16:45 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
74			FROM='[email protected]',
75			encoding='html',
76			):
77			msg = MIMEMultipart()
78			msg['Subject'] = SUBJECT
79			msg['From'] = FROM
80			msg['To'] = TO
81			if encoding == 'text':
82			msg.attach(MIMEText(body))
83			else:
84			msg.attach(MIMEText(body, encoding))
85			print(msg.as_string())
86			_send_email(msg, FROM)
87
88
89			def send_bcc_email(BCC, SUBJECT, body,
			0 ignored issues – show Duplication introduced 2015-11-24 16:45 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
90			FROM='[email protected]',
91			encoding='html',
92			):
93			msg = MIMEMultipart()
94			msg['Subject'] = SUBJECT
95			msg['From'] = FROM
96			msg['BCC'] = BCC
97			if encoding == 'text':
98			msg.attach(MIMEText(body))
99			else:
100			msg.attach(MIMEText(body, encoding))
101			print(msg.as_string())
102			_send_email(msg, FROM)
103
104
105			def user_password_reset(user):
106			password = user.random_password()
107			send_email(TO=user.uniqname+'@umich.edu',
108			SUBJECT='Chez Betty Login',
109			body=render('templates/admin/email_password.jinja2', {'user': user, 'password': password}))
110
111			def notify_new_top_wall_of_shame(user):
112			# First, double-check to make sure this user is in enough debt to be on the
113			# wall of shame
114			# TODO: This threshold should be a configurable parameter
115			if user.balance > -5:
116			print("Suppressed new wall of shame e-mail b/c user is under the limit")
117			return
118			send_email(
119			TO=user.uniqname+'@umich.edu',
120			SUBJECT='[Chez Betty] You are on top of the Wall of Shame! :(',
121			body=render('templates/admin/email_new_top_wall_of_shame.jinja2',
122			{'user': user})
123			)
124
125			def notify_pool_out_of_credit(owner, pool):
126			send_email(
127			TO=owner.uniqname+'@umich.edu',
128			SUBJECT='[Chez Betty] Credit limit reached for {} pool'.format(pool.name),
129			body=render('templates/admin/email_pool_credit_limit.jinja2',
130			{'pool': pool, 'owner': owner})
131			)
132
133
134			def new_user_email(user):
135			send_email(
136			TO=user.uniqname+'@umich.edu',
137			SUBJECT='[Chez Betty] Welcome to Chez Betty!',
138			body=render('templates/admin/email_new_user.jinja2',
139			{'user': user})
140			)
141
142
143			def string_to_qrcode(s):
144			factory = qrcode.image.svg.SvgPathImage
145			img = qrcode.make(s, image_factory=factory, box_size=14,
146			version=4,
147			border=0)
148			img.save('/dev/null') # This is needed, I swear.
149			return ET.tostring(img._img).decode('utf-8')
150
151			class InvalidGroupPeriod(Exception):
152			pass
153
154			def group(rows, period='day'):
155
156			def fix_timezone(i):
157			return i.timestamp.to('America/Detroit')
158			def group_month(i):
159			dt = fix_timezone(i)
160			return dt.replace(day=1)
161			def group_year(i):
162			dt = fix_timezone(i)
163			return dt.replace(month=1,day=1)
164
165			if period == 'day':
166			group_function = lambda i: fix_timezone(i).date()
167			elif period == 'month':
168			group_function = group_month
169			elif period == 'year':
170			group_function = group_year
171			elif period == 'month_each':
172			group_function = lambda i: fix_timezone(i).month
173			elif period == 'day_each':
174			group_function = lambda i: fix_timezone(i).day
175			elif period == 'weekday_each':
176			group_function = lambda i: fix_timezone(i).weekday()
177			elif period == 'hour_each':
178			group_function = lambda i: fix_timezone(i).hour
179			else:
180			raise(InvalidGroupPeriod(period))
181
182
183			if 'each' in period:
184			# If we are grouping in a very finite set of bins (like days of the
185			# week), then use a hash map instead of a list as a return
186			sums = {}
187			for row in rows:
188			item_period = group_function(row)
189			if item_period not in sums:
190			sums[item_period] = 0
191			sums[item_period] += row.summable
192
193			else:
194			# If we are grouping in a long list of things (days over some range)
195			# then a list is better.
196			sums = []
197			for (item_period, items) in itertools.groupby(rows, group_function):
198			total = 0
199			for item in items:
200			total += item.summable
201			sums.append((item_period, total))
202
203			return sums
204
205
206			def get_days_on_shame(user, rows):
207			directions = {
208			'purchase': -1, # user balance goes down by amount
209			'cashdeposit': 1, # user balance goes up by amount
210			'ccdeposit': 1,
211			'btcdeposit': 1,
212			'adjustment': 1
213			}
214
215			if user.balance >= 0:
216			return 0
217			balance = user.balance
218			for r in reversed(rows):
219			amount = r[0]
220			trtype = r[1]
221			to_uid = r[2]
222			fr_uid = r[3]
223			timest = r[4]
224			t = timest.timestamp*1000
225
226			# We get the user/pool id from whether we care about where the
227			# money came from or went
228			if directions[trtype] == -1:
229			userid = fr_uid
230			else:
231			userid = to_uid
232
233			balance = balance - (directions[trtype]*amount)
234
235			if balance > -5:
236			break
237
238			return (arrow.now() - timest).days
239
240			# Returns an array of tuples where the first item in the tuple is a millisecond
241			# timestamp and the second item is the total number of things so far.
242			def timeseries_cumulative(rows):
243			total = 0
244			out = []
245
246			for r in rows:
247			if len(r) == 1:
248			total += 1
249			else:
250			total += r[1]
251			t = r[0].timestamp*1000
252			out.append((t, total))
253
254			return out
255
256			# [(milliseconds, debt, bank_balance, debt/# users in debt), ...]
257			def timeseries_balance_total_daily(rows):
258
259			# Is debt going away or coming in
260			directions = {
261			'purchase': -1, # user balance goes down by amount
262			'cashdeposit': 1, # user balance goes up by amount
263			'ccdeposit': 1,
264			'btcdeposit': 1,
265			'adjustment': 1
266			}
267
268			user_balances = {}
269			total_debt = 0
270			total_balance = 0
271			users_in_debt = 0
272
273			out = []
274
275			for r in rows:
276			amount = r[0]
277			trtype = r[1]
278			to_uid = r[2]
279			fr_uid = r[3]
280			timest = r[4]
281			t = timest.timestamp*1000
282
283			# We get the user/pool id from whether we care about where the
284			# money came from or went
285			if directions[trtype] == -1:
286			userid = fr_uid
287			else:
288			userid = to_uid
289
290			# Calculate the new balance so we can compare it to the old
291			old_balance = user_balances.get(userid, 0)
292			new_balance = old_balance + (directions[trtype]*amount)
293			user_balances[userid] = new_balance
294
295			# Look for swings from in debt to not in debt and vice-versa.
296			# This is how we update the running totals of debt and bank holdings.
297			if old_balance < 0 and new_balance >= 0:
298			# Was in debt, now not
299			total_debt -= -1*old_balance
300			total_balance += new_balance
301			users_in_debt -= 1
302			elif old_balance >= 0 and new_balance < 0:
303			# Wasn't in debt, now is
304			total_balance -= old_balance
305			total_debt += -1*new_balance
306			users_in_debt += 1
307			elif new_balance < 0:
308			# still in debt
309			total_debt += -1directions[trtype]amount
310			else:
311			# hey, in the black!
312			total_balance += directions[trtype]*amount
313
314			# Add to output array
315			debt_per_user = 0 if users_in_debt == 0 else round((total_debt*100)/users_in_debt)
316			out.append((t, round(total_debt100), round(total_balance100), debt_per_user))
317
318			return out
319
320			def post_stripe_payment(
321			datalayer, # Need to pass as argument to avoid circular import, ugh
322			request,
323			token,
324			amount,
325			total_cents,
326			account_making_payment,
327			account_depositing_into,
328			):
329			# See http://stripe.com/docs/tutorials/charges
330			stripe.api_key = request.registry.settings['stripe.secret_key']
331
332			charge = (amount + Decimal('0.3')) / Decimal('0.971')
333			fee = charge - amount
334			if total_cents != int(round((amount + fee)*100)):
335			print("Stripe total mismatch. total_cents {} != {}".format(
336			total_cents, int(round((amount + fee)*100))))
337			request.session.flash('Unexpected error processing transaction. Card NOT charged.', 'error')
338			return False
339
340			if amount <= 0.0:
341			request.session.flash(
342			_('Deposit amount must be greater than $0.00. Card NOT charged.'),
343			'error'
344			)
345			return False
346
347			try:
348			charge = stripe.Charge.create(
349			amount = total_cents,
350			currency="usd",
351			source=token,
352			description=account_making_payment.uniqname+'@umich.edu'
353			)
354
355			except stripe.CardError as e:
356			traceback.print_exc()
357			request.session.flash('Card error processing transaction. Card NOT charged.', 'error')
358			return False
359			except stripe.StripeError as e:
360			traceback.print_exc()
361			request.session.flash('Unexpected error processing transaction. Card NOT charged.', 'error')
362			request.session.flash('Please e-mail [email protected] so we can correct this error', 'error')
363			return False
364
365			try:
366			deposit = datalayer.cc_deposit(
367			account_making_payment,
368			account_depositing_into,
369			amount,
370			charge['id'],
371			charge['source']['last4'])
372
373			request.session.flash('Deposit added successfully.', 'success')
374
375			except Exception as e:
376			traceback.print_exc()
377			request.session.flash('A unknown error has occured.', 'error')
378			request.session.flash('Your card HAS been charged, but your account HAS NOT been credited.', 'error')
379			request.session.flash('Please e-mail [email protected] so we can correct this error', 'error')
380
381			return True
382
383

um-cseg / chez-betty

Push — master ( 9e58f4...6e6bb8 )

chezbetty.get_days_on_shame() B

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like