364 lines
12 KiB
Python
364 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import six
|
|
import json
|
|
import datetime
|
|
from time import strftime
|
|
|
|
from python_ifirma.exceptions import PythonIfirmaExceptionFactory
|
|
from python_ifirma.helpers import Helpers
|
|
import requests
|
|
|
|
|
|
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():
|
|
__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
|
|
|
|
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)
|
|
|
|
@staticmethod
|
|
def __execute_post_request(headers, request_content, url):
|
|
response = requests.post(url, data=request_content, 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)
|
|
raise PythonIfirmaExceptionFactory.throw_exception_by_code(response_code)
|
|
|
|
return response_dict
|
|
|
|
def __get_auth_header(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 __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.__execute_post_request(headers, request_content, url)
|
|
|
|
if response_dict["response"].get("Identyfikator"):
|
|
invoice_id = response_dict["response"]["Identyfikator"]
|
|
return invoice_id
|
|
else:
|
|
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)
|
|
if invoice_id:
|
|
invoice_number = self.__get_invoice_number(invoice_id)
|
|
return invoice_id, invoice_number
|
|
return None, None
|
|
|
|
def post_cost_phone(self, cost_json):
|
|
url = 'https://www.ifirma.pl/iapi/oplatatelefon.json'
|
|
self.__post_cost(cost_json, url)
|
|
|
|
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.__get_auth_header(request_hash_text, self.__user_key_value)
|
|
}
|
|
resp = requests.get(url, headers=headers)
|
|
content = resp.content
|
|
return content
|
|
|
|
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.__execute_post_request(headers, request_content, url)
|
|
|
|
if response_dict["response"].get("Identyfikator"):
|
|
invoice_id = response_dict["response"]["Identyfikator"]
|
|
return invoice_id
|
|
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.__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(
|
|
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)
|
|
}
|
|
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
|
|
|
|
def get_invoice_pdf(self, 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 = f"https://www.ifirma.pl/iapi/fakturakraj/{invoice_id}.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)
|
|
}
|
|
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,
|
|
self.__username,
|
|
self.__invoice_key_name,
|
|
)
|
|
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)
|
|
|
|
content = resp.content
|
|
file_obj = six.BytesIO()
|
|
file_obj.write(content)
|
|
file_obj.seek(0)
|
|
return file_obj
|