Refactor invoice generation

Refactor REST API helpers

python_ifirma/core.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+
 import six
 import json
 import datetime
@@ -9,6 +10,9 @@ from python_ifirma.helpers import Helpers
 import requests
 
 
+from urllib.parse import urljoin, quote
+
+
 class VAT:
     VAT_0 = 0.00
     VAT_5 = 0.05
@@ -50,17 +54,17 @@ class Client:
             "Ulica": self.address.street,
             "Miejscowosc": self.address.city,
             "Kraj": self.address.country,
-            #"Email": self.email,
-            #"Telefon": self.phone_number,
+            # "Email": self.email,
+            # "Telefon": self.phone_number,
             "OsobaFizyczna": False,
         }
         if self.export == "eu":
             d.update({"PrefiksUE": self.tax_id[:2]})
         if self.export == "eu" or self.export == "yes":
             d.update({"NIP": self.tax_id[2:]})
-            #if self.export == "eu":
-            #d.update({"PrefiksUE": self.tax_id[:2]})
-            #elif self.export == "yes":
+            # if self.export == "eu":
+            #    d.update({"PrefiksUE": self.tax_id[:2]})
+            # elif self.export == "yes":
             d.update({"Kraj": self.tax_id[:2]})
         else:
             d.update({"NIP": self.tax_id})
@@ -147,7 +151,7 @@ class NewInvoiceParams:
             d.update({
                 "DataObowiazkuPodatkowego": self.__get_issue_date(),
                 "NazwaUslugi": "services",
-                "Waluta": self.currency, #"PLN",
+                "Waluta": self.currency,
                 "Jezyk": "en",
                 "KursWalutyWidoczny": False,
                 "KursWalutyZDniaPoprzedzajacegoDzienWystawieniaFaktury": 1.00,
@@ -164,116 +168,134 @@ class NewInvoiceParams:
 
 
 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)
 
-    @staticmethod
-    def __execute_post_request(headers, request_content, url):
-        response = requests.post(url, data=request_content, headers=headers)
+    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, params={}):
+        """Send GET request to IFirma API"""
+
+        url = urljoin(self.__base_url, url_path)
+        request_hash_text = f"{url}{self.__username}{key_name}"
+        headers = {
+            "Accept": "application/json",
+            "Authentication": self.__get_auth_header(request_hash_text,
+                                                     key_name)
+        }
+        response = requests.get(url, headers=headers, params=params)
+        return json.loads(response.content.decode('utf-8'))
+
+
+    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)
         real_response_content = response_dict["response"]
         response_code = real_response_content.get("Kod", -1)
 
-        if response_code != 0:
-            # print(response_code)
-            print(real_response_content)
+        if response_code not in (0,
+                                 202): # 'Numer faktury' nie jest unikalne
             raise PythonIfirmaExceptionFactory.throw_exception_by_code(response_code)
 
         return response_dict
 
-    def __create_authentication_header_value(self, request_hash_text, key_value=""):
-        key_value = key_value or self.__invoice_key_value
-        return "IAPIS user={}, hmac-sha1={}".format(
-            self.__username,
-            Helpers.get_hmac_of_text(key_value, request_hash_text)
-        )
+    def find_partner(self, keyword):
+        return self.__get(f"kontrahenci/{quote(keyword)}.json",
+                          self.__invoice_key_name)
 
+    # Invoice generation
     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.__create_authentication_header_value(request_hash_text)
-        }
-
-        response_dict = self.__execute_post_request(headers, request_content, url)
+        response_dict = self.__request("POST", url, self.__invoice_key_name,
+                                           invoice.get_request_data())
 
+        breakpoint()
         if response_dict["response"].get("Identyfikator"):
             invoice_id = response_dict["response"]["Identyfikator"]
             return invoice_id
         else:
             return None
 
+    def __get_invoice_number(self, invoice_id):
+        rj = self.__get(f"fakturakraj/{invoice_id}.json",
+                        self.__user_key_name)
+
+        if "Kod" in rj["response"] and rj["response"]["Kod"] != 200:
+            return None
+        return rj["response"]["PelnyNumer"]
+
     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):
-        url = 'https://www.ifirma.pl/iapi/abonent/miesiacksiegowy.json'
-        request_hash_text = "{}{}{}".format(
-            url,
-            self.__username,
-            self.__user_key_name,
-        )
-        headers = {
-            "Accept": "application/json",
-            "Content-type": "application/json; charset=UTF-8",
-            "Authentication":
-            self.__create_authentication_header_value(request_hash_text, self.__user_key_value)
-        }
-        resp = requests.get(url, headers=headers)
-        content = resp.content
-        return content
+        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.__create_authentication_header_value(request_hash_text)
-        }
-
-        response_dict = self.__execute_post_request(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"]
@@ -281,67 +303,18 @@ class iFirmaAPI():
         else:
             return None
 
-    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.__create_authentication_header_value(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 list_invoices(self, date_from="", date_to=""):
+        params = {"dataOd": date_from}
+        if date_to:
+            params.update(dataDo=date_to)
 
-    def get_invoice(self, invoice_id):
-        url = "https://www.ifirma.pl/iapi/fakturaeksportuslug/{}.json".format(invoice_id)
-        request_hash_text = "{}{}{}".format(
-            url,
-            self.__username,
-            self.__invoice_key_name,
-        )
-        headers = {
-            "Accept": "application/json",
-            "Content-type": "application/json; charset=UTF-8",
-            "Authentication": self.__create_authentication_header_value(request_hash_text)
-        }
-        resp = requests.get(url, headers=headers)
-        rj = json.loads(resp.content.decode('utf-8'))
-        if "Kod" in rj["response"] and rj["response"]["Kod"] != 200:
-            print(rj)
-        else:
-            return rj
+        return self.__get("faktury.json", self.__invoice_key_name,
+                          params=params)
 
     def get_invoice_pdf(self, invoice_id):
-        url = "https://www.ifirma.pl/iapi/fakturakraj/{}.pdf".format(invoice_id)
+        url = f"https://www.ifirma.pl/iapi/fakturakraj/{invoice_id}.pdf"
         return self.__download_pdf(url)
 
-    def __get_invoice_number(self, invoice_id):
-        url = "https://www.ifirma.pl/iapi/fakturakraj/{}.json".format(invoice_id)
-        request_hash_text = "{}{}{}".format(
-            url,
-            self.__username,
-            self.__invoice_key_name,
-        )
-        headers = {
-            "Accept": "application/json",
-            "Content-type": "application/json; charset=UTF-8",
-            "Authentication": self.__create_authentication_header_value(request_hash_text)
-        }
-        resp = requests.get(url, headers=headers)
-        rj = json.loads(resp.content.decode('utf-8'))
-        if "Kod" in rj["response"] and rj["response"]["Kod"] != 200:
-            print(rj)
-        else:
-            return rj["response"]["PelnyNumer"]
-
     def __download_pdf(self, url):
         request_hash_text = "{}{}{}".format(
             url,
@@ -350,8 +323,7 @@ class iFirmaAPI():
         )
         headers = {
             "Accept": "application/pdf",
-            "Content-type": "application/pdf; charset=UTF-8",
-            "Authentication": self.__create_authentication_header_value(request_hash_text)
+            "Authentication": self.__get_auth_header(request_hash_text)
         }
         resp = requests.get(url, headers=headers)
 

setup.py

@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
 
 setup(
     name="python-ifirma",
-    version="0.0.5",
+    version="0.1.0b2",
     packages=find_packages(),
     install_requires=[
         'requests',
@@ -12,4 +12,4 @@ setup(
     author_email='dariusz@aniszewski.eu',
     license='BSD License',
 
-)
+)