From 4ee3a993108f2ba95f69b770f229edbb95723a9a Mon Sep 17 00:00:00 2001 From: Kuba Winnicki Date: Sat, 11 Jan 2025 12:40:48 +0100 Subject: [PATCH] Add accounting month traversal methods. Refactor REST API helpers --- python_ifirma/core.py | 147 ++++++++++++++++++++------------------------------ 1 file changed, 59 insertions(+), 88 deletions(-) diff --git a/python_ifirma/core.py b/python_ifirma/core.py index 36cd28c..482fb35 100644 --- a/python_ifirma/core.py +++ b/python_ifirma/core.py @@ -171,47 +171,55 @@ class iFirmaAPI(): __base_url = "https://www.ifirma.pl/iapi/" __username = None - __invoice_key_name = 'faktura' - __invoice_key_value = None - - __cost_key_name = 'wydatek' - __cost_key_value = None - - __user_key_name = 'abonent' - __user_key_value = None + __invoice_key_name = "faktura" + __cost_key_name = "wydatek" + __user_key_name = "abonent" + keys = { + __invoice_key_name: None, + __cost_key_name : None, + __user_key_name: None + } def __init__(self, _username, _invoice_key_value, _user_key_value=None): self.__username = _username - self.__invoice_key_value = Helpers.unhex_key_value(_invoice_key_value) - self.__user_key_value = Helpers.unhex_key_value(_user_key_value) + self.keys["abonent"] = Helpers.unhex_key_value(_user_key_value) + self.keys["faktura"] = Helpers.unhex_key_value(_invoice_key_value) - def __get_auth_header(self, request_hash_text, key_value=""): - key_value = key_value or self.__invoice_key_value + def __get_auth_header(self, request_hash_text, key_name=""): + key_value = self.keys[key_name or self.__invoice_key_name] return "IAPIS user={}, hmac-sha1={}".format( self.__username, Helpers.get_hmac_of_text(key_value, request_hash_text) ) - def __get(self, url_path, key_name, key_value="", params={}): + def __get(self, url_path, key_name, params={}): """Send GET request to IFirma API""" url = urljoin(self.__base_url, url_path) request_hash_text = f"{url}{self.__username}{key_name}" headers = { - # Content-type not needed in GET requests - # "Content-type": "application/json; charset=UTF-8", "Accept": "application/json", "Authentication": self.__get_auth_header(request_hash_text, - key_value=key_value) + key_name) } response = requests.get(url, headers=headers, params=params) return json.loads(response.content.decode('utf-8')) - @staticmethod - def __post(headers, request_content, url): - """POST request to IFirma API""" - response = requests.post(url, data=request_content, headers=headers) + def __request(self, method, url_path, key_name, request_data={}): + """Send request with given method to IFirma API""" + + url = urljoin(self.__base_url, url_path) + data = json.dumps(request_data, separators=(',', ':')) + request_hash_text = "".join([url, self.__username, key_name, data]) + headers = { + "Accept": "application/json", + "Content-type": "application/json; charset=UTF-8", + "Authentication": self.__get_auth_header(request_hash_text, + key_name) + } + + response = requests.request(method, url, data=data, headers=headers) response_dict = json.loads(response.content.decode("utf-8")) if "response" not in response_dict: raise PythonIfirmaExceptionFactory.throw_exception_by_code(-1) @@ -219,7 +227,6 @@ class iFirmaAPI(): response_code = real_response_content.get("Kod", -1) if response_code != 0: - # print(response_code) print(real_response_content) raise PythonIfirmaExceptionFactory.throw_exception_by_code(response_code) @@ -230,23 +237,8 @@ class iFirmaAPI(): self.__invoice_key_name) def __create_invoice_and_return_id(self, invoice, url): - # from pprint import pprint - # pprint(invoice.get_request_data()) - request_content = json.dumps(invoice.get_request_data(), - separators=(',', ':')) - request_hash_text = "{}{}{}{}".format( - url, - self.__username, - self.__invoice_key_name, - request_content, - ) - headers = { - "Accept": "application/json", - "Content-type": "application/json; charset=UTF-8", - "Authentication": self.__get_auth_header(request_hash_text) - } - - response_dict = self.__post(headers, request_content, url) + response_dict = self.__request("POST", url, self.__invoice_key_name, + invoice.get_request_data()) if response_dict["response"].get("Identyfikator"): invoice_id = response_dict["response"]["Identyfikator"] @@ -255,45 +247,45 @@ class iFirmaAPI(): return None def generate_invoice(self, invoice, endpoint="fakturakraj.json"): - url = "https://www.ifirma.pl/iapi/" + endpoint - invoice_id = self.__create_invoice_and_return_id(invoice, url) + invoice_id = self.__create_invoice_and_return_id(invoice, endpoint) if invoice_id: invoice_number = self.__get_invoice_number(invoice_id) return invoice_id, invoice_number return None, None + def get_accounting_month(self): + """ + Return currently open accounting year and month + """ + + response = self.__get("abonent/miesiacksiegowy.json", + self.__user_key_name) + return (response['response']['RokKsiegowy'], + response['response']["MiesiacKsiegowy"]) + + def set_accounting_month(self, direction): + """ + Set next or previous accounting month + """ + + assert direction in ["NAST", "POPRZ"] + data = {"MiesiacKsiegowy": direction, + "PrzeniesDaneZPoprzedniegoRoku": True} + self.__request("PUT", + "abonent/miesiacksiegowy.json", + self.__user_key_name, + data) + + # Post methods need extensive testing def post_cost_phone(self, cost_json): - url = 'https://www.ifirma.pl/iapi/oplatatelefon.json' - self.__post_cost(cost_json, url) + self.__post_cost(cost_json, "oplatatelefon.json") def post_cost_vat(self, cost_json): - url = 'https://www.ifirma.pl/iapi/kosztdzialalnoscivat.json' - self.__post_cost(cost_json, url) - - def get_accounting_month(self): - response = self.__get("abonent/miesiacksiegowy.json", - self.__user_key_name, - key_value=self.__user_key_value) - - month = response['response']["MiesiacKsiegowy"] - year = response['response']['RokKsiegowy'] - return year, month + self.__post_cost(cost_json, "kosztdzialalnoscivat.json") def __post_cost(self, cost_json, url): - request_content = json.dumps(cost_json, separators=(',', ':')) - request_hash_text = "{}{}{}{}".format( - url, - self.__username, - self.__cost_key_name, - request_content, - ) - headers = { - "Accept": "application/json", - "Content-type": "application/json; charset=UTF-8", - "Authentication": self.__get_auth_header(request_hash_text) - } - - response_dict = self.__post(headers, request_content, url) + response_dict = self.__request("POST", url, self.__cost_key_name, + cost_json) if response_dict["response"].get("Identyfikator"): invoice_id = response_dict["response"]["Identyfikator"] @@ -309,25 +301,6 @@ class iFirmaAPI(): return self.__get("faktury.json", self.__invoice_key_name, params=params) - def get_invoice_list(self, limit=0): - url = "https://www.ifirma.pl/iapi/fakturaeksportuslugue/list.json" - request_hash_text = "{}{}{}".format( - url, - self.__username, - self.__invoice_key_name, - ) - headers = { - "Accept": "application/json", - "Content-type": "application/json; charset=UTF-8", - "Authentication": self.__get_auth_header(request_hash_text) - } - params = {} - if limit: - params['limit'] = limit - resp = requests.get(url, headers=headers, params=params) - rj = json.loads(resp.content.decode('utf-8')) - return rj - def get_invoice(self, invoice_id): url = f"https://www.ifirma.pl/iapi/fakturaeksportuslug/{invoice_id}.json" request_hash_text = "{}{}{}".format( @@ -360,7 +333,6 @@ class iFirmaAPI(): ) headers = { "Accept": "application/json", - "Content-type": "application/json; charset=UTF-8", "Authentication": self.__get_auth_header(request_hash_text) } resp = requests.get(url, headers=headers) @@ -378,7 +350,6 @@ class iFirmaAPI(): ) headers = { "Accept": "application/pdf", - "Content-type": "application/pdf; charset=UTF-8", "Authentication": self.__get_auth_header(request_hash_text) } resp = requests.get(url, headers=headers)