ngalaiko/tink.py

## tink.py
#!/usr/bin/python3

import json
import webbrowser
from http.server import HTTPServer, BaseHTTPRequestHandler
from datetime import date, time, datetime, timedelta
import argparse
from urllib.parse import urlencode, urlparse, parse_qs
from urllib import request
from decimal import Decimal


def get_tink_link_url(client_id, redirect_uri):
    params = {
        'client_id': client_id,
        'redirect_uri': redirect_uri,
        'market': 'SE',
        'locale': 'en_US',
        'scope': 'accounts:read,transactions:read',
        'input_provider': 'se-nordea-ob'
    }
    return f'https://link.tink.com/1.0/authorize/?{urlencode(params)}'


def wait_for_code():
    class TinkLinkRedirectServer(HTTPServer):
        def __init__(self, *args, **kwargs):
            HTTPServer.__init__(self, *args, **kwargs)
            self.code = None

    class TinkLinkRedirectHandler(BaseHTTPRequestHandler):
        def do_GET(self):
            o = urlparse(self.path)
            q = parse_qs(o.query)
            self.server.code = q['code'][0]
            self.wfile.write(
                '<html><body onload="window.close()"></body></html>'.encode())

    httpd = TinkLinkRedirectServer(
        ('localhost', 3000), TinkLinkRedirectHandler)
    httpd.handle_request()
    return httpd.code


def exchange_code(code, client_id, client_secret):
    data = urlencode({
        'code': code,
        'client_id': client_id,
        'client_secret': client_secret,
        'grant_type': 'authorization_code'
    }).encode()
    req = request.Request('https://api.tink.com/api/v1/oauth/token', data=data)
    resp = json.loads(request.urlopen(req).read())
    return {
        'access_token': resp['access_token'],
        'expires': datetime.now() + timedelta(0, int(resp['expires_in'])),
        'refresh_token': resp['refresh_token'],
    }


def list_transactions(access_token, begin, end, page_token=None):
    params = {}
    if page_token:
        params['pageToken'] = page_token

    url = f'https://api.tink.com/data/v2/transactions?{urlencode(params)}'
    print(f'GET {url}')
    req = request.Request(url, headers={
        'Authorization': f'Bearer {access_token}'
    })

    def parse_date(raw):
        if len(raw) == 0:
            return None
        [year, month, day] = raw.split('-')
        return date(int(year), int(month), int(day))

    page = json.loads(request.urlopen(req).read())
    transactions = []
    for transaction in page['transactions']:
        dates = transaction['dates']
        booked_date = parse_date(dates['booked'])
        value_date = parse_date(dates['value'])

        if booked_date < begin:
            return transactions

        if booked_date > end:
            continue

        transactions.append({
            'booked': booked_date,
            'value': value_date,
            'status': transaction['status'],
            'reference': transaction['reference'] if 'reference' in transaction else None,
            'description': transaction['descriptions']['original'],
            'currency': transaction['amount']['currencyCode'],
            'amount': Decimal.from_float(float(transaction['amount']['value']['unscaledValue'])) / (10 ** int(transaction['amount']['value']['scale']))
        })

    if not page['nextPageToken']:
        return transactions

    return transactions + list_transactions(access_token, begin, end, page['nextPageToken'])


def main(client_id, client_secret, begin, end):
    try:
        begin = date.fromisoformat(begin)
    except ValueError:
        print("Invalid end date {}".format(end))
        sys.exit(1)

    try:
        end = date.fromisoformat(end)
    except ValueError:
        print("Invalid end date {}".format(end))
        sys.exit(1)

    tink_link_url = get_tink_link_url(
        client_id, 'http://localhost:3000/callback')

    print(f'open: {tink_link_url}')
    webbrowser.open(tink_link_url)

    tink_code = wait_for_code()
    token = exchange_code(code=tink_code, client_id=client_id,
                          client_secret=client_secret)

    transactions = list_transactions(
        access_token=token['access_token'], begin=begin, end=end)

    print(transactions)
    # TODO


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--client_id", default='90323e9502d94aedb2cf47e4d4abe556')
    parser.add_argument("--client_secret",
                        required=True)
    parser.add_argument('--begin', default="2018-01-01")
    parser.add_argument('--end', default=str(date.today()))
    args = parser.parse_args()
    args.begin = args.begin or "1970-01-01"

    main(client_id=args.client_id, client_secret=args.client_secret,
         begin=args.begin, end=args.end)
	#!/usr/bin/python3

	import json
	import webbrowser
	from http.server import HTTPServer, BaseHTTPRequestHandler
	from datetime import date, time, datetime, timedelta
	import argparse
	from urllib.parse import urlencode, urlparse, parse_qs
	from urllib import request
	from decimal import Decimal


	def get_tink_link_url(client_id, redirect_uri):
	params = {
	'client_id': client_id,
	'redirect_uri': redirect_uri,
	'market': 'SE',
	'locale': 'en_US',
	'scope': 'accounts:read,transactions:read',
	'input_provider': 'se-nordea-ob'
	}
	return f'https://link.tink.com/1.0/authorize/?{urlencode(params)}'


	def wait_for_code():
	class TinkLinkRedirectServer(HTTPServer):
	def __init__(self, args, *kwargs):
	HTTPServer.__init__(self, args, *kwargs)
	self.code = None

	class TinkLinkRedirectHandler(BaseHTTPRequestHandler):
	def do_GET(self):
	o = urlparse(self.path)
	q = parse_qs(o.query)
	self.server.code = q['code'][0]
	self.wfile.write(
	'<html><body onload="window.close()"></body></html>'.encode())

	httpd = TinkLinkRedirectServer(
	('localhost', 3000), TinkLinkRedirectHandler)
	httpd.handle_request()
	return httpd.code


	def exchange_code(code, client_id, client_secret):
	data = urlencode({
	'code': code,
	'client_id': client_id,
	'client_secret': client_secret,
	'grant_type': 'authorization_code'
	}).encode()
	req = request.Request('https://api.tink.com/api/v1/oauth/token', data=data)
	resp = json.loads(request.urlopen(req).read())
	return {
	'access_token': resp['access_token'],
	'expires': datetime.now() + timedelta(0, int(resp['expires_in'])),
	'refresh_token': resp['refresh_token'],
	}


	def list_transactions(access_token, begin, end, page_token=None):
	params = {}
	if page_token:
	params['pageToken'] = page_token

	url = f'https://api.tink.com/data/v2/transactions?{urlencode(params)}'
	print(f'GET {url}')
	req = request.Request(url, headers={
	'Authorization': f'Bearer {access_token}'
	})

	def parse_date(raw):
	if len(raw) == 0:
	return None
	[year, month, day] = raw.split('-')
	return date(int(year), int(month), int(day))

	page = json.loads(request.urlopen(req).read())
	transactions = []
	for transaction in page['transactions']:
	dates = transaction['dates']
	booked_date = parse_date(dates['booked'])
	value_date = parse_date(dates['value'])

	if booked_date < begin:
	return transactions

	if booked_date > end:
	continue

	transactions.append({
	'booked': booked_date,
	'value': value_date,
	'status': transaction['status'],
	'reference': transaction['reference'] if 'reference' in transaction else None,
	'description': transaction['descriptions']['original'],
	'currency': transaction['amount']['currencyCode'],
	'amount': Decimal.from_float(float(transaction['amount']['value']['unscaledValue'])) / (10 ** int(transaction['amount']['value']['scale']))
	})

	if not page['nextPageToken']:
	return transactions

	return transactions + list_transactions(access_token, begin, end, page['nextPageToken'])


	def main(client_id, client_secret, begin, end):
	try:
	begin = date.fromisoformat(begin)
	except ValueError:
	print("Invalid end date {}".format(end))
	sys.exit(1)

	try:
	end = date.fromisoformat(end)
	except ValueError:
	print("Invalid end date {}".format(end))
	sys.exit(1)

	tink_link_url = get_tink_link_url(
	client_id, 'http://localhost:3000/callback')

	print(f'open: {tink_link_url}')
	webbrowser.open(tink_link_url)

	tink_code = wait_for_code()
	token = exchange_code(code=tink_code, client_id=client_id,
	client_secret=client_secret)

	transactions = list_transactions(
	access_token=token['access_token'], begin=begin, end=end)

	print(transactions)
	# TODO


	if __name__ == "__main__":
	parser = argparse.ArgumentParser()
	parser.add_argument(
	"--client_id", default='90323e9502d94aedb2cf47e4d4abe556')
	parser.add_argument("--client_secret",
	required=True)
	parser.add_argument('--begin', default="2018-01-01")
	parser.add_argument('--end', default=str(date.today()))
	args = parser.parse_args()
	args.begin = args.begin or "1970-01-01"

	main(client_id=args.client_id, client_secret=args.client_secret,
	begin=args.begin, end=args.end)