from __future__ import annotations import secrets import csv import io import re from datetime import date from decimal import Decimal from flask import Blueprint, Response, abort, flash, redirect, render_template, request, url_for from flask_login import current_user, login_required from sqlalchemy import func from werkzeug.utils import secure_filename from ..extensions import db from ..models.client import Client from ..models.company import Company from ..models.invoice import Invoice from ..models.invoice_line import InvoiceLine from ..services.audit import log_activity from ..services.formatting import normalize_upper, to_title_case from ..services.invoice_numbering import ( compute_default_year, format_number_text, next_invoice_sequence, parse_number_text, sanitize_transmission_progressivo, ) from ..services.pdf_reportlab import render_invoice_pdf from ..services.invoice_xml_store import ( load_original_filename, load_original_invoice_xml, save_original_invoice_xml, ) from ..services.invoice_party_snapshot import ( build_client_snapshot, build_company_snapshot, client_display_name, company_display_name, dumps_snapshot, loads_snapshot, snapshot_to_namespace, ) from ..services.xml_fatturapa import build_fpr12_xml, parse_fpr12 from ..services.prestazioni import remember_prestazioni bp = Blueprint("invoices", __name__, url_prefix="/invoices") def _sanitize_client_piva_cf(*, sigla_nazione: str | None, piva_cf: str | None) -> str | None: if not piva_cf: return None v = normalize_upper(piva_cf) cc = normalize_upper(sigla_nazione or "") if sigla_nazione else "" # For foreign customers, optionally keep IdCodice without a country prefix. # IMPORTANT: only strip the prefix when it matches the selected country code. if cc and cc != "IT" and len(v) >= 4 and v[:2].isalpha() and any(ch.isdigit() for ch in v[2:]): prefix = v[:2] rest = v[2:] if prefix == cc: return rest or None return v or None def _dec(value: str | None, default: str = "0") -> Decimal: try: return Decimal((value or "").replace(",", ".") or default) except Exception: return Decimal(default) def _can_edit(inv: Invoice) -> bool: if getattr(current_user, "role", None) == "ADMIN": return True return getattr(inv, "created_by", None) == getattr(current_user, "id", None) def _parse_bool(value: str | None) -> bool: v = (value or "").strip().lower() return v in {"1", "true", "yes", "y", "si", "sì", "on", "x"} def _safe_decode_csv(data: bytes) -> str: for enc in ("utf-8-sig", "utf-8", "latin-1"): try: return data.decode(enc) except Exception: continue return data.decode("utf-8", errors="replace") def _normalize_header(h: str) -> str: return (h or "").strip().lower().replace(" ", "").replace("_", "") def _detect_csv_dialect(sample: str) -> csv.Dialect: try: return csv.Sniffer().sniff(sample, delimiters=";,\t") except Exception: return csv.excel def _find_invoice_for_csv_row(row: dict[str, str]) -> Invoice | None: # Match priority: # 1) ID # 2) Numero (parseable -> year+seq) raw_id = (row.get("invoice_id") or row.get("id") or "").strip() if raw_id.isdigit(): inv = Invoice.query.get(int(raw_id)) if inv and _can_edit(inv): return inv numero = (row.get("numero") or row.get("number") or row.get("number_text") or "").strip() if numero: seq, yy = parse_number_text(numero) if seq is not None and yy: inv = Invoice.query.filter_by(year=yy, number_seq=seq).first() if inv and _can_edit(inv): return inv inv = Invoice.query.filter_by(number_text=numero).first() if inv and _can_edit(inv): return inv return None def _coerce_client(value: str) -> Client | None: v = (value or "").strip() if not v: return None if v.isdigit(): return Client.query.get(int(v)) # Best-effort match by name fields parts = [p for p in v.split(" ") if p] if len(parts) >= 2: first = parts[0] last = " ".join(parts[1:]) c = Client.query.filter_by(first_name=first, last_name=last).first() if c: return c c = Client.query.filter_by(name=v).first() if c: return c return None def _coerce_company(value: str) -> Company | None: v = (value or "").strip() if not v: return None if v.isdigit(): return Company.query.get(int(v)) return Company.query.filter_by(name=v).first() @bp.get("") @login_required def index(): invoices = ( Invoice.query.order_by(Invoice.year.desc(), Invoice.number_seq.desc(), Invoice.issue_date.desc()) .limit(200) .all() ) # Prefer displaying parties from the stored DB snapshot (fast), then XML snapshot. # This keeps the invoice list stable even if contact records change later. invoice_clients_map: dict[int, str] = {} invoice_companies_map: dict[int, str] = {} for inv in invoices: cs = loads_snapshot(getattr(inv, "client_snapshot_json", None)) if cs: invoice_clients_map[inv.id] = client_display_name(cs) cps = loads_snapshot(getattr(inv, "company_snapshot_json", None)) if cps: invoice_companies_map[inv.id] = company_display_name(cps) if inv.id in invoice_clients_map and inv.id in invoice_companies_map: continue original = load_original_invoice_xml(inv.id) if not original: continue try: parsed = parse_fpr12(original) def _party_name(p) -> str: if getattr(p, "name", None): return p.name or "" fn = (getattr(p, "first_name", None) or "").strip() ln = (getattr(p, "last_name", None) or "").strip() return (fn + (" " + ln if ln else "")).strip() invoice_clients_map[inv.id] = _party_name(parsed.client) invoice_companies_map[inv.id] = _party_name(parsed.company) except Exception: continue client_ids = sorted({inv.client_id for inv in invoices if inv.client_id}) clients_map: dict[int, str] = {} if client_ids: rows = Client.query.filter(Client.id.in_(client_ids)).all() clients_map = {c.id: c.display_name() for c in rows} company_ids = sorted({inv.company_id for inv in invoices if inv.company_id}) companies_map: dict[int, str] = {} if company_ids: rows = Company.query.filter(Company.id.in_(company_ids)).all() companies_map = {c.id: (c.name or "") for c in rows} return render_template( "invoices/index.html", invoices=invoices, clients_map=clients_map, companies_map=companies_map, invoice_clients_map=invoice_clients_map, invoice_companies_map=invoice_companies_map, ) @bp.get("//edit") @login_required def edit(invoice_id: int): inv = Invoice.query.get_or_404(invoice_id) if not _can_edit(inv): abort(403) company = Company.query.get(inv.company_id) if inv.company_id else None client = Client.query.get(inv.client_id) if inv.client_id else None lines = InvoiceLine.query.filter_by(invoice_id=inv.id).order_by(InvoiceLine.line_no.asc()).all() return render_template( "invoices/edit.html", inv=inv, company=company, client=client, lines=lines, ) @bp.post("//update") @login_required def update(invoice_id: int): inv = Invoice.query.get_or_404(invoice_id) if not _can_edit(inv): abort(403) client_id = (request.form.get("client_id") or "").strip() company_id = (request.form.get("company_id") or "").strip() client_name = (request.form.get("client_name") or "").strip() company_name = (request.form.get("company_name") or "").strip() client_piva_cf = (request.form.get("client_piva_cf") or "").strip() client_sigla_nazione = (request.form.get("client_sigla_nazione") or "").strip() client_indirizzo = (request.form.get("client_indirizzo") or "").strip() client_civico = (request.form.get("client_civico") or "").strip() client_cap = (request.form.get("client_cap") or "").strip() client_citta = (request.form.get("client_citta") or "").strip() client_provincia = (request.form.get("client_provincia") or "").strip() client_nazione = (request.form.get("client_nazione") or "").strip() client_email_pec = (request.form.get("client_email_pec") or "").strip() client_cod_univoco = (request.form.get("client_cod_univoco") or "").strip() client_codice_destinatario = (request.form.get("client_codice_destinatario") or "").strip() client_telefono = (request.form.get("client_telefono") or "").strip() client_sito = (request.form.get("client_sito") or "").strip() company_piva_cf = (request.form.get("company_piva_cf") or "").strip() company_sigla_nazione = (request.form.get("company_sigla_nazione") or "").strip() company_indirizzo = (request.form.get("company_indirizzo") or "").strip() company_civico = (request.form.get("company_civico") or "").strip() company_cap = (request.form.get("company_cap") or "").strip() company_citta = (request.form.get("company_citta") or "").strip() company_provincia = (request.form.get("company_provincia") or "").strip() company_nazione = (request.form.get("company_nazione") or "").strip() company_email_pec = (request.form.get("company_email_pec") or "").strip() company_cod_univoco = (request.form.get("company_cod_univoco") or "").strip() company_telefono = (request.form.get("company_telefono") or "").strip() company_sito = (request.form.get("company_sito") or "").strip() client = Client.query.get(int(client_id)) if client_id else None company = Company.query.get(int(company_id)) if company_id else None created_client = False created_company = False if not client and (client_name or client_piva_cf or client_indirizzo or client_email_pec): if not client_name and not client_piva_cf: flash("Per creare un nuovo cliente, inserisci almeno Nome oppure C.F./P.IVA.", "warning") return redirect(url_for("invoices.edit", invoice_id=inv.id)) existing = None if client_piva_cf: key = normalize_upper(client_piva_cf) existing = Client.query.filter_by(piva_cf=key).first() if not existing and client_name: existing = ( Client.query.filter(Client.name.isnot(None)) .filter(func.lower(Client.name) == client_name.lower()) .first() ) # Safety: never overwrite an existing client when editing an invoice. # If a client already exists (same C.F./P.IVA or same name), the user must # explicitly select it from the search results. if existing: flash( "Cliente già presente. Per evitare sovrascritture, selezionalo dalla ricerca (non usare 'nuovo').", "danger", ) return redirect(url_for("invoices.edit", invoice_id=inv.id)) client = Client() if client_name and not (client.first_name or client.last_name): client.name = to_title_case(client_name) or client.name if client_sigla_nazione: client.sigla_nazione = normalize_upper(client_sigla_nazione) if client_piva_cf: client.piva_cf = _sanitize_client_piva_cf(sigla_nazione=client.sigla_nazione, piva_cf=client_piva_cf) if client_indirizzo: client.address = to_title_case(client_indirizzo) if client_civico: client.street_number = to_title_case(client_civico) if client_cap: client.cap = client_cap.strip() or None if client_citta: client.city = to_title_case(client_citta) if client_provincia: client.province = normalize_upper(client_provincia) if client_nazione: client.country = normalize_upper(client_nazione) if client.sigla_nazione and not client.country: client.country = client.sigla_nazione if client.country and not client.sigla_nazione and len(client.country) == 2: client.sigla_nazione = client.country if client_email_pec: client.email_pec = client_email_pec or None if client_cod_univoco: client.codice_univoco = normalize_upper(client_cod_univoco) if client_codice_destinatario: client.codice_destinatario = normalize_upper(client_codice_destinatario) if client_telefono: client.phone = client_telefono or None if client_sito: client.website = client_sito or None if client.id is None: db.session.add(client) db.session.flush() created_client = True if not company and (company_name or company_piva_cf or company_indirizzo or company_email_pec): if not company_name and not company_piva_cf: flash("Per creare una nuova azienda, inserisci almeno Nome oppure C.F./P.IVA.", "warning") return redirect(url_for("invoices.edit", invoice_id=inv.id)) existing = None if company_piva_cf: key = normalize_upper(company_piva_cf) if re.fullmatch(r"\d{11}", key): existing = Company.query.filter_by(vat_number=key).first() else: existing = Company.query.filter_by(tax_code=key).first() if not existing and company_name: existing = Company.query.filter(func.lower(Company.name) == company_name.lower()).first() # Safety: never overwrite an existing company when editing an invoice. if existing: flash( "Azienda già presente. Per evitare sovrascritture, selezionala dalla ricerca (non usare 'nuova').", "danger", ) return redirect(url_for("invoices.edit", invoice_id=inv.id)) company = Company(name="") if company_name: company.name = to_title_case(company_name) or company.name or "(Senza nome)" if company_sigla_nazione: company.sigla_nazione = normalize_upper(company_sigla_nazione) if company_piva_cf: key = normalize_upper(company_piva_cf) if re.fullmatch(r"\d{11}", key): company.vat_country = company.sigla_nazione or "IT" company.vat_number = key company.tax_code = company.tax_code or None else: company.tax_code = key if company_indirizzo: company.address = to_title_case(company_indirizzo) if company_civico: company.street_number = to_title_case(company_civico) if company_cap: company.cap = company_cap.strip() or None if company_citta: company.city = to_title_case(company_citta) if company_provincia: company.province = normalize_upper(company_provincia) if company_nazione: company.country = normalize_upper(company_nazione) if company_email_pec: company.email_pec = company_email_pec or None if company_cod_univoco: company.codice_univoco = normalize_upper(company_cod_univoco) if company_telefono: company.phone = company_telefono or None if company_sito: company.website = company_sito or None company.transmitter_id_country = company.transmitter_id_country or "IT" company.transmitter_id_code = company.transmitter_id_code or (company.tax_code or "") if company.id is None: db.session.add(company) db.session.flush() created_company = True issue_date_raw = (request.form.get("issue_date") or "").strip() issue_date = date.fromisoformat(issue_date_raw) if issue_date_raw else inv.issue_date inv.issue_date = issue_date inv.tipo_documento = (request.form.get("tipo_documento") or inv.tipo_documento or "TD01").strip() or "TD01" inv.currency = (request.form.get("currency") or inv.currency or "EUR").strip() or "EUR" number_text = (request.form.get("number_text") or "").strip() if number_text: inv.number_text = number_text seq, parsed_year = parse_number_text(number_text) if parsed_year: inv.year = parsed_year if seq is not None: inv.number_seq = seq else: # If cleared, regenerate based on (possibly updated) issue_date inv.year = issue_date.year inv.number_seq = next_invoice_sequence(inv.year) inv.number_text = format_number_text(inv.number_seq, inv.year) existing_same_number = ( Invoice.query.filter_by(year=inv.year, number_seq=inv.number_seq) .filter(Invoice.id != inv.id) .first() ) if existing_same_number and number_text and parse_number_text(number_text)[0] is not None: flash("Numero fattura già presente per questo anno. Scegli un altro numero o lascia vuoto per auto.", "warning") return redirect(url_for("invoices.edit", invoice_id=inv.id)) # Ensure uniqueness for auto-numbered cases (avoid constraint errors) while ( Invoice.query.filter_by(year=inv.year, number_seq=inv.number_seq) .filter(Invoice.id != inv.id) .first() ): inv.number_seq += 1 if not number_text: inv.number_text = format_number_text(inv.number_seq, inv.year) bollo_virtuale = (request.form.get("bollo_virtuale") or "") == "on" bollo_amount = _dec(request.form.get("bollo_amount"), "0") if bollo_virtuale else None inv.bollo_virtuale = bollo_virtuale inv.bollo_amount = bollo_amount inv.causale = (request.form.get("causale") or "").strip() or None inv.client_id = client.id if client else None inv.company_id = company.id if company else None line_desc = request.form.getlist("line_description") line_qty = request.form.getlist("line_qty") line_price = request.form.getlist("line_unit_price") line_vat = request.form.getlist("line_vat_rate") line_natura = request.form.getlist("line_natura") new_lines: list[InvoiceLine] = [] totale = Decimal("0") for idx, desc in enumerate(line_desc): desc = (desc or "").strip() if not desc: continue qty = _dec(line_qty[idx] if idx < len(line_qty) else "1", "1") unit_price = _dec(line_price[idx] if idx < len(line_price) else "0", "0") line_total = (qty * unit_price).quantize(Decimal("0.01")) vat_rate = _dec(line_vat[idx] if idx < len(line_vat) else "0", "0") natura = (line_natura[idx] if idx < len(line_natura) else "") or None totale += line_total new_lines.append( InvoiceLine( invoice_id=inv.id, line_no=len(new_lines) + 1, description=desc, qty=qty, unit_price=unit_price, line_total=line_total, vat_rate=vat_rate, natura=(natura or "").strip() or None, sort_order=len(new_lines), ) ) if not new_lines: flash("Inserire almeno una prestazione", "warning") return redirect(url_for("invoices.edit", invoice_id=inv.id)) importo_totale_documento = request.form.get("importo_totale_documento") if importo_totale_documento: itdoc = _dec(importo_totale_documento, str(totale)) else: # ImportoTotaleDocumento: senza bollo (bollo gestito separatamente) itdoc = totale inv.totale_prestazioni = totale inv.importo_totale_documento = itdoc # Save/refresh prestazioni library remember_prestazioni([ln.description for ln in new_lines]) # Replace lines InvoiceLine.query.filter_by(invoice_id=inv.id).delete() for ln in new_lines: db.session.add(ln) db.session.add(inv) db.session.commit() # Snapshot the invoice to XML on every save so later edits to Client/Company # records cannot change already-issued invoices. try: codice_destinatario = (client.codice_destinatario or "0000000") if client else "0000000" xml_bytes = build_fpr12_xml( company=company, client=client, invoice=inv, lines=new_lines, progressivo_invio=inv.transmission_progressivo or secrets.token_hex(4)[:10].upper(), codice_destinatario=codice_destinatario, ) save_original_invoice_xml(invoice_id=inv.id, xml_bytes=xml_bytes, original_filename=None) except Exception: pass # Always persist frozen parties snapshot to DB. try: inv.client_snapshot_json = dumps_snapshot(build_client_snapshot(client)) inv.company_snapshot_json = dumps_snapshot(build_company_snapshot(company)) db.session.add(inv) db.session.commit() except Exception: db.session.rollback() if created_client: flash(f"Creato nuovo cliente: {client.display_name()}", "info") if created_company: flash(f"Creata nuova azienda: {company.name}", "info") flash("Fattura aggiornata", "success") log_activity(None, action="invoice_update", entity_type="invoice", entity_id=inv.id) return redirect(url_for("invoices.index")) @bp.post("//delete") @login_required def delete(invoice_id: int): inv = Invoice.query.get_or_404(invoice_id) if not _can_edit(inv): abort(403) InvoiceLine.query.filter_by(invoice_id=inv.id).delete() db.session.delete(inv) db.session.commit() flash("Fattura eliminata", "success") log_activity(None, action="invoice_delete", entity_type="invoice", entity_id=invoice_id) return redirect(url_for("invoices.index")) @bp.post("/bulk-delete") @login_required def bulk_delete(): raw = (request.form.get("invoice_ids") or "").strip() ids: list[int] = [] # Support both comma-separated and repeated fields. if raw: parts = re.split(r"[^0-9]+", raw) ids.extend([int(p) for p in parts if p.isdigit()]) for v in request.form.getlist("invoice_ids"): vv = (v or "").strip() if vv.isdigit(): ids.append(int(vv)) ids = sorted({i for i in ids if i > 0}) if not ids: flash("Nessuna fattura selezionata", "warning") return redirect(url_for("invoices.index")) invoices = Invoice.query.filter(Invoice.id.in_(ids)).all() if len(invoices) != len(ids): abort(404) for inv in invoices: if not _can_edit(inv): abort(403) InvoiceLine.query.filter(InvoiceLine.invoice_id.in_(ids)).delete(synchronize_session=False) Invoice.query.filter(Invoice.id.in_(ids)).delete(synchronize_session=False) db.session.commit() flash(f"Eliminate {len(ids)} fatture", "success") for invoice_id in ids: log_activity(None, action="invoice_delete", entity_type="invoice", entity_id=invoice_id) return redirect(url_for("invoices.index")) @bp.post("/create") @login_required def create(): client_id = (request.form.get("client_id") or "").strip() company_id = (request.form.get("company_id") or "").strip() client_name = (request.form.get("client_name") or "").strip() company_name = (request.form.get("company_name") or "").strip() # Optional name parts (same semantics as modalClient/modalCompany) client_nome = (request.form.get("client_nome") or "").strip() client_cognome = (request.form.get("client_cognome") or "").strip() company_nome = (request.form.get("company_nome") or "").strip() company_cognome = (request.form.get("company_cognome") or "").strip() client_full_name = (client_nome + (" " + client_cognome if client_cognome else "")).strip() company_full_name = (company_nome + (" " + company_cognome if company_cognome else "")).strip() if not client_name and client_full_name: client_name = client_full_name if not company_name and company_full_name: company_name = company_full_name # Optional extra fields for quick creation client_piva_cf = (request.form.get("client_piva_cf") or "").strip() client_sigla_nazione = (request.form.get("client_sigla_nazione") or "").strip() client_indirizzo = (request.form.get("client_indirizzo") or "").strip() client_civico = (request.form.get("client_civico") or "").strip() client_cap = (request.form.get("client_cap") or "").strip() client_citta = (request.form.get("client_citta") or "").strip() client_provincia = (request.form.get("client_provincia") or "").strip() client_nazione = (request.form.get("client_nazione") or "").strip() client_email_pec = (request.form.get("client_email_pec") or "").strip() client_cod_univoco = (request.form.get("client_cod_univoco") or "").strip() client_codice_destinatario = (request.form.get("client_codice_destinatario") or "").strip() client_telefono = (request.form.get("client_telefono") or "").strip() client_sito = (request.form.get("client_sito") or "").strip() company_piva_cf = (request.form.get("company_piva_cf") or "").strip() company_sigla_nazione = (request.form.get("company_sigla_nazione") or "").strip() company_indirizzo = (request.form.get("company_indirizzo") or "").strip() company_civico = (request.form.get("company_civico") or "").strip() company_cap = (request.form.get("company_cap") or "").strip() company_citta = (request.form.get("company_citta") or "").strip() company_provincia = (request.form.get("company_provincia") or "").strip() company_nazione = (request.form.get("company_nazione") or "").strip() company_email_pec = (request.form.get("company_email_pec") or "").strip() company_cod_univoco = (request.form.get("company_cod_univoco") or "").strip() company_codice_destinatario = (request.form.get("company_codice_destinatario") or "").strip() company_telefono = (request.form.get("company_telefono") or "").strip() company_sito = (request.form.get("company_sito") or "").strip() client = Client.query.get(int(client_id)) if client_id else None company = Company.query.get(int(company_id)) if company_id else None created_client = False created_company = False # If user didn't select an existing record, allow quick creation with extra fields. if not client and (client_name or client_piva_cf or client_indirizzo or client_email_pec): if not client_name and not client_piva_cf: flash("Per creare un nuovo cliente, inserisci almeno Nome oppure C.F./P.IVA.", "warning") return redirect(url_for("invoices.index")) existing = None if client_piva_cf: key = normalize_upper(client_piva_cf) existing = Client.query.filter_by(piva_cf=key).first() if not existing and client_name: existing = ( Client.query.filter(Client.name.isnot(None)) .filter(func.lower(Client.name) == client_name.lower()) .first() ) if existing: flash( "Cliente già presente. Per evitare sovrascritture, selezionalo dalla ricerca (non usare 'nuovo').", "danger", ) return redirect(url_for("invoices.index")) client = Client() # Name handling aligned with modalClient: # - If cognome present -> first_name/last_name, and clear .name # - Else -> use .name and clear first/last if client_cognome: client.first_name = to_title_case(client_nome) or None client.last_name = to_title_case(client_cognome) or None client.name = None elif client_nome or client_name: client.name = to_title_case(client_nome or client_name) or client.name client.first_name = None client.last_name = None if client_sigla_nazione: client.sigla_nazione = normalize_upper(client_sigla_nazione) if client_piva_cf: client.piva_cf = normalize_upper(client_piva_cf) if client_indirizzo: client.address = to_title_case(client_indirizzo) if client_civico: client.street_number = to_title_case(client_civico) if client_cap: client.cap = client_cap.strip() or None if client_citta: client.city = to_title_case(client_citta) if client_provincia: client.province = normalize_upper(client_provincia) if client_nazione: client.country = normalize_upper(client_nazione) if client_email_pec: client.email_pec = client_email_pec or None if client_cod_univoco: client.codice_univoco = normalize_upper(client_cod_univoco) if client_codice_destinatario: client.codice_destinatario = normalize_upper(client_codice_destinatario) if client_telefono: client.phone = client_telefono or None if client_sito: client.website = client_sito or None if client.id is None: db.session.add(client) db.session.flush() created_client = True if not company and (company_name or company_piva_cf or company_indirizzo or company_email_pec): if not company_name and not company_piva_cf: flash("Per creare una nuova azienda, inserisci almeno Nome oppure C.F./P.IVA.", "warning") return redirect(url_for("invoices.index")) existing = None if company_piva_cf: key = normalize_upper(company_piva_cf) if re.fullmatch(r"\d{11}", key): existing = Company.query.filter_by(vat_number=key).first() else: existing = Company.query.filter_by(tax_code=key).first() if not existing and company_name: existing = Company.query.filter(func.lower(Company.name) == company_name.lower()).first() if existing: flash( "Azienda già presente. Per evitare sovrascritture, selezionala dalla ricerca (non usare 'nuova').", "danger", ) return redirect(url_for("invoices.index")) company = Company(name="") if company_name: company.name = to_title_case(company_name) or company.name or "(Senza nome)" if company_sigla_nazione: company.sigla_nazione = normalize_upper(company_sigla_nazione) if company_piva_cf: key = normalize_upper(company_piva_cf) if re.fullmatch(r"\d{11}", key): company.vat_country = company.sigla_nazione or "IT" company.vat_number = key company.tax_code = company.tax_code or None else: company.tax_code = key if company_indirizzo: company.address = to_title_case(company_indirizzo) if company_civico: company.street_number = to_title_case(company_civico) if company_cap: company.cap = company_cap.strip() or None if company_citta: company.city = to_title_case(company_citta) if company_provincia: company.province = normalize_upper(company_provincia) if company_nazione: company.country = normalize_upper(company_nazione) if company_email_pec: company.email_pec = company_email_pec or None if company_cod_univoco: company.codice_univoco = normalize_upper(company_cod_univoco) if company_codice_destinatario: company.codice_destinatario = normalize_upper(company_codice_destinatario) if company_telefono: company.phone = company_telefono or None if company_sito: company.website = company_sito or None company.transmitter_id_country = company.transmitter_id_country or "IT" company.transmitter_id_code = company.transmitter_id_code or (company.tax_code or "") if company.id is None: db.session.add(company) db.session.flush() created_company = True issue_date_raw = (request.form.get("issue_date") or "").strip() issue_date = date.fromisoformat(issue_date_raw) if issue_date_raw else date.today() year = issue_date.year manual_number = (request.form.get("number_text") or "").strip() if manual_number: seq, parsed_year = parse_number_text(manual_number) if parsed_year: year = parsed_year if seq is None: flash("Numero fattura manuale non valido. Usa formato 07/2026 oppure lascia vuoto per auto.", "warning") return redirect(url_for("invoices.index")) # Reject duplicates for manual numbering if Invoice.query.filter_by(year=year, number_seq=seq).first(): flash("Numero fattura già presente per questo anno. Scegli un altro numero o lascia vuoto per auto.", "warning") return redirect(url_for("invoices.index")) number_seq = seq number_text = manual_number else: number_seq = next_invoice_sequence(year) number_text = format_number_text(number_seq, year) # Ensure uniqueness while Invoice.query.filter_by(year=year, number_seq=number_seq).first(): number_seq += 1 if not manual_number: number_text = format_number_text(number_seq, year) bollo_virtuale = (request.form.get("bollo_virtuale") or "") == "on" bollo_amount = _dec(request.form.get("bollo_amount"), "0") if bollo_virtuale else None line_desc = request.form.getlist("line_description") line_qty = request.form.getlist("line_qty") line_price = request.form.getlist("line_unit_price") line_vat = request.form.getlist("line_vat_rate") line_natura = request.form.getlist("line_natura") lines: list[InvoiceLine] = [] totale = Decimal("0") for idx, desc in enumerate(line_desc): desc = (desc or "").strip() if not desc: continue qty = _dec(line_qty[idx] if idx < len(line_qty) else "1", "1") unit_price = _dec(line_price[idx] if idx < len(line_price) else "0", "0") line_total = (qty * unit_price).quantize(Decimal("0.01")) vat_rate = _dec(line_vat[idx] if idx < len(line_vat) else "0", "0") natura = (line_natura[idx] if idx < len(line_natura) else "") or None totale += line_total lines.append( InvoiceLine( line_no=len(lines) + 1, description=desc, qty=qty, unit_price=unit_price, line_total=line_total, vat_rate=vat_rate, natura=(natura or "").strip() or None, sort_order=len(lines), ) ) if not lines: flash("Inserire almeno una prestazione", "warning") return redirect(url_for("invoices.index")) importo_totale_documento = request.form.get("importo_totale_documento") if importo_totale_documento: itdoc = _dec(importo_totale_documento, str(totale)) else: # ImportoTotaleDocumento: senza bollo (bollo gestito separatamente) itdoc = totale inv = Invoice( year=year, number_seq=number_seq, number_text=number_text, tipo_documento=(request.form.get("tipo_documento") or "TD01").strip() or "TD01", currency=(request.form.get("currency") or "EUR").strip() or "EUR", issue_date=issue_date, bollo_virtuale=bollo_virtuale, bollo_amount=bollo_amount, totale_prestazioni=totale, importo_totale_documento=itdoc, causale=(request.form.get("causale") or "").strip() or None, transmission_progressivo=sanitize_transmission_progressivo(secrets.token_urlsafe(6)[:10]), client_id=client.id if client else None, company_id=company.id if company else None, created_by=getattr(current_user, "id", None), ) db.session.add(inv) db.session.flush() remember_prestazioni([ln.description for ln in lines]) for ln in lines: ln.invoice_id = inv.id db.session.add(ln) db.session.commit() # Snapshot the invoice to XML on creation so later edits to Client/Company # records cannot change an already-issued invoice. try: codice_destinatario = (client.codice_destinatario or "0000000") if client else "0000000" xml_snapshot = build_fpr12_xml( company=company, client=client, invoice=inv, lines=lines, progressivo_invio=inv.transmission_progressivo or secrets.token_hex(4)[:10].upper(), codice_destinatario=codice_destinatario, ) save_original_invoice_xml(invoice_id=inv.id, xml_bytes=xml_snapshot, original_filename=None) except Exception: pass # Always persist frozen parties snapshot to DB. try: inv.client_snapshot_json = dumps_snapshot(build_client_snapshot(client)) inv.company_snapshot_json = dumps_snapshot(build_company_snapshot(company)) db.session.add(inv) db.session.commit() except Exception: db.session.rollback() if created_client: flash(f"Creato nuovo cliente: {client.display_name()}", "info") if created_company: flash(f"Creata nuova azienda: {company.name}", "info") flash("Fattura creata", "success") if itdoc != totale: flash("ImportoTotaleDocumento diverso dal totale prestazioni: salvato comunque", "warning") log_activity(None, action="invoice_create", entity_type="invoice", entity_id=inv.id) return redirect(url_for("invoices.index")) @bp.post("/import") @login_required def import_xml(): up = request.files.get("xml_file") if not up: flash("Seleziona un file XML", "warning") return redirect(url_for("invoices.index")) filename = secure_filename(up.filename or "fattura.xml") xml_bytes = up.read() parsed = parse_fpr12(xml_bytes) def _find_existing_company(party) -> Company | None: if getattr(party, "vat_number", None): existing = Company.query.filter_by(vat_number=party.vat_number).first() if existing: return existing if getattr(party, "tax_code", None): existing = Company.query.filter_by(tax_code=party.tax_code).first() if existing: return existing return None # Heuristic: # - Outgoing invoice: our company is CedentePrestatore # - Incoming invoice: our company is CessionarioCommittente existing_company_as_cedente = _find_existing_company(parsed.company) existing_company_as_cessionario = _find_existing_company(parsed.client) if existing_company_as_cedente: company_party = parsed.company client_party = parsed.client is_outgoing = True company_existing = existing_company_as_cedente elif existing_company_as_cessionario: # Incoming invoice: swap roles company_party = parsed.client client_party = parsed.company is_outgoing = False company_existing = existing_company_as_cessionario else: company_party = parsed.company client_party = parsed.client is_outgoing = True company_existing = None # Company upsert by VAT number when possible company = company_existing if not company and getattr(company_party, "vat_number", None): company = Company.query.filter_by(vat_number=company_party.vat_number).first() if not company: company = Company(name=company_party.name or "(Azienda)") company.name = company_party.name or company.name company.vat_country = company_party.vat_country or company.vat_country company.vat_number = company_party.vat_number or company.vat_number company.tax_code = company_party.tax_code or company.tax_code company.regime_fiscale = company_party.regime_fiscale or company.regime_fiscale company.address = company_party.address or company.address company.cap = company_party.cap or company.cap company.city = company_party.city or company.city company.province = company_party.province or company.province company.country = company_party.country or company.country company.transmitter_id_country = company.transmitter_id_country or "IT" company.transmitter_id_code = company.transmitter_id_code or (company.tax_code or "") db.session.add(company) db.session.flush() # Client upsert by tax_code/VAT when possible client = None client_key = client_party.tax_code or client_party.vat_number if client_key: client = Client.query.filter_by(piva_cf=client_key).first() if not client: client = Client() client.first_name = client_party.first_name or client.first_name client.last_name = client_party.last_name or client.last_name if not client.first_name and not client.last_name: client.name = client_party.name or client.name client.piva_cf = client_key or client.piva_cf client.address = client_party.address or client.address client.street_number = client_party.street_number or client.street_number client.cap = client_party.cap or client.cap client.city = client_party.city or client.city client.province = client_party.province or client.province client.country = client_party.country or client.country if is_outgoing: client.codice_destinatario = client_party.codice_destinatario or client.codice_destinatario db.session.add(client) db.session.flush() seq, parsed_year = parse_number_text(parsed.invoice.number_text) inv_year = parsed_year or parsed.invoice.issue_date.year or compute_default_year() inv_seq = seq or next_invoice_sequence(inv_year) # Ensure uniqueness while Invoice.query.filter_by(year=inv_year, number_seq=inv_seq).first(): inv_seq += 1 inv = Invoice( year=inv_year, number_seq=inv_seq, number_text=parsed.invoice.number_text or format_number_text(inv_seq, inv_year), tipo_documento=parsed.invoice.tipo_documento, currency=parsed.invoice.currency, issue_date=parsed.invoice.issue_date, bollo_virtuale=parsed.invoice.bollo_virtuale, bollo_amount=parsed.invoice.bollo_amount, totale_prestazioni=sum((ln.line_total for ln in parsed.lines), Decimal("0")), importo_totale_documento=parsed.invoice.importo_totale_documento or sum((ln.line_total for ln in parsed.lines), Decimal("0")), causale=parsed.invoice.causale, transmission_progressivo=secrets.token_urlsafe(6)[:5], client_id=client.id, company_id=company.id, created_by=getattr(current_user, "id", None), ) db.session.add(inv) db.session.flush() remember_prestazioni([ln.description for ln in parsed.lines]) for ln in parsed.lines: db.session.add( InvoiceLine( invoice_id=inv.id, line_no=ln.line_no, description=ln.description, qty=ln.qty, unit_price=ln.unit_price, line_total=ln.line_total, vat_rate=ln.vat_rate, natura=ln.natura, sort_order=ln.line_no, ) ) db.session.commit() # Keep the original (possibly signed) XML for exact re-export. try: save_original_invoice_xml(invoice_id=inv.id, xml_bytes=xml_bytes, original_filename=filename) except Exception: flash( "Import OK, ma non riesco a salvare l'XML originale per il re-export identico (permessi/cartella tmp)", "warning", ) flash(f"Import OK: {filename}", "success") log_activity(None, action="invoice_import_xml", entity_type="invoice", entity_id=inv.id) return redirect(url_for("invoices.index")) @bp.post("/import-csv") @login_required def import_csv(): up = request.files.get("csv_file") if not up: flash("Seleziona un file CSV", "warning") return redirect(url_for("invoices.index")) selected_fields = set(request.form.getlist("fields")) if not selected_fields: flash("Seleziona almeno un campo da importare", "warning") return redirect(url_for("invoices.index")) raw = up.read() or b"" text = _safe_decode_csv(raw) sample = text[:4096] dialect = _detect_csv_dialect(sample) f = io.StringIO(text) reader = csv.reader(f, dialect) rows = list(reader) if not rows: flash("CSV vuoto", "warning") return redirect(url_for("invoices.index")) # Skip Excel 'sep=;' line if present if rows and rows[0] and (rows[0][0] or "").strip().lower().startswith("sep="): rows = rows[1:] if not rows: flash("CSV vuoto", "warning") return redirect(url_for("invoices.index")) header = rows[0] data_rows = rows[1:] if not header: flash("CSV senza intestazione", "warning") return redirect(url_for("invoices.index")) # Map headers (Italian labels or internal keys) -> internal keys header_map: dict[int, str] = {} LABEL_TO_KEY = { "numero": "numero", "number": "numero", "numbertext": "numero", "data": "data", "cliente": "cliente", "azienda": "azienda", "totaleprestazioni": "totale_prestazioni", "importototaleDocumento".lower().replace(" ", ""): "importo_totale_documento", "importototaledocumento": "importo_totale_documento", "tipodocumento": "tipo_documento", "divisa": "divisa", "bollovirtuale": "bollo_virtuale", "importobollo": "importo_bollo", "causale": "causale", "progressivoinvio": "progressivo_invio", "id": "invoice_id", "invoiceid": "invoice_id", "creatoda": "creato_da", "creatoil": "created_at", } for idx, h in enumerate(header): key = LABEL_TO_KEY.get(_normalize_header(h)) if key: header_map[idx] = key if not header_map: flash("Intestazioni CSV non riconosciute. Usa un CSV esportato dal sistema.", "warning") return redirect(url_for("invoices.index")) updated = 0 skipped = 0 for r in data_rows: if not any((c or "").strip() for c in r): continue row: dict[str, str] = {} for idx, val in enumerate(r): if idx in header_map: row[header_map[idx]] = (val or "").strip() inv = _find_invoice_for_csv_row(row) if not inv: skipped += 1 continue try: if "data" in selected_fields and row.get("data"): inv.issue_date = date.fromisoformat(row["data"].strip()) if "numero" in selected_fields and row.get("numero"): new_number_text = row["numero"].strip() seq, yy = parse_number_text(new_number_text) if seq is None or not yy: raise ValueError("Numero fattura non valido") # Reject duplicates (excluding self) other = ( Invoice.query.filter_by(year=yy, number_seq=seq) .filter(Invoice.id != inv.id) .first() ) if other: raise ValueError("Numero già presente") inv.number_text = new_number_text inv.year = yy inv.number_seq = seq if "tipo_documento" in selected_fields and row.get("tipo_documento"): inv.tipo_documento = row["tipo_documento"].strip() or inv.tipo_documento if "divisa" in selected_fields and row.get("divisa"): inv.currency = row["divisa"].strip() or inv.currency if "causale" in selected_fields: v = (row.get("causale") or "").strip() inv.causale = v or None if "bollo_virtuale" in selected_fields and row.get("bollo_virtuale"): inv.bollo_virtuale = _parse_bool(row.get("bollo_virtuale")) if not inv.bollo_virtuale: inv.bollo_amount = None if "importo_bollo" in selected_fields and row.get("importo_bollo"): inv.bollo_amount = _dec(row.get("importo_bollo"), "0") if "totale_prestazioni" in selected_fields and row.get("totale_prestazioni"): inv.totale_prestazioni = _dec(row.get("totale_prestazioni"), "0") if "importo_totale_documento" in selected_fields and row.get("importo_totale_documento"): inv.importo_totale_documento = _dec(row.get("importo_totale_documento"), "0") if "progressivo_invio" in selected_fields and row.get("progressivo_invio"): inv.transmission_progressivo = sanitize_transmission_progressivo(row.get("progressivo_invio")) if "cliente" in selected_fields and row.get("cliente"): c = _coerce_client(row.get("cliente")) if c: inv.client_id = c.id if "azienda" in selected_fields and row.get("azienda"): c = _coerce_company(row.get("azienda")) if c: inv.company_id = c.id db.session.add(inv) updated += 1 except Exception: skipped += 1 if updated: db.session.commit() flash(f"Import CSV completato: aggiornate {updated} fatture, saltate {skipped} righe", "success") log_activity(None, action="invoice_import_csv", entity_type="invoice", entity_id=None) else: db.session.rollback() flash(f"Nessuna fattura aggiornata. Righe saltate: {skipped}", "warning") return redirect(url_for("invoices.index")) @bp.get("//xml") @login_required def export_xml(invoice_id: int): inv = Invoice.query.get_or_404(invoice_id) if not _can_edit(inv): abort(403) company = Company.query.get(inv.company_id) if inv.company_id else None client = Client.query.get(inv.client_id) if inv.client_id else None company_snap = loads_snapshot(getattr(inv, "company_snapshot_json", None)) client_snap = loads_snapshot(getattr(inv, "client_snapshot_json", None)) original = load_original_invoice_xml(inv.id) if original: vat = (getattr(company, "vat_number", None) if company else None) or (company_snap or {}).get("vat_number") or "IT" fn = load_original_filename(inv.id) or f"{vat}_{inv.number_text.replace('/', '-')}.xml" log_activity(None, action="invoice_export_xml", entity_type="invoice", entity_id=inv.id) return Response( original, mimetype="application/xml", headers={"Content-Disposition": f"attachment; filename={fn}"}, ) if not company or not client: if not (company_snap and client_snap): flash("Fattura incompleta: manca azienda o cliente", "danger") return redirect(url_for("invoices.index")) company = snapshot_to_namespace(company_snap) client = snapshot_to_namespace(client_snap) lines = InvoiceLine.query.filter_by(invoice_id=inv.id).order_by(InvoiceLine.line_no.asc()).all() codice_destinatario = getattr(client, "codice_destinatario", None) or "0000000" xml_bytes = build_fpr12_xml( company=company, client=client, invoice=inv, lines=lines, progressivo_invio=inv.transmission_progressivo or secrets.token_hex(4)[:10].upper(), codice_destinatario=codice_destinatario, ) log_activity(None, action="invoice_export_xml", entity_type="invoice", entity_id=inv.id) vat = getattr(company, "vat_number", None) or "IT" fn = f"{vat}_{inv.number_text.replace('/', '-')}.xml" return Response( xml_bytes, mimetype="application/xml", headers={"Content-Disposition": f"attachment; filename={fn}"}, ) @bp.get("//pdf") @login_required def pdf(invoice_id: int): inv = Invoice.query.get_or_404(invoice_id) if not _can_edit(inv): abort(403) company = Company.query.get(inv.company_id) if inv.company_id else None client = Client.query.get(inv.client_id) if inv.client_id else None company_snap = loads_snapshot(getattr(inv, "company_snapshot_json", None)) client_snap = loads_snapshot(getattr(inv, "client_snapshot_json", None)) original = load_original_invoice_xml(inv.id) if original: parsed = parse_fpr12(original) totale = sum((ln.line_total for ln in parsed.lines), Decimal("0")) try: from types import SimpleNamespace invoice_view = SimpleNamespace( number_text=parsed.invoice.number_text or inv.number_text, issue_date=parsed.invoice.issue_date, causale=parsed.invoice.causale, totale_prestazioni=totale, importo_totale_documento=parsed.invoice.importo_totale_documento or totale, bollo_virtuale=parsed.invoice.bollo_virtuale, bollo_amount=parsed.invoice.bollo_amount, ) except Exception: invoice_view = inv pdf_bytes = render_invoice_pdf( invoice=invoice_view, company=parsed.company, client=parsed.client, lines=parsed.lines, ) else: if not company or not client: if not (company_snap and client_snap): flash("Fattura incompleta: manca azienda o cliente", "danger") return redirect(url_for("invoices.index")) company = snapshot_to_namespace(company_snap) client = snapshot_to_namespace(client_snap) lines = InvoiceLine.query.filter_by(invoice_id=inv.id).order_by(InvoiceLine.line_no.asc()).all() pdf_bytes = render_invoice_pdf(invoice=inv, company=company, client=client, lines=lines) log_activity(None, action="invoice_pdf", entity_type="invoice", entity_id=inv.id) fn = f"Fattura_{inv.number_text.replace('/', '-')}.pdf" return Response( pdf_bytes, mimetype="application/pdf", headers={"Content-Disposition": f"attachment; filename={fn}"}, )