from __future__ import annotations import csv import io import re import secrets import unicodedata from sqlalchemy.exc import IntegrityError from flask import Blueprint, abort, flash, redirect, render_template, request, url_for from flask_login import login_required from ..extensions import db from ..models.client import Client from ..models.company import Company from ..models.country_code import CountryCode from ..models.invoice import Invoice from ..models.invoice_line import InvoiceLine from ..services.audit import log_activity from ..services.authz import admin_required from ..services.custom_fields import replace_custom_fields from ..services.formatting import normalize_upper, to_title_case from ..services.invoice_xml_store import load_original_invoice_xml, save_original_invoice_xml from ..services.invoice_party_snapshot import build_company_snapshot, dumps_snapshot from ..services.xml_fatturapa import build_fpr12_xml bp = Blueprint("companies", __name__, url_prefix="/companies") @bp.get("") @login_required def index(): companies = Company.query.order_by(Company.updated_at.desc()).limit(200).all() country_codes = CountryCode.query.order_by(CountryCode.code.asc()).all() return render_template( "companies/index.html", companies=companies, country_codes=country_codes ) def _clean(value: str, *, upper: bool = False) -> str | None: value = (value or "").strip() if not value: return None return normalize_upper(value) if upper else to_title_case(value) 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: s = (h or "").strip().lower() s = unicodedata.normalize("NFKD", s) s = "".join(ch for ch in s if not unicodedata.combining(ch)) return "".join(ch for ch in s if ch.isalnum()) def _detect_csv_dialect(sample: str) -> csv.Dialect: try: return csv.Sniffer().sniff(sample, delimiters=";,\t") except Exception: return csv.excel def _find_company_for_csv_row(row: dict[str, str]) -> Company | None: raw_id = (row.get("company_id") or row.get("id") or "").strip() if raw_id.isdigit(): return Company.query.get(int(raw_id)) # Safety: do not auto-match by VAT/Tax code during import. # Updating an existing company based on these fields can silently overwrite # unrelated records when CSV data is wrong or reused. return None @bp.post("/save") @login_required def save(): raw_id = (request.form.get("id") or "").strip() company = Company.query.get(int(raw_id)) if raw_id else Company(name="") nome = (request.form.get("nome") or "").strip() cognome = (request.form.get("cognome") or "").strip() full_name = (nome + (" " + cognome if cognome else "")).strip() company.name = to_title_case(full_name) or "(Senza nome)" company.sigla_nazione = _clean(request.form.get("sigla_nazione", ""), upper=True) piva_cf = _clean(request.form.get("piva_cf", ""), upper=True) # Heuristic: if 11 digits -> VAT, otherwise treat as tax code. if piva_cf and re.fullmatch(r"\d{11}", piva_cf): company.vat_country = company.sigla_nazione or "IT" company.vat_number = piva_cf company.tax_code = company.tax_code or None else: company.tax_code = piva_cf company.address = _clean(request.form.get("indirizzo", "")) company.street_number = _clean(request.form.get("civico", "")) company.cap = _clean(request.form.get("cap", ""), upper=False) company.city = _clean(request.form.get("citta", "")) company.province = _clean(request.form.get("provincia", ""), upper=True) company.country = _clean(request.form.get("nazione", ""), upper=True) company.email_pec = (request.form.get("email_pec") or "").strip() or None company.codice_univoco = _clean(request.form.get("cod_univoco", ""), upper=True) company.codice_destinatario = _clean(request.form.get("codice_destinatario", ""), upper=True) company.phone = (request.form.get("telefono") or "").strip() or None company.website = (request.form.get("sito") or "").strip() or None # Transmission fields (optional now; needed for XML) company.transmitter_id_country = company.transmitter_id_country or "IT" company.transmitter_id_code = company.transmitter_id_code or (company.tax_code or "") is_new = company.id is None db.session.add(company) db.session.commit() labels = request.form.getlist("custom_label") values = request.form.getlist("custom_value") replace_custom_fields(owner_type="company", owner_id=company.id, labels=labels, values=values) db.session.commit() log_activity( None, action="company_create" if is_new else "company_update", entity_type="company", entity_id=company.id, ) return redirect(url_for("companies.index")) @bp.post("/delete/") @admin_required def delete(company_id: int): company = Company.query.get_or_404(company_id) def _ensure_snapshot(inv: Invoice, *, client_obj: Client | None) -> bool: if load_original_invoice_xml(inv.id): return True if not client_obj: return False lines = ( InvoiceLine.query.filter_by(invoice_id=inv.id) .order_by(InvoiceLine.sort_order.asc(), InvoiceLine.id.asc()) .all() ) try: codice_destinatario = client_obj.codice_destinatario or "0000000" xml_bytes = build_fpr12_xml( company=company, client=client_obj, 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_bytes, original_filename=None ) return True except Exception: return False linked_invoices = Invoice.query.filter_by(company_id=company.id).all() failed_invoice_ids: list[int] = [] for inv in linked_invoices: try: inv.company_snapshot_json = dumps_snapshot(build_company_snapshot(company)) db.session.add(inv) except Exception: pass client_obj = Client.query.get(inv.client_id) if inv.client_id else None if not _ensure_snapshot(inv, client_obj=client_obj): failed_invoice_ids.append(inv.id) if failed_invoice_ids: db.session.rollback() flash( "Impossibile eliminare l'azienda: alcune fatture collegate non hanno uno snapshot XML e non è stato possibile generarlo.", "danger", ) return redirect(url_for("companies.index")) for inv in linked_invoices: inv.company_id = None db.session.add(inv) db.session.commit() db.session.delete(company) try: db.session.commit() except IntegrityError: db.session.rollback() flash("Impossibile eliminare l'azienda: collegata ad altri dati.", "danger") return redirect(url_for("companies.index")) log_activity(None, action="company_delete", entity_type="company", entity_id=company_id) return redirect(url_for("companies.index")) @bp.post("/bulk-delete") @admin_required def bulk_delete(): raw = (request.form.get("company_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("company_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 azienda selezionata", "warning") return redirect(url_for("companies.index")) companies = Company.query.filter(Company.id.in_(ids)).all() if len(companies) != len(ids): abort(404) by_id = {c.id: c for c in companies} def _ensure_snapshot(inv: Invoice, *, company_obj: Company, client_obj: Client | None) -> bool: if load_original_invoice_xml(inv.id): return True if not client_obj: return False lines = ( InvoiceLine.query.filter_by(invoice_id=inv.id) .order_by(InvoiceLine.sort_order.asc(), InvoiceLine.id.asc()) .all() ) try: codice_destinatario = client_obj.codice_destinatario or "0000000" xml_bytes = build_fpr12_xml( company=company_obj, client=client_obj, 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_bytes, original_filename=None ) return True except Exception: return False linked_invoices = Invoice.query.filter(Invoice.company_id.in_(ids)).all() failed_invoice_ids: list[int] = [] for inv in linked_invoices: company_obj = by_id.get(inv.company_id) if company_obj: try: inv.company_snapshot_json = dumps_snapshot(build_company_snapshot(company_obj)) db.session.add(inv) except Exception: pass client_obj = Client.query.get(inv.client_id) if inv.client_id else None if not company_obj: failed_invoice_ids.append(inv.id) continue if not _ensure_snapshot(inv, company_obj=company_obj, client_obj=client_obj): failed_invoice_ids.append(inv.id) if failed_invoice_ids: db.session.rollback() flash( "Eliminazione interrotta: alcune fatture collegate non hanno uno snapshot XML e non è stato possibile generarlo.", "danger", ) return redirect(url_for("companies.index")) for inv in linked_invoices: inv.company_id = None db.session.add(inv) db.session.commit() Company.query.filter(Company.id.in_(ids)).delete(synchronize_session=False) db.session.commit() flash(f"Eliminate {len(ids)} aziende", "success") for company_id in ids: log_activity( None, action="company_delete", entity_type="company", entity_id=company_id ) return redirect(url_for("companies.index")) @bp.post("/import-csv") @login_required def import_csv(): up = request.files.get("csv_file") if not up: return redirect(url_for("companies.index")) selected_fields = set(request.form.getlist("fields")) if not selected_fields: return redirect(url_for("companies.index")) raw = up.read() or b"" text = _safe_decode_csv(raw) # Excel-style directive: first line like "sep=;". # If present, force delimiter; Sniffer can mis-detect when this line exists. sep_delim: str | None = None try: first_line = (text.splitlines()[0] if text else "").strip() m = re.match(r"^sep\s*=\s*(.)\s*$", first_line, flags=re.IGNORECASE) if m: sep_delim = m.group(1) except Exception: sep_delim = None f = io.StringIO(text) if sep_delim: reader = csv.reader(f, delimiter=sep_delim) else: dialect = _detect_csv_dialect(text[:4096]) reader = csv.reader(f, dialect) rows = list(reader) if not rows: return redirect(url_for("companies.index")) if rows and rows[0] and (rows[0][0] or "").strip().lower().startswith("sep="): rows = rows[1:] if not rows: return redirect(url_for("companies.index")) header = rows[0] data_rows = rows[1:] existing_country_codes = {cc.code for cc in CountryCode.query.all()} def _ensure_country_code(code: str | None) -> None: c = normalize_upper(code or "") if not c or len(c) != 2 or not c.isalpha() or c == "IT": return if c in existing_country_codes: return db.session.add(CountryCode(code=c)) existing_country_codes.add(c) LABEL_TO_KEY = { "id": "company_id", "companyid": "company_id", "nome": "nome", "cf/p.iva": "piva_cf", "cfpiva": "piva_cf", "pivacf": "piva_cf", "piva_cf": "piva_cf", "piva": "vat_number", "vatnumber": "vat_number", "codfiscale": "tax_code", "taxcode": "tax_code", "siglanazione": "sigla_nazione", "indirizzo": "indirizzo", "civico": "civico", "cap": "cap", "città": "citta", "citta": "citta", "provincia": "provincia", "nazione": "nazione", "email/pec": "email_pec", "emailpec": "email_pec", "cod.univoco": "cod_univoco", "codunivoco": "cod_univoco", "cod.destinatario": "codice_destinatario", "coddestinatario": "codice_destinatario", "codicedestinatario": "codice_destinatario", "telefono": "telefono", "sitoweb": "sito", "sito": "sito", } header_map: dict[int, str] = {} for idx, h in enumerate(header): key = LABEL_TO_KEY.get(_normalize_header(h)) if key: header_map[idx] = key created = 0 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() company = _find_company_for_csv_row(row) is_new = company is None if is_new: company = Company(name="") try: if "nome" in selected_fields and row.get("nome"): company.name = to_title_case(row.get("nome") or "") or company.name or "(Senza nome)" if "sigla_nazione" in selected_fields and row.get("sigla_nazione"): company.sigla_nazione = _clean(row.get("sigla_nazione") or "", upper=True) _ensure_country_code(company.sigla_nazione) # Support either piva_cf or explicit vat_number/tax_code if "piva_cf" in selected_fields and row.get("piva_cf"): piva_cf = _clean(row.get("piva_cf") or "", upper=True) # Safety: never let the import overwrite an existing company indirectly. # If VAT/Tax code is already used by another company, skip the row unless we # are explicitly updating that company by ID. if piva_cf: existing = None if re.fullmatch(r"\d{11}", piva_cf): existing = Company.query.filter_by(vat_number=piva_cf).first() else: existing = Company.query.filter_by(tax_code=piva_cf).first() if existing and existing.id != company.id: skipped += 1 continue if piva_cf and re.fullmatch(r"\d{11}", piva_cf): company.vat_country = company.sigla_nazione or "IT" company.vat_number = piva_cf else: company.tax_code = piva_cf if "vat_number" in selected_fields and row.get("vat_number"): vn = _clean(row.get("vat_number") or "", upper=True) if vn: existing = Company.query.filter_by(vat_number=vn).first() if existing and existing.id != company.id: skipped += 1 continue company.vat_country = company.sigla_nazione or "IT" company.vat_number = vn if "tax_code" in selected_fields and row.get("tax_code"): tc = _clean(row.get("tax_code") or "", upper=True) if tc: existing = Company.query.filter_by(tax_code=tc).first() if existing and existing.id != company.id: skipped += 1 continue company.tax_code = tc if "indirizzo" in selected_fields: company.address = _clean(row.get("indirizzo") or "") if "civico" in selected_fields: company.street_number = _clean(row.get("civico") or "") if "cap" in selected_fields: company.cap = _clean(row.get("cap") or "", upper=False) if "citta" in selected_fields: company.city = _clean(row.get("citta") or "") if "provincia" in selected_fields: company.province = _clean(row.get("provincia") or "", upper=True) if "nazione" in selected_fields: company.country = _clean(row.get("nazione") or "", upper=True) # If CSV stores ISO2 in "Nazione", also feed it into the list. if company.country and len(company.country) == 2: _ensure_country_code(company.country) if "email_pec" in selected_fields: company.email_pec = (row.get("email_pec") or "").strip() or None if "cod_univoco" in selected_fields: company.codice_univoco = _clean(row.get("cod_univoco") or "", upper=True) if "codice_destinatario" in selected_fields: company.codice_destinatario = _clean(row.get("codice_destinatario") or "", upper=True) if "telefono" in selected_fields: company.phone = (row.get("telefono") or "").strip() or None if "sito" in selected_fields: company.website = (row.get("sito") or "").strip() 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 "") db.session.add(company) if is_new: created += 1 else: updated += 1 except Exception: skipped += 1 if created or updated: db.session.commit() log_activity(None, action="company_import_csv", entity_type="company", entity_id=None) else: db.session.rollback() return redirect(url_for("companies.index"))