From 05375ecec2dd001acac6ad6ec539f4d17d3bf4d9 Mon Sep 17 00:00:00 2001 From: peterhpo Date: Wed, 2 Oct 2024 23:19:05 +0200 Subject: [PATCH] =?UTF-8?q?Refactoret=20koden=20til=20=C3=A5=20bruke=20env?= =?UTF-8?q?ironment=20variabler=20+=20gjorde=20den=20litt=20mer=20robust?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- zettle-gebyr/README.md | 19 ++++- zettle-gebyr/main.py | 158 ++++++++++++++++++-------------------- zettle-gebyr/tripletex.py | 105 ++++++++++--------------- zettle-gebyr/zettle.py | 76 +++++++++--------- 4 files changed, 174 insertions(+), 184 deletions(-) diff --git a/zettle-gebyr/README.md b/zettle-gebyr/README.md index a8eca00..beddd43 100644 --- a/zettle-gebyr/README.md +++ b/zettle-gebyr/README.md @@ -4,4 +4,21 @@ CYB benytter regnskapssystemet Tripletex og betalingsterminaler fra Zettle. Ette Zettle trekker et gebyr på x% fra utbetalingene de gjør, er det nødvendig å føre disse gebyrene daglig for å klare å lukke poster i Tripletex. -2024-09-11: Dette er veldig enkel og lite robust kode. Du vil ikke få beskjed hvis noe ikke fungerer. \ No newline at end of file +2024-09-11: Dette er veldig enkel og lite robust kode. Du vil ikke få beskjed hvis noe ikke fungerer. + +2024-10-02: Nå er koden mere robust, samt refactoret til å bruke en .env fil istedenfor hardkodede verdier i egne config filer + +## Env fil + +Env-filen må inneholde disse feltene: + +``` bash +# Tripletex nøkler/tokens fås fra wikien (eller noen i økonomigruppen som sitter med infoen) +TRIPLETEX_CONSUMER_TOKEN= +TRIPLETEX_EMPLOYEE_TOKEN= +TRIPLETEX_BASE_URL=https://tripletex.no/v2 + +# Zettle nøkkel kan lages på https://my.zettle.com/apps/api-keys hvis du har tilgang +ZETTLE_ID= +ZETTLE_SECRET= +``` diff --git a/zettle-gebyr/main.py b/zettle-gebyr/main.py index d84bc83..0ce4cfb 100644 --- a/zettle-gebyr/main.py +++ b/zettle-gebyr/main.py @@ -1,12 +1,32 @@ +import os import sys -from datetime import datetime, timedelta, date -from zoneinfo import ZoneInfo +from datetime import datetime, timedelta from calendar import monthrange -from tripletex import Tripletex -from tripletex_config import Tripletex_Config +from zoneinfo import ZoneInfo +from dotenv import load_dotenv from zettle import Zettle -from zettle_config import Zettle_Config +from tripletex import Tripletex + +load_dotenv() + +def parse_custom_dates(start_date_str, end_date_str): + start_date = datetime.strptime(start_date_str, "%Y-%m-%d").strftime("%Y-%m-%d") + end_date = datetime.strptime(end_date_str, "%Y-%m-%d").strftime("%Y-%m-%d") + return start_date, end_date + +def auto_dates(): + now = datetime.now() + start_date = datetime(now.year, now.month, 1).strftime("%Y-%m-%d") + last_day_of_month = monthrange(now.year, now.month)[1] + end_date = datetime(now.year, now.month, last_day_of_month).strftime("%Y-%m-%d") + return start_date, end_date +def manual_dates(manual_date): + date = datetime.strptime(manual_date, "%Y-%m-%d") + start_date = datetime(date.year, date.month, 1).strftime("%Y-%m-%d") + last_day_of_month = monthrange(date.year, date.month)[1] + end_date = datetime(date.year, date.month, last_day_of_month).strftime("%Y-%m-%d") + return start_date, end_date def month(date): date_format = '%Y-%m-%d' @@ -17,7 +37,7 @@ def month(date): "march": "mars", "april": "april", "may": "mai", - "june": "june", + "june": "juni", "july": "juli", "august": "august", "september": "september", @@ -25,111 +45,83 @@ def month(date): "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): + return translate.get(month, month) +def payload(fee_dict): postings_list = [] row_num = 1 - for date in fee_dict: - fee_sum = fee_dict[date] + for date, fee_sum in fee_dict.items(): posting = { "row": row_num, "date": date, "description": "Daglig gebyr Zettle", "amountGross": -fee_sum, "amountGrossCurrency": -fee_sum, - "account": { - "id": 15311097 - } - } + "account": {"id": 15311097} + } contra_posting = { "row": row_num, "date": date, "description": "Daglig gebyr Zettle", "amountGross": fee_sum, "amountGrossCurrency": fee_sum, - "account": { - "id": 57458194 - } - } + "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 + description_date = datetime.now(ZoneInfo("Europe/Oslo")).strftime('%Y-%m-%d') + description_text = "Gebyr Zettle " + month(postings_list[0]["date"]) + return { + "date": description_date, + "description": description_text, + "postings": postings_list + } 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) + if len(sys.argv) == 2 and sys.argv[1] == "auto": + start_date, end_date = auto_dates() + elif len(sys.argv) == 2 and sys.argv[1] != "auto": + start_date, end_date = manual_dates(sys.argv[1]) + elif len(sys.argv) == 3: + start_date, end_date = parse_custom_dates(sys.argv[1], sys.argv[2]) else: - print("Usage manual: python main.py 2022-01-01") - print("Usage auto : python main.py auto") - return + print("syntax: auto or yyyy-mm-dd or startdate enddate") + exit() - zettle_config = Zettle_Config() - tripletex_config = Tripletex_Config() + print("Program running...") - zettle_client = Zettle( - zettle_config.zettle_id, - zettle_config.zettle_secret - ) - fee_dict = zettle_client.get_fees(date1, date2) + zettle_client = Zettle(os.getenv("ZETTLE_ID"), os.getenv("ZETTLE_SECRET")) + fee_dict = zettle_client.get_fees(start_date, end_date) + print(f"Retrieved Zettle fees from {start_date} to {end_date}: {fee_dict}") - 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)) + tripletex_base_url = os.getenv("TRIPLETEX_BASE_URL", "https://api.tripletex.io/v2") + tripletex_consumer_token = os.getenv("TRIPLETEX_CONSUMER_TOKEN") + tripletex_employee_token = os.getenv("TRIPLETEX_EMPLOYEE_TOKEN") - print('Finished - check tripletex journal') + if not tripletex_consumer_token or not tripletex_employee_token: + print("Missing Tripletex tokens. Exiting...") + return + + expiration_date = (datetime.today() + timedelta(days=1)).strftime('%Y-%m-%d') + + tripletex_client = Tripletex( + tripletex_base_url, + tripletex_consumer_token, + tripletex_employee_token, + expiration_date # Future date + ) + + if not tripletex_client.session_token: + print("Unable to create Tripletex client. Exiting...") + return + voucher_payload = payload(fee_dict) + response = tripletex_client.create_voucher(voucher_payload) + print(f"Completed: {response}") -if __name__ == '__main__': +if __name__ == "__main__": main() diff --git a/zettle-gebyr/tripletex.py b/zettle-gebyr/tripletex.py index f253f2b..b2173ab 100644 --- a/zettle-gebyr/tripletex.py +++ b/zettle-gebyr/tripletex.py @@ -10,76 +10,53 @@ def __init__(self, base_url, consumer_token, employee_token, expiration_date): 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) + self.session_token = None + self.auth = None + + self.create_session() + self.authenticate(self.session_token) + + def create_session(self): + url = f"{self.base_url}/token/session/:create" + params = { + 'consumerToken': self.consumer_token, + 'employeeToken': self.employee_token, + 'expirationDate': self.expiration_date + } + + response = requests.put(url, params=params) + if response.status_code == 200: + self.session_token = self.map(response).value.token + print("Session token created successfully:", self.session_token) else: - print(r.status_code, r.text, r.reason) - + print("Failed to create session token:", response.status_code, response.text) + def authenticate(self, session_token): - return HTTPBasicAuth('0', session_token) + self.auth = 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) + if not self.session_token: + print("No session token available, cannot create voucher.") + return + + url = f"{self.base_url}/ledger/voucher" + headers = { + 'Authorization': f'Bearer {self.session_token}', + 'Content-Type': 'application/json' + } + + response = requests.post(url, json=payload, auth=self.auth, headers=headers) + if response.status_code == 201: + return response.json() + else: + return { + 'status_code': response.status_code, + 'text': response.text + } # helpers @staticmethod - def map(responce): - data = json.dumps(responce.json()) + def map(response): + data = json.dumps(response.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 index 3a9fa3c..e426576 100644 --- a/zettle-gebyr/zettle.py +++ b/zettle-gebyr/zettle.py @@ -1,67 +1,71 @@ +# zettle.py import requests -import json -from datetime import datetime, timedelta - +from datetime import timedelta, datetime 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() + self.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) + headers = { + 'Content-Type': 'application/x-www-form-urlencoded' + } + data = { + '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', headers=headers, data=data) if r.status_code == 200: - return r.json()['access_token'] + return r.json().get('access_token') else: - print(r.status_code, r.text, r.reason) + print("Failed to get Zettle token:", r.status_code, r.text, r.reason) + return None - def get_fees(self,startdate, enddate): + def get_fees(self, start_date, end_date): + if not self.token: + print("No Zettle token available. Exiting get_fees.") + return {} - header = {'Authorization': 'Bearer ' + self.session_token} + header = {'Authorization': 'Bearer ' + self.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 + offset = 0 + limit = 1000 + fee_dict = {} 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) + while True: + url = (f"{url_base}?start={start_date}T04:00:00-00:00&end={end_date}T04:00:00-00:00" + f"&includeTransactionType=PAYMENT_FEE&limit={limit}&offset={offset}") r = requests.get(url, headers=header) - if r.status_code == 200: - data = r.json() - else: - print(r.status_code, r.text, r.reason) + if r.status_code != 200: + print("Failed to get Zettle fees:", r.status_code, r.text, r.reason) + break - timestamp = data[0]['timestamp'][0:10] + data = r.json() 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) + if hour < 4: 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'] + timestamp = (datetime.strptime(timestamp, date_format).date() - timedelta(days=1)).strftime(date_format) + + amount = transaction['amount'] / 100 + if timestamp in fee_dict: + fee_dict[timestamp] += amount + else: + fee_dict[timestamp] = amount + total_fees += amount offset += limit - if len(data) < 1000: # no more transactions to fetch, since the last request retrieved below maximum of 1000 + if len(data) < 1000: break return fee_dict