chezbetty.timeseries_balance_total_daily() - Code Metrics - Inspection of "Add balance over time graph to user detail page" - um-cseg/chez-betty - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 957b4e...55017b )

by Pat

created 2015-12-30 01:20 UTC

chezbetty.timeseries_balance_total_daily() D

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	9
dl	0
loc	62
rs	4.7927

How to fix Long Method

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 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

# 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 + 0.3) / 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
    amount = Decimal(amount)

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

um-cseg / chez-betty

Push — master ( 957b4e...55017b )

chezbetty.timeseries_balance_total_daily() D

Complexity

Size

Duplication

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like