From 2a75177520c0c26eeb0e7a9907634837e335155c Mon Sep 17 00:00:00 2001 From: Trym Auren Date: Wed, 11 Sep 2024 09:58:12 +0200 Subject: [PATCH] Initial commit --- zettle-gebyr/main.py | 135 ++++++++++++++++++++++++++++++++++++++ zettle-gebyr/tripletex.py | 85 ++++++++++++++++++++++++ zettle-gebyr/zettle.py | 67 +++++++++++++++++++ 3 files changed, 287 insertions(+) create mode 100644 zettle-gebyr/main.py create mode 100644 zettle-gebyr/tripletex.py create mode 100644 zettle-gebyr/zettle.py diff --git a/zettle-gebyr/main.py b/zettle-gebyr/main.py new file mode 100644 index 0000000..d84bc83 --- /dev/null +++ b/zettle-gebyr/main.py @@ -0,0 +1,135 @@ +import sys +from datetime import datetime, timedelta, date +from zoneinfo import ZoneInfo +from calendar import monthrange +from tripletex import Tripletex +from tripletex_config import Tripletex_Config +from zettle import Zettle +from zettle_config import Zettle_Config + + +def month(date): + date_format = '%Y-%m-%d' + month = datetime.strftime(datetime.strptime(date,date_format),"%B").lower() + translate = { + "january": "januar", + "february": "februar", + "march": "mars", + "april": "april", + "may": "mai", + "june": "june", + "july": "juli", + "august": "august", + "september": "september", + "october": "oktober", + "november": "november", + "december": "desember" + } + if month in translate.keys(): + return translate[month] + return False + + +def auto_dates(): + date_now = datetime.now(ZoneInfo("Europe/Oslo")).date() # for automatic + year = date_now.year + month = date_now.month + days_in_month = monthrange(year, month)[1] + this_date = (date_now).strftime('%Y-%m-%d')[0:7] + next_date = (date_now + timedelta(days=days_in_month)).strftime('%Y-%m-%d')[0:7] + from_date = f'{this_date}-01' + to_date = f'{next_date}-01' + return from_date, to_date + + +def manual_dates(manual_date): + date_format = '%Y-%m-%d' + date_now = datetime.strptime(manual_date, date_format).date() + year = date_now.year + month = date_now.month + days_in_month = monthrange(year, month)[1] + this_date = (date_now).strftime('%Y-%m-%d')[0:7] + next_date = (date_now + timedelta(days=days_in_month)).strftime('%Y-%m-%d')[0:7] + from_date = f'{this_date}-01' + to_date = f'{next_date}-01' + return from_date, to_date + + +def payload(tripletex_client, fee_dict): + + postings_list = [] + row_num = 1 + + for date in fee_dict: + fee_sum = fee_dict[date] + posting = { + "row": row_num, + "date": date, + "description": "Daglig gebyr Zettle", + "amountGross": -fee_sum, + "amountGrossCurrency": -fee_sum, + "account": { + "id": 15311097 + } + } + contra_posting = { + "row": row_num, + "date": date, + "description": "Daglig gebyr Zettle", + "amountGross": fee_sum, + "amountGrossCurrency": fee_sum, + "account": { + "id": 57458194 + } + } + postings_list.append(posting) + postings_list.append(contra_posting) + desc_date = datetime.now(ZoneInfo("Europe/Oslo")).strftime('%Y-%m-%d') + desc_text = "Gebyr Zettle " + month(postings_list[0]["date"]) + row_num += 1 + + ret = { + "date": desc_date, + "description": desc_text, + "postings": postings_list + } + + return ret + + +def main(): + date1 = 0 + date2 = 0 + if len(sys.argv) == 2: + arg = sys.argv[1] + if arg == "auto": + date1, date2 = auto_dates() + else: + date1, date2 = manual_dates(arg) + else: + print("Usage manual: python main.py 2022-01-01") + print("Usage auto : python main.py auto") + return + + zettle_config = Zettle_Config() + tripletex_config = Tripletex_Config() + + zettle_client = Zettle( + zettle_config.zettle_id, + zettle_config.zettle_secret + ) + fee_dict = zettle_client.get_fees(date1, date2) + + tripletex_client = Tripletex( + tripletex_config.base_url, + tripletex_config.consumer_token, + tripletex_config.employee_token, + tripletex_config.expiration_date + ) + tripletex_client.create_voucher(payload(tripletex_client, fee_dict)) + + print('Finished - check tripletex journal') + + +if __name__ == '__main__': + main() diff --git a/zettle-gebyr/tripletex.py b/zettle-gebyr/tripletex.py new file mode 100644 index 0000000..f253f2b --- /dev/null +++ b/zettle-gebyr/tripletex.py @@ -0,0 +1,85 @@ +import json +from types import SimpleNamespace +import requests +from requests.auth import HTTPBasicAuth + + +class Tripletex: + def __init__(self, base_url, consumer_token, employee_token, expiration_date): + self.base_url = base_url + self.consumer_token = consumer_token + self.employee_token = employee_token + self.expiration_date = expiration_date + self.session_token = self.create_session_token().value.token + self.auth = self.authenticate(self.session_token) + self.headers = {'Content-Type': 'application/json' } + + @classmethod + def from_config(cls, config): + return cls(config.base_url, config.consumer_token, config.employee_token, config.expiration_date) + + def create_session_token(self): + params = {'consumerToken' : self.consumer_token, 'employeeToken' : self.employee_token, 'expirationDate' : self.expiration_date} + r = requests.put(f'{self.base_url}/token/session/:create', params=params) + if (r.status_code == 200): + return self.map(r) + else: + print(r.status_code, r.text, r.reason) + + def authenticate(self, session_token): + return HTTPBasicAuth('0', session_token) + + def who_am_i(self, fields=''): + params = {'fields': fields} + r = requests.get(f'{self.base_url}/token/session/>whoAmI', params=params, auth=self.auth) + return self.map(r) + +# ledger + def create_voucher(self, payload): + r = requests.post(f'{self.base_url}/ledger/voucher', data=json.dumps(payload), auth=self.auth, headers=self.headers) + return self.map(r) + +# account + def get_accounts(self, fields=''): + params = {'fields': fields} + r = requests.get( + f'{self.base_url}/ledger/account', + params=params, + auth=self.auth + ) + return self.map(r) + +# subscribe, see https://developer.tripletex.no/docs/documentation/webhooks/ + def list_available_subscriptions(self, fields=''): + params = {'fields': fields} + r = requests.get( + f'{self.base_url}/event', + params=params, + auth=self.auth + ) + return self.map(r) + + def list_subscriptions(self, fields=''): + params = {'fields': fields} + r = requests.get( + f'{self.base_url}/event/subscription', + params=params, + auth=self.auth + ) + return self.map(r) + + def subscribe_to_voucher_inbox(self, payload): + # params = {'fields' : fields} + r = requests.post( + f'{self.base_url}/event/subscription', + data=payload, + auth=self.auth, + headers=self.headers) + return self.map(r) + +# helpers + @staticmethod + def map(responce): + data = json.dumps(responce.json()) + #print(json.dumps(responce.json(), indent=4, sort_keys=True, ensure_ascii=False)) + return json.loads(data, object_hook=lambda d: SimpleNamespace(**d)) diff --git a/zettle-gebyr/zettle.py b/zettle-gebyr/zettle.py new file mode 100644 index 0000000..3a9fa3c --- /dev/null +++ b/zettle-gebyr/zettle.py @@ -0,0 +1,67 @@ +import requests +import json +from datetime import datetime, timedelta + + +class Zettle: + + def __init__(self, zettle_id, zettle_secret): + self.zettle_id = zettle_id + self.zettle_secret = zettle_secret + self.session_token = self.get_zettle_token() + + def get_zettle_token(self): + + params = { + 'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer', + 'client_id': self.zettle_id, + 'assertion': self.zettle_secret + } + r = requests.post('https://oauth.zettle.com/token', params) + + if r.status_code == 200: + return r.json()['access_token'] + else: + print(r.status_code, r.text, r.reason) + + def get_fees(self,startdate, enddate): + + header = {'Authorization': 'Bearer ' + self.session_token} + url_base = 'https://finance.izettle.com/v2/accounts/liquid/transactions' + + offset = 0 # start get() at offset xx + limit = 1000 # maximum transactions to fetch per get(). Zettle supports up to 1000 + fee_dict = {} # this is where the day:fee will be stored + total_fees = 0 + + while(True): + + url = url_base + '?start=' + str(startdate) + 'T04:00:00-00:00&end=' + str(enddate) + 'T04:00:00-00:00&includeTransactionType=PAYMENT_FEE&limit=' + str(limit) + '&offset=' + str(offset) + + r = requests.get(url, headers=header) + + if r.status_code == 200: + data = r.json() + else: + print(r.status_code, r.text, r.reason) + + timestamp = data[0]['timestamp'][0:10] + + for transaction in data: + + timestamp = transaction['timestamp'][0:10] + hour = int(transaction['timestamp'][11:13]) + if(hour < 4): # transactions registered at zettle between 00.00 and 05.00 (Norway +1 hour offset) + date_format = '%Y-%m-%d' + timestamp = datetime.strftime(datetime.strptime(timestamp, date_format).date() - timedelta(days=1),date_format) + if timestamp in fee_dict.keys(): # adds fee from transaction to sum of fees for day + fee_dict[timestamp] += transaction['amount']/100 + else: # adds new day with fees to datastructure + fee_dict[timestamp] = transaction['amount']/100 + total_fees += transaction['amount'] + + offset += limit + if len(data) < 1000: # no more transactions to fetch, since the last request retrieved below maximum of 1000 + break + + return fee_dict