138 lines
3.8 KiB
Python
138 lines
3.8 KiB
Python
import stripe as _stripe
|
|
_stripe.api_version = '2016-06-15'
|
|
|
|
import logging
|
|
import time
|
|
import sys
|
|
import csv
|
|
import codecs
|
|
|
|
from itertools import groupby
|
|
from datetime import datetime, timedelta, date
|
|
from cStringIO import StringIO
|
|
|
|
from app import billing as stripe
|
|
|
|
|
|
def _format_timestamp(stripe_timestamp):
|
|
date_obj = date.fromtimestamp(stripe_timestamp)
|
|
return date_obj.strftime('%m/%d/%Y')
|
|
|
|
|
|
def _format_money(stripe_money):
|
|
return stripe_money/100.0
|
|
|
|
|
|
def list_charges(num_days):
|
|
""" List all charges that have occurred in the past specified number of days.
|
|
"""
|
|
now = datetime.utcnow()
|
|
starting_from = now - timedelta(days=num_days)
|
|
starting_timestamp = str(int(time.mktime(starting_from.timetuple())))
|
|
created = {'gte': starting_timestamp}
|
|
starting_after_kw = {}
|
|
|
|
while True:
|
|
charges = stripe.Charge.list(limit=100, expand=['data.invoice'], created=created,
|
|
**starting_after_kw)
|
|
for charge in charges.data:
|
|
yield charge
|
|
|
|
if not charges.has_more:
|
|
break
|
|
|
|
starting_after_kw = {'starting_after': charge.id}
|
|
|
|
|
|
def _date_key(line_item):
|
|
return line_item.start, line_item.end
|
|
|
|
|
|
def expand_invoice(invoice, total_amount):
|
|
if invoice is None:
|
|
yield total_amount, None, None, None
|
|
else:
|
|
data_iter = groupby(invoice.lines.data, lambda li: (li.period.start, li.period.end))
|
|
for (period_start, period_end), line_items_iter in data_iter:
|
|
line_items = list(line_items_iter)
|
|
period_amount = sum(line_item.amount for line_item in line_items)
|
|
yield period_amount, period_start, period_end, line_items[-1].plan
|
|
|
|
|
|
def format_charge(charge):
|
|
""" Generator which will return one or more line items corresponding to the
|
|
line items for this charge.
|
|
"""
|
|
ch_status = 'Paid'
|
|
if charge.failure_code is not None:
|
|
ch_status = 'Failed'
|
|
elif charge.amount_refunded > 0:
|
|
ch_status = 'Refunded'
|
|
|
|
card = charge.source
|
|
|
|
invoice_iterable = expand_invoice(charge.invoice, charge.amount)
|
|
for line_amount, period_start, period_end, plan in invoice_iterable:
|
|
yield [
|
|
_format_timestamp(charge.created),
|
|
_format_timestamp(period_start) if period_start is not None else None,
|
|
_format_timestamp(period_end) if period_end is not None else None,
|
|
_format_money(line_amount),
|
|
'credit_card',
|
|
ch_status,
|
|
plan.name if plan is not None else None,
|
|
charge.id,
|
|
charge.customer,
|
|
card.address_city,
|
|
card.address_state,
|
|
card.address_country,
|
|
card.address_zip,
|
|
card.country,
|
|
]
|
|
|
|
|
|
class _UnicodeWriter(object):
|
|
"""
|
|
A CSV writer which will write rows to CSV file "f",
|
|
which is encoded in the given encoding.
|
|
"""
|
|
def __init__(self, f, dialect=csv.excel, encoding='utf-8', **kwds):
|
|
# Redirect output to a queue
|
|
self.queue = StringIO()
|
|
self.writer = csv.writer(self.queue, dialect=dialect, **kwds)
|
|
self.stream = f
|
|
self.encoder = codecs.getincrementalencoder(encoding)()
|
|
|
|
@staticmethod
|
|
def _encode_cell(cell):
|
|
if cell is None:
|
|
return cell
|
|
return unicode(cell).encode('utf-8')
|
|
|
|
def writerow(self, row):
|
|
self.writer.writerow([self._encode_cell(s) for s in row])
|
|
# Fetch UTF-8 output from the queue ...
|
|
data = self.queue.getvalue()
|
|
data = data.decode('utf-8')
|
|
# ... and reencode it into the target encoding
|
|
data = self.encoder.encode(data)
|
|
# write to the target stream
|
|
self.stream.write(data)
|
|
# empty queue
|
|
self.queue.truncate(0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
logging.basicConfig(level=logging.WARN)
|
|
|
|
days = 30
|
|
if len(sys.argv) > 1:
|
|
days = int(sys.argv[1])
|
|
|
|
transaction_writer = _UnicodeWriter(sys.stdout)
|
|
rows = (line_item
|
|
for one_charge in list_charges(days)
|
|
for line_item in format_charge(one_charge))
|
|
for row in rows:
|
|
transaction_writer.writerow(row)
|
|
sys.stdout.flush()
|