Initial commit

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()

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))

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