paypal-reports/paypal_report.py

"""Download subscriber info from PayPal.

Run it like this:

$ export PAYPAL_CLIENT_ID=XXX
$ export PAYPAL_CLIENT_SECRET=YYY
$ python3 paypal_report.py profiles 2021-11-01T00:00:00-0700 2021-11-30T23:59:59-0700 > out.txt
$ python3 paypal_report.py transactions 2021-11-01T00:00:00-0700 2021-11-30T23:59:59-0700 > out.txt

"""
import argparse
import collections
import datetime
import sys
import os

import requests


def main():
    args = parse_args()
    client_id, client_secret = load_paypal_keys()
    access_token = get_paypal_access_token(client_id, client_secret)
    transactions = get_all_transactions(access_token, args.start_date, args.end_date)
    if args.report == 'profiles':
        report_on_unique_profiles(transactions)
    else:
        report_on_transactions(transactions)


def parse_args():
    parser = argparse.ArgumentParser(description='Download PayPal subscriber info.')
    parser.add_argument('report', help='report to run"', choices=['profiles', 'transactions'])
    parser.add_argument('start_date', help='start date inclusive (eg. 2021-11-01T00:00:00-07:00)', type=parse_iso_time)
    parser.add_argument('end_date', help='end date inclusive (eg. 2021-11-30T23:59:59-07:00)', type=parse_iso_time)
    return parser.parse_args()


def parse_iso_time(time_str):
    try:
        time = datetime.datetime.fromisoformat(time_str)
    except ValueError:
        raise argparse.ArgumentTypeError('Date did not match ISO format eg. 2021-11-01T00:00:00-07:00')
    return time


def load_paypal_keys():
    try:
        client_id = os.environ['PAYPAL_CLIENT_ID']
        client_secret = os.environ['PAYPAL_CLIENT_SECRET']
    except KeyError:
        sys.exit('Please set PAYPAL_CLIENT_ID and PAYPAL_CLIENT_SECRET environment variables.')
    return client_id, client_secret


def get_paypal_access_token(client_id, client_secret):
    url = 'https://api-m.paypal.com/v1/oauth2/token?grant_type=client_credentials'
    response = requests.post(
        url,
        auth=(client_id, client_secret),
        headers={
            'Accept': 'application/json',
            'Accept-Language': 'en_US',
        },
    )
    data = response.json()
    return data['access_token']


def get_all_transactions(access_token, start_date, end_date):
    periods = get_time_periods(start_date, end_date)
    transactions = []
    for start, end in periods:
        print(f'Period {start} - {end}.', file=sys.stderr)
        transactions += get_transactions_for_period(access_token, start, end)
    return transactions


def get_time_periods(start_date, end_date):
    """Break a time period into smaller periods of up to 31 days.

    PayPal only supports querying for up to 31 days of transactions at a time.
    """
    periods = []
    while start_date < end_date:
        periods.append((start_date, start_date + datetime.timedelta(days=31)))
        start_date += datetime.timedelta(days=31)
    return periods


def get_transactions_for_period(access_token, start_date, end_date, page=1):
    """Downloads all pages of transactions for a time period of 31 days or less."""
    print(f'Fetching transactions page {page}.', file=sys.stderr)
    url = f'https://api-m.paypal.com/v1/reporting/transactions?start_date={start_date.isoformat()}&end_date={end_date.isoformat()}&fields=all&page_size=500&page={page}'
    response = requests.get(
        url,
        headers={
            'Content-Type': 'application/json',
            'Authorization': f'Bearer {access_token}',
        },
    )
    data = response.json()
    if page < data['total_pages']:
        return data['transaction_details'] + get_transactions_for_period(access_token, start_date, end_date, page + 1)
    else:
        return data['transaction_details']


def report_on_unique_profiles(transactions):
    records = set()
    for t in transactions:
        transaction_info = t['transaction_info']
        if 'paypal_reference_id' in transaction_info:
            records.add(
                (
                    transaction_info['paypal_reference_id'],
                    transaction_info.get('transaction_subject', 'NO TRANSACTION SUBJECT'),
                ),
            )
        else:
            print(f'Skipping transaction {transaction_info["transaction_id"]} with no PayPal Reference ID.', file=sys.stderr)

    for id, desc in sorted(records):
        print(f'{id}   {desc}')
    count = collections.Counter([id for id, _ in records])
    dupes = [id for id, total in count.items() if total > 1]
    print(f'Reference IDs with changing subject: {dupes}', file=sys.stderr)


def report_on_transactions(transactions):
    for t in transactions:
        transaction_info = t['transaction_info']
        if 'paypal_reference_id' in transaction_info:
            print("%-19s %s" % (transaction_info['transaction_id'], transaction_info['paypal_reference_id']))
            print("%-19s %s" % (transaction_info['paypal_reference_id'], transaction_info['paypal_reference_id']))
        else:
            print(f'Skipping transaction {transaction_info["transaction_id"]} with no PayPal Reference ID.', file=sys.stderr)

    
if __name__ == '__main__':
    main()