321 lines
11 KiB
Python
321 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import json
|
|
import datetime
|
|
import requests
|
|
|
|
from time import strftime
|
|
from collections import defaultdict
|
|
from python_ifirma.exceptions import PythonIfirmaExceptionFactory
|
|
from python_ifirma.helpers import Helpers
|
|
|
|
|
|
from urllib.parse import urljoin, quote
|
|
|
|
|
|
class VAT:
|
|
VAT_0 = 0.00
|
|
VAT_5 = 0.05
|
|
VAT_8 = 0.08
|
|
VAT_23 = 0.23
|
|
|
|
|
|
class Address:
|
|
def __init__(self, city, zip_code, street=None, country=None):
|
|
super().__init__()
|
|
self.city = city
|
|
self.zip_code = zip_code
|
|
self.street = street
|
|
self.country = country
|
|
|
|
|
|
class Client:
|
|
def __init__(self, name, tax_id, address, email=None, phone_number=None,
|
|
export="no"):
|
|
self.name = name
|
|
self.tax_id = tax_id
|
|
self.address = address
|
|
self.email = email
|
|
self.phone_number = phone_number
|
|
endpoints = {
|
|
"no": "fakturakraj.json",
|
|
"yes": "fakturaeksportuslug.json",
|
|
"eu": "fakturaeksportuslugue.json"
|
|
}
|
|
if export not in endpoints.keys():
|
|
raise ValueError("Unknown export type " + export)
|
|
self.export = export
|
|
self.endpoint = endpoints[export]
|
|
|
|
def get_dict(self):
|
|
d = {
|
|
"Nazwa": self.name,
|
|
"KodPocztowy": self.address.zip_code,
|
|
"Ulica": self.address.street,
|
|
"Miejscowosc": self.address.city,
|
|
"Kraj": self.address.country,
|
|
# "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":
|
|
d.update({"Kraj": self.tax_id[:2]})
|
|
else:
|
|
d.update({"NIP": self.tax_id})
|
|
return d
|
|
|
|
|
|
class Position:
|
|
def __init__(self, vat_rate, quantity, base_price, full_name, unit,
|
|
pkwiu=None, discount_percent=None, gtu=None, export=False):
|
|
self.vat_rate = vat_rate
|
|
self.quantity = quantity
|
|
self.base_price = base_price
|
|
self.full_name = full_name
|
|
self.unit = unit
|
|
self.pkwiu = pkwiu
|
|
self.discount_percent = discount_percent
|
|
self.gtu = gtu
|
|
|
|
def get_dict(self, export=False):
|
|
d = {
|
|
"TypStawkiVat": "PRC",
|
|
"StawkaVat": self.vat_rate,
|
|
"Ilosc": self.quantity,
|
|
"CenaJednostkowa": self.base_price,
|
|
"NazwaPelna": self.full_name,
|
|
"Jednostka": self.unit,
|
|
}
|
|
if self.gtu:
|
|
d["GTU"] = self.gtu
|
|
if self.discount_percent:
|
|
d["Rabat"] = self.discount_percent
|
|
if export:
|
|
d.update({
|
|
"NazwaPelnaObca": self.full_name,
|
|
"JednostkaObca": self.unit == 'szt.' and 'pcs' or self.unit,
|
|
})
|
|
return d
|
|
|
|
|
|
class NewInvoiceParams:
|
|
def __init__(self, client, positions, number=None,
|
|
issue_date=datetime.date.today(), currency="PLN"):
|
|
self.client = client
|
|
self.positions = positions
|
|
self.number = number
|
|
self.issue_date = issue_date
|
|
self.currency = currency
|
|
|
|
def __get_issue_date(self):
|
|
return strftime("%Y-%m-%d", self.issue_date.timetuple())
|
|
|
|
def __get_total_price(self):
|
|
return sum(
|
|
[
|
|
position.quantity * position.base_price * (
|
|
(1 - position.discount_percent / 100) if position.discount_percent else 1
|
|
) for position in self.positions
|
|
]
|
|
)
|
|
|
|
def get_request_data(self):
|
|
series = self.number.split('/')[0][-1]
|
|
number = self.number.split('/')[0][:-1]
|
|
d = {
|
|
"ZaplaconoNaDokumencie": self.__get_total_price(),
|
|
"DataWystawienia": self.__get_issue_date(),
|
|
"DataSprzedazy": self.__get_issue_date(),
|
|
"FormatDatySprzedazy": "DZN",
|
|
"SposobZaplaty": "ELE",
|
|
"RodzajPodpisuOdbiorcy": "BPO",
|
|
"WidocznyNumerGios": False,
|
|
"Numer": number,
|
|
"NazwaSeriiNumeracji": series,
|
|
"Pozycje": [position.get_dict(export=(self.client.export != "no"))
|
|
for position in self.positions],
|
|
"Kontrahent": self.client.get_dict(),
|
|
}
|
|
|
|
if self.client.export == "yes":
|
|
d.update({
|
|
"UslugaSwiadczonaTrybArt28b": True,
|
|
})
|
|
if self.client.export in ("eu", "yes"):
|
|
d.update({
|
|
"DataObowiazkuPodatkowego": self.__get_issue_date(),
|
|
"NazwaUslugi": "services",
|
|
"Waluta": self.currency,
|
|
"Jezyk": "en",
|
|
"KursWalutyWidoczny": False,
|
|
"KursWalutyZDniaPoprzedzajacegoDzienWystawieniaFaktury": 1.00,
|
|
})
|
|
if self.client.export in ("eu", "no"):
|
|
d.update({
|
|
"Zaplacono": self.__get_total_price(),
|
|
})
|
|
if self.client.export in "no":
|
|
d["LiczOd"] = "BRT"
|
|
# from pprint import pprint
|
|
# pprint(d)
|
|
return d
|
|
|
|
|
|
class iFirmaAPI():
|
|
__base_url = "https://www.ifirma.pl/iapi/"
|
|
|
|
__username = 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.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_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)
|
|
}
|
|
return requests.get(url, headers=headers, params=params)
|
|
|
|
def __get_json(self, url_path, key_name, params={}):
|
|
return json.loads(self.__get(url_path, key_name, params).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 not in (0,
|
|
202): # 'Numer faktury' nie jest unikalne
|
|
raise PythonIfirmaExceptionFactory.throw_exception_by_code(response_code)
|
|
|
|
return response_dict
|
|
|
|
def find_partner(self, keyword):
|
|
return self.__get_json(f"kontrahenci/{quote(keyword)}.json",
|
|
self.__invoice_key_name)
|
|
|
|
# Invoice generation
|
|
def __create_invoice_and_return_id(self, invoice, 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"]
|
|
return invoice_id
|
|
else:
|
|
return None
|
|
|
|
def __get_invoice_number(self, invoice_id):
|
|
rj = self.__get_json(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"):
|
|
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_json("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):
|
|
self.__post_cost(cost_json, "oplatatelefon.json")
|
|
|
|
def post_cost_vat(self, cost_json):
|
|
self.__post_cost(cost_json, "kosztdzialalnoscivat.json")
|
|
|
|
def __post_cost(self, cost_json, 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"]
|
|
return invoice_id
|
|
else:
|
|
return None
|
|
|
|
def list_invoices(self, date_from="", date_to=""):
|
|
params = {"dataOd": date_from}
|
|
if date_to:
|
|
params.update(dataDo=date_to)
|
|
|
|
return self.__get_json("faktury.json", self.__invoice_key_name,
|
|
params=params)
|
|
|
|
# Beware: IFIRMA returns errors printed into the pdf...
|
|
def get_invoice_pdf(self, invoice_id, kind=""):
|
|
kinds = defaultdict(lambda: "fakturakraj",
|
|
{"prz_eksport_dost_uslug_nie_ue": "fakturaeksportuslug",
|
|
"prz_eksport_dost_uslug_ue": "fakturaeksportuslugue"})
|
|
return self.__get(f"{kinds[kind]}/{invoice_id}.pdf",
|
|
self.__invoice_key_name)
|