from __future__ import annotations
from collections import defaultdict
from datetime import date, datetime
from decimal import Decimal
from pathlib import Path
from xml.sax.saxutils import escape
import xml.etree.ElementTree as ET
from flask import current_app
from sqlalchemy import or_
from app.extensions import db
from app.models.catalog import InvoiceLine
from app.models.invoice import Invoice, InvoiceStatus, InvoiceType, Tag, SyncEvent
from app.repositories.invoice_repository import InvoiceRepository
from app.services.company_service import CompanyService
from app.services.ksef_service import KSeFService
from app.services.settings_service import SettingsService
class InvoiceService:
PERIOD_LABELS = {
"month": "miesięczne",
"quarter": "kwartalne",
"year": "roczne",
}
def __init__(self):
self.repo = InvoiceRepository()
def upsert_from_ksef(self, document, company):
invoice = self.repo.get_by_ksef_number(document.ksef_number, company_id=company.id)
created = False
if not invoice:
invoice = Invoice(ksef_number=document.ksef_number, company_id=company.id)
db.session.add(invoice)
created = True
invoice.invoice_number = document.invoice_number
invoice.contractor_name = document.contractor_name
invoice.contractor_nip = document.contractor_nip
invoice.issue_date = document.issue_date
invoice.received_date = document.received_date
invoice.fetched_at = document.fetched_at
invoice.net_amount = document.net_amount
invoice.vat_amount = document.vat_amount
invoice.gross_amount = document.gross_amount
invoice.invoice_type = InvoiceType(document.invoice_type)
invoice.last_synced_at = datetime.utcnow()
invoice.external_metadata = document.metadata
invoice.source_hash = KSeFService.calc_hash(document.xml_content)
invoice.xml_path = self._save_xml(company.id, invoice.ksef_number, document.xml_content)
metadata = dict(document.metadata or {})
payment_details = self.extract_payment_details_from_xml(document.xml_content)
if payment_details.get('payment_form_code') and not metadata.get('payment_form_code'):
metadata['payment_form_code'] = payment_details['payment_form_code']
if payment_details.get('payment_form_label') and not metadata.get('payment_form_label'):
metadata['payment_form_label'] = payment_details['payment_form_label']
if payment_details.get('bank_account') and not metadata.get('seller_bank_account'):
metadata['seller_bank_account'] = payment_details['bank_account']
if payment_details.get('bank_name') and not metadata.get('seller_bank_name'):
metadata['seller_bank_name'] = payment_details['bank_name']
if payment_details.get('payment_due_date') and not metadata.get('payment_due_date'):
metadata['payment_due_date'] = payment_details['payment_due_date']
invoice.external_metadata = metadata
invoice.seller_bank_account = payment_details.get('bank_account') or metadata.get('seller_bank_account') or invoice.seller_bank_account
invoice.contractor_address = (
metadata.get("contractor_address")
or ", ".join([
part for part in [
metadata.get("contractor_street"),
metadata.get("contractor_postal_code"),
metadata.get("contractor_city"),
metadata.get("contractor_country"),
] if part
])
or ""
)
db.session.flush()
existing_lines = invoice.lines.count() if hasattr(invoice.lines, "count") else len(list(invoice.lines))
if existing_lines == 0:
for line in self.extract_lines_from_xml(document.xml_content):
db.session.add(
InvoiceLine(
invoice_id=invoice.id,
description=line["description"],
quantity=line["quantity"],
unit=line["unit"],
unit_net=line["unit_net"],
vat_rate=line["vat_rate"],
net_amount=line["net_amount"],
vat_amount=line["vat_amount"],
gross_amount=line["gross_amount"],
)
)
if not invoice.html_preview:
invoice.html_preview = self.render_invoice_html(invoice)
db.session.flush()
db.session.add(
SyncEvent(
invoice_id=invoice.id,
status="created" if created else "updated",
message="Pobrano z KSeF",
)
)
SettingsService.set_many({"ksef.last_sync_at": datetime.utcnow().isoformat()}, company_id=company.id)
return invoice, created
def _save_xml(self, company_id, ksef_number, xml_content):
base_path = SettingsService.storage_path("app.archive_path", current_app.config["ARCHIVE_PATH"]) / f"company_{company_id}"
base_path.mkdir(parents=True, exist_ok=True)
safe_name = ksef_number.replace("/", "_") + ".xml"
path = Path(base_path) / safe_name
path.write_text(xml_content, encoding="utf-8")
return str(path)
def extract_lines_from_xml(self, xml_content):
def to_decimal(value, default='0'):
raw = str(value or '').strip()
if not raw:
return Decimal(default)
raw = raw.replace(' ', '').replace(',', '.')
try:
return Decimal(raw)
except Exception:
return Decimal(default)
def to_vat_rate(value):
raw = str(value or '').strip().lower()
if not raw:
return Decimal('0')
raw = raw.replace('%', '').replace(' ', '').replace(',', '.')
try:
return Decimal(raw)
except Exception:
return Decimal('0')
def looks_numeric(value):
raw = str(value or '').strip().replace(' ', '').replace(',', '.')
if not raw:
return False
try:
Decimal(raw)
return True
except Exception:
return False
lines = []
try:
root = ET.fromstring(xml_content)
namespace_uri = root.tag.split('}')[0].strip('{') if '}' in root.tag else ''
ns = {'fa': namespace_uri} if namespace_uri else {}
row_path = './/fa:FaWiersz' if ns else './/FaWiersz'
text_path = lambda name: f'fa:{name}' if ns else name
for row in root.findall(row_path, ns):
description = (row.findtext(text_path('P_7'), default='', namespaces=ns) or '').strip()
p_8a = row.findtext(text_path('P_8A'), default='', namespaces=ns)
p_8b = row.findtext(text_path('P_8B'), default='', namespaces=ns)
if looks_numeric(p_8a):
qty_raw = p_8a
unit_raw = p_8b or 'szt.'
else:
unit_raw = p_8a or 'szt.'
qty_raw = p_8b or '1'
unit_net = (
row.findtext(text_path('P_9A'), default='', namespaces=ns)
or row.findtext(text_path('P_9B'), default='', namespaces=ns)
or '0'
)
net = (
row.findtext(text_path('P_11'), default='', namespaces=ns)
or row.findtext(text_path('P_11A'), default='', namespaces=ns)
or '0'
)
vat_rate = row.findtext(text_path('P_12'), default='0', namespaces=ns)
vat = (
row.findtext(text_path('P_12Z'), default='', namespaces=ns)
or row.findtext(text_path('P_11Vat'), default='', namespaces=ns)
or '0'
)
net_dec = to_decimal(net)
vat_dec = to_decimal(vat)
lines.append({
'description': description,
'quantity': to_decimal(qty_raw, '1'),
'unit': (unit_raw or 'szt.').strip(),
'unit_net': to_decimal(unit_net),
'vat_rate': to_vat_rate(vat_rate),
'net_amount': net_dec,
'vat_amount': vat_dec,
'gross_amount': net_dec + vat_dec,
})
except Exception as exc:
current_app.logger.warning(f'KSeF XML line parse error: {exc}')
return lines
def extract_payment_details_from_xml(self, xml_content):
details = {
"payment_form_code": "",
"payment_form_label": "",
"bank_account": "",
"bank_name": "",
"payment_due_date": "",
}
if not xml_content:
return details
try:
root = ET.fromstring(xml_content)
namespace_uri = root.tag.split('}')[0].strip('{') if '}' in root.tag else ''
ns = {'fa': namespace_uri} if namespace_uri else {}
def find_text(path):
return (root.findtext(path, default='', namespaces=ns) or '').strip()
form_code = find_text('.//fa:Platnosc/fa:FormaPlatnosci' if ns else './/Platnosc/FormaPlatnosci')
bank_account = find_text('.//fa:Platnosc/fa:RachunekBankowy/fa:NrRB' if ns else './/Platnosc/RachunekBankowy/NrRB')
bank_name = find_text('.//fa:Platnosc/fa:RachunekBankowy/fa:NazwaBanku' if ns else './/Platnosc/RachunekBankowy/NazwaBanku')
payment_due_date = find_text('.//fa:Platnosc/fa:TerminPlatnosci/fa:Termin' if ns else './/Platnosc/TerminPlatnosci/Termin')
form_labels = {
'1': 'gotówka',
'2': 'karta',
'3': 'bon',
'4': 'czek',
'5': 'weksel',
'6': 'przelew',
'7': 'kompensata',
'8': 'pobranie',
'9': 'akredytywa',
'10': 'polecenie zapłaty',
'11': 'inny',
}
details.update({
'payment_form_code': form_code,
'payment_form_label': form_labels.get(form_code, form_code),
'bank_account': self._normalize_bank_account(bank_account),
'bank_name': bank_name,
'payment_due_date': payment_due_date,
})
except Exception as exc:
current_app.logger.warning(f'KSeF XML payment parse error: {exc}')
return details
@staticmethod
def _first_non_empty(*values, default=""):
for value in values:
if value is None:
continue
if isinstance(value, str):
if value.strip():
return value.strip()
continue
text = str(value).strip()
if text:
return text
return default
@staticmethod
def _normalize_metadata_container(value):
return value if isinstance(value, dict) else {}
def _get_external_metadata(self, invoice):
return self._normalize_metadata_container(getattr(invoice, "external_metadata", {}) or {})
def _get_ksef_metadata(self, invoice):
external_metadata = self._get_external_metadata(invoice)
return self._normalize_metadata_container(external_metadata.get("ksef", {}) or {})
def _resolve_purchase_seller_data(self, invoice):
external_metadata = self._get_external_metadata(invoice)
ksef_meta = self._get_ksef_metadata(invoice)
seller_name = self._first_non_empty(
getattr(invoice, "contractor_name", ""),
external_metadata.get("contractor_name"),
ksef_meta.get("contractor_name"),
)
seller_tax_id = self._first_non_empty(
getattr(invoice, "contractor_nip", ""),
external_metadata.get("contractor_nip"),
ksef_meta.get("contractor_nip"),
)
seller_street = self._first_non_empty(
external_metadata.get("contractor_street"),
ksef_meta.get("contractor_street"),
external_metadata.get("seller_street"),
ksef_meta.get("seller_street"),
)
seller_city = self._first_non_empty(
external_metadata.get("contractor_city"),
ksef_meta.get("contractor_city"),
external_metadata.get("seller_city"),
ksef_meta.get("seller_city"),
)
seller_postal_code = self._first_non_empty(
external_metadata.get("contractor_postal_code"),
ksef_meta.get("contractor_postal_code"),
external_metadata.get("seller_postal_code"),
ksef_meta.get("seller_postal_code"),
)
seller_country = self._first_non_empty(
external_metadata.get("contractor_country"),
ksef_meta.get("contractor_country"),
external_metadata.get("seller_country"),
ksef_meta.get("seller_country"),
)
seller_address = self._first_non_empty(
external_metadata.get("contractor_address"),
ksef_meta.get("contractor_address"),
external_metadata.get("seller_address"),
ksef_meta.get("seller_address"),
getattr(invoice, "contractor_address", ""),
)
if not seller_address:
address_parts = [part for part in [seller_street, seller_postal_code, seller_city, seller_country] if part]
seller_address = ", ".join(address_parts)
return {
"name": seller_name,
"tax_id": seller_tax_id,
"address": seller_address,
}
def update_metadata(self, invoice, form):
invoice.status = InvoiceStatus(form.status.data)
invoice.internal_note = form.internal_note.data
invoice.pinned = bool(form.pinned.data)
invoice.queue_accounting = bool(form.queue_accounting.data)
invoice.is_unread = invoice.status == InvoiceStatus.NEW
requested_tags = [t.strip() for t in (form.tags.data or "").split(",") if t.strip()]
invoice.tags.clear()
for name in requested_tags:
tag = Tag.query.filter_by(name=name).first()
if not tag:
tag = Tag(name=name, color="primary")
db.session.add(tag)
invoice.tags.append(tag)
db.session.commit()
return invoice
def mark_read(self, invoice):
if invoice.is_unread:
invoice.is_unread = False
invoice.status = InvoiceStatus.READ if invoice.status == InvoiceStatus.NEW else invoice.status
invoice.read_at = datetime.utcnow()
db.session.commit()
def render_invoice_html(self, invoice):
def esc(value):
return str(value or "—").replace("&", "&").replace("<", "<").replace(">", ">")
def money(value):
return f"{Decimal(value):,.2f} {invoice.currency or 'PLN'}".replace(",", " ").replace(".", ",")
company_name_raw = invoice.company.name if invoice.company else "Twoja firma"
company_name = esc(company_name_raw)
company_tax_id = esc(getattr(invoice.company, "tax_id", "") if invoice.company else "")
company_address = esc(getattr(invoice.company, "address", "") if invoice.company else "")
customer = getattr(invoice, "customer", None)
customer_name = esc(getattr(customer, "name", invoice.contractor_name))
customer_tax_id = esc(getattr(customer, "tax_id", invoice.contractor_nip))
customer_address = esc(getattr(customer, "address", ""))
customer_email = esc(getattr(customer, "email", ""))
if invoice.invoice_type == InvoiceType.PURCHASE:
purchase_seller = self._resolve_purchase_seller_data(invoice)
seller_name = esc(purchase_seller["name"])
seller_tax_id = esc(purchase_seller["tax_id"])
seller_address = esc(purchase_seller["address"])
buyer_name = company_name
buyer_tax_id = company_tax_id
buyer_address = company_address
buyer_email = ""
else:
seller_name = company_name
seller_tax_id = company_tax_id
seller_address = company_address
buyer_name = customer_name
buyer_tax_id = customer_tax_id
buyer_address = customer_address
buyer_email = customer_email
nfz_meta = (invoice.external_metadata or {}).get("nfz", {})
invoice_kind = "FAKTURA NFZ" if invoice.source == "nfz" else "FAKTURA VAT"
page_title = self._first_non_empty(
getattr(invoice, "invoice_number", ""),
getattr(invoice, "ksef_number", ""),
company_name_raw,
"Faktura",
)
lines = invoice.lines.order_by("id").all() if hasattr(invoice.lines, "order_by") else list(invoice.lines)
if not lines and invoice.xml_path:
try:
xml_content = Path(invoice.xml_path).read_text(encoding="utf-8")
extracted = self.extract_lines_from_xml(xml_content)
lines = [type("TmpLine", (), line) for line in extracted]
except Exception:
lines = []
lines_html = "".join(
[
(
f""
f"| {esc(line.description)} | "
f"{esc(line.quantity)} | "
f"{esc(line.unit)} | "
f"{esc(line.vat_rate)}% | "
f"{money(line.net_amount)} | "
f"{money(line.gross_amount)} | "
f"
"
)
for line in lines
]
) or (
""
"| Brak pozycji na fakturze. | "
"
"
)
nfz_rows = ""
if nfz_meta:
nfz_pairs = [
("Oddział NFZ (IDWew)", nfz_meta.get("recipient_branch_id")),
("Nazwa oddziału", nfz_meta.get("recipient_branch_name")),
("Okres rozliczeniowy od", nfz_meta.get("settlement_from")),
("Okres rozliczeniowy do", nfz_meta.get("settlement_to")),
("Identyfikator świadczeniodawcy", nfz_meta.get("provider_identifier")),
("Kod zakresu / świadczenia", nfz_meta.get("service_code")),
("Numer umowy / aneksu", nfz_meta.get("contract_number")),
("Identyfikator szablonu", nfz_meta.get("template_identifier")),
("Schemat", nfz_meta.get("nfz_schema", "FA(3)")),
]
nfz_rows = "".join(
[
f"| {esc(label)} | {esc(value)} |
"
for label, value in nfz_pairs
]
)
nfz_rows = (
"Dane NFZ
"
""
"| Pole NFZ | Wartość |
|---|
"
f"{nfz_rows}
"
)
buyer_email_html = f"
E-mail: {buyer_email}" if buyer_email not in {"", "—"} else ""
split_payment_html = "Mechanizm podzielonej płatności
" if getattr(invoice, 'split_payment', False) else ""
payment_details = self.resolve_payment_details(invoice)
seller_bank_account = esc(self._resolve_seller_bank_account(invoice))
payment_form_html = f"
Forma płatności: {esc(payment_details.get('payment_form_label'))}" if payment_details.get('payment_form_label') else ''
seller_bank_account_html = f"
Rachunek bankowy: {seller_bank_account}" if seller_bank_account not in {'', '—'} else ''
seller_bank_name_html = f"
Bank: {esc(payment_details.get('bank_name'))}" if payment_details.get('bank_name') else ''
return (
f""
f"
{invoice_kind}
"
f"
"
f"Numer faktury: {esc(invoice.invoice_number)}
Data wystawienia: {esc(invoice.issue_date)}
Waluta: {esc(invoice.currency or 'PLN')} | "
f"Numer KSeF: {esc(invoice.ksef_number)}
Status: {esc(invoice.issued_status)}
Typ źródła: {esc(invoice.source)} | "
f"
"
f"
"
f"Sprzedawca
{seller_name}
NIP: {seller_tax_id}
Adres: {seller_address}{payment_form_html}{seller_bank_account_html}{seller_bank_name_html} | "
f"Nabywca
{buyer_name}
NIP: {buyer_tax_id}
Adres: {buyer_address}{buyer_email_html} | "
f"
"
f"{nfz_rows}"
f"{split_payment_html}"
"
Pozycje faktury
"
"
"
""
""
"| Pozycja | "
"Ilość | "
"JM | "
"VAT | "
"Netto | "
"Brutto | "
"
"
""
f"{lines_html}
"
"
"
f"| Netto | {money(invoice.net_amount)} |
"
f"| VAT | {money(invoice.vat_amount)} |
"
f"| Razem brutto | {money(invoice.gross_amount)} |
"
"
"
f"
{'Dokument zawiera pola wymagane dla rozliczeń NFZ i został przygotowany do wysyłki w schemacie FA(3).' if nfz_meta else 'Dokument wygenerowany przez KSeF Manager.'}
"
"