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_client_snapshot, dumps_snapshot from ..services.xml_fatturapa import build_fpr12_xml bp = Blueprint("clients", __name__, url_prefix="/clients") @bp.get("") @login_required def index(): clients = Client.query.order_by(Client.updated_at.desc()).limit(200).all() country_codes = CountryCode.query.order_by(CountryCode.code.asc()).all() return render_template( "clients/index.html", clients=clients, 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: # Normalize CSV header labels so variants like "C.F./P.IVA", "Email/PEC", # "N° Civico" all map to the same internal keys. 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_client_for_csv_row(row: dict[str, str]) -> Client | None: raw_id = (row.get("client_id") or row.get("id") or "").strip() if raw_id.isdigit(): return Client.query.get(int(raw_id)) return 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 _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 store IdCodice without a country prefix. # IMPORTANT: only strip the prefix when it matches the selected country code. # Never strip arbitrary alphabetic prefixes (e.g. "OO999..." must stay as-is). 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 @bp.post("/save") @login_required def save(): raw_id = (request.form.get("id") or "").strip() client = Client.query.get(int(raw_id)) if raw_id else Client() nome = (request.form.get("nome") or "").strip() cognome = (request.form.get("cognome") or "").strip() if cognome: client.first_name = to_title_case(nome) or None client.last_name = to_title_case(cognome) or None client.name = None else: client.name = to_title_case(nome) or None client.first_name = None client.last_name = None client.sigla_nazione = _clean(request.form.get("sigla_nazione", ""), upper=True) client.piva_cf = _sanitize_client_piva_cf( sigla_nazione=client.sigla_nazione, piva_cf=_clean(request.form.get("piva_cf", ""), upper=True), ) # Safety: prevent accidental reuse of a piva_cf for IT clients. # For foreign clients (sigla != IT) we allow duplicates, because many sources # may reuse placeholder IDs and we must never overwrite existing records. if client.piva_cf: sigla = normalize_upper(client.sigla_nazione or client.country or "IT") if sigla == "IT": candidates = ( Client.query.filter_by(piva_cf=client.piva_cf) .filter(Client.id != client.id) .all() ) for existing in candidates: existing_sigla = normalize_upper(existing.sigla_nazione or existing.country or "IT") if existing_sigla == "IT": flash( "Impossibile salvare: C.F./P.IVA è già usato da un altro cliente IT.", "danger", ) db.session.rollback() return redirect(url_for("clients.index")) client.address = _clean(request.form.get("indirizzo", "")) client.street_number = _clean(request.form.get("civico", "")) client.cap = _clean(request.form.get("cap", ""), upper=False) client.city = _clean(request.form.get("citta", "")) client.province = _clean(request.form.get("provincia", ""), upper=True) client.country = _clean(request.form.get("nazione", ""), upper=True) # Keep country code consistent when user stores ISO codes in both fields. 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 client.email_pec = (request.form.get("email_pec") or "").strip() or None client.codice_univoco = _clean(request.form.get("cod_univoco", ""), upper=True) client.codice_destinatario = _clean( request.form.get("codice_destinatario", ""), upper=True ) client.phone = (request.form.get("telefono") or "").strip() or None client.website = (request.form.get("sito") or "").strip() or None is_new = client.id is None db.session.add(client) db.session.commit() labels = request.form.getlist("custom_label") values = request.form.getlist("custom_value") replace_custom_fields(owner_type="client", owner_id=client.id, labels=labels, values=values) db.session.commit() log_activity( None, action="client_create" if is_new else "client_update", entity_type="client", entity_id=client.id, ) return redirect(url_for("clients.index")) @bp.post("/delete/") @admin_required def delete(client_id: int): client = Client.query.get_or_404(client_id) def _ensure_snapshot(inv: Invoice, *, company: Company | None) -> bool: if load_original_invoice_xml(inv.id): return True if not company: 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.codice_destinatario 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, ) 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(client_id=client.id).all() failed_invoice_ids: list[int] = [] for inv in linked_invoices: # Always store a DB snapshot before detaching. try: inv.client_snapshot_json = dumps_snapshot(build_client_snapshot(client)) db.session.add(inv) except Exception: pass company = Company.query.get(inv.company_id) if inv.company_id else None if not _ensure_snapshot(inv, company=company): failed_invoice_ids.append(inv.id) if failed_invoice_ids: db.session.rollback() flash( "Impossibile eliminare il cliente: alcune fatture collegate non hanno uno snapshot XML e non è stato possibile generarlo.", "danger", ) return redirect(url_for("clients.index")) # Detach invoices so issued documents remain stable and the client can be deleted. for inv in linked_invoices: inv.client_id = None db.session.add(inv) db.session.commit() db.session.delete(client) db.session.commit() log_activity(None, action="client_delete", entity_type="client", entity_id=client_id) return redirect(url_for("clients.index")) @bp.post("/bulk-delete") @admin_required def bulk_delete(): raw = (request.form.get("client_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("client_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("Nessun cliente selezionato", "warning") return redirect(url_for("clients.index")) clients = Client.query.filter(Client.id.in_(ids)).all() if len(clients) != len(ids): abort(404) def _ensure_snapshot(inv: Invoice, *, client_obj: Client, company: Company | None) -> bool: if load_original_invoice_xml(inv.id): return True if not company: 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 # Snapshot and detach invoices for selected clients, so clients can be deleted. by_id = {c.id: c for c in clients} linked_invoices = Invoice.query.filter(Invoice.client_id.in_(ids)).all() failed_invoice_ids: list[int] = [] for inv in linked_invoices: client_obj = by_id.get(inv.client_id) if client_obj: try: inv.client_snapshot_json = dumps_snapshot(build_client_snapshot(client_obj)) db.session.add(inv) except Exception: pass company = Company.query.get(inv.company_id) if inv.company_id else None if not client_obj: failed_invoice_ids.append(inv.id) continue if not _ensure_snapshot(inv, client_obj=client_obj, company=company): 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("clients.index")) for inv in linked_invoices: inv.client_id = None db.session.add(inv) db.session.commit() Client.query.filter(Client.id.in_(ids)).delete(synchronize_session=False) db.session.commit() flash(f"Eliminati {len(ids)} clienti", "success") for cid in ids: log_activity(None, action="client_delete", entity_type="client", entity_id=cid) return redirect(url_for("clients.index")) @bp.post("/import-csv") @login_required def import_csv(): up = request.files.get("csv_file") if not up: return redirect(url_for("clients.index")) selected_fields = set(request.form.getlist("fields")) if not selected_fields: return redirect(url_for("clients.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("clients.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("clients.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) # Some exported/hand-edited CSVs may contain accidental unquoted newlines inside a row. # csv.reader will split them into multiple rows; try to stitch them back when possible. header_len = len(header) if header_len: fixed: list[list[str]] = [] buf: list[str] | None = None for r in data_rows: if buf is None: buf = r continue if len(buf) < header_len: cont = r # Common case: continuation line starts with the delimiter (e.g. ";US;"), # which csv.reader yields as ["", "US", ...]. Drop the leading empty cell. if cont and (cont[0] or "") == "": cont = cont[1:] buf = buf + cont continue fixed.append(buf) buf = r if buf is not None: fixed.append(buf) data_rows = fixed LABEL_TO_KEY = { "id": "client_id", "clientid": "client_id", "nome": "nome", "cognome": "cognome", "nome(visualizzato)": "display_name", "nomevisualizzato": "display_name", "cf/p.iva": "piva_cf", "cfp.iva": "piva_cf", "cfpiva": "piva_cf", "pivacf": "piva_cf", "piva_cf": "piva_cf", "siglanazione": "sigla_nazione", "indirizzo": "indirizzo", "civico": "civico", "ncivico": "civico", "numerocivico": "civico", "cap": "cap", "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", "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 mapped_fields = set(header_map.values()) effective_mapped_fields = set(mapped_fields) # We can derive ISO2 codes between these two fields during import. if "nazione" in mapped_fields: effective_mapped_fields.add("sigla_nazione") if "sigla_nazione" in mapped_fields: effective_mapped_fields.add("nazione") if not (effective_mapped_fields & selected_fields): sel = ", ".join(sorted(selected_fields)) det = ", ".join(sorted(effective_mapped_fields)) flash( f"Import CSV: nessuna colonna del file corrisponde ai campi selezionati. Selezionati: [{sel}]. Riconosciuti dal CSV: [{det}].", "danger", ) return redirect(url_for("clients.index")) created = 0 updated = 0 skipped = 0 processed = 0 skip_reasons: dict[str, int] = {} def _norm_key(value: str | None) -> str: s = (value or "").strip().lower() if not s: return "" 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 _client_display_name(c: Client) -> str: if c.name: return c.name parts = [(c.first_name or "").strip(), (c.last_name or "").strip()] return " ".join([p for p in parts if p]) def _find_duplicate_non_it(*, incoming_sigla: str, incoming_name: str, incoming_piva_cf: str | None, incoming_email: str | None, incoming_phone: str | None, incoming_address: str | None, incoming_cap: str | None, incoming_city: str | None, ) -> Client | None: name_key = _norm_key(incoming_name) if not name_key: return None candidates: list[Client] = [] # Prefer stable identifiers for lookup to keep queries fast. if incoming_email: candidates = Client.query.filter(Client.email_pec == incoming_email).all() elif incoming_phone: candidates = Client.query.filter(Client.phone == incoming_phone).all() elif incoming_address and incoming_city: q = Client.query.filter( Client.address == incoming_address, Client.city == incoming_city, ) if incoming_cap: q = q.filter(Client.cap == incoming_cap) candidates = q.all() elif incoming_piva_cf: candidates = Client.query.filter(Client.piva_cf == incoming_piva_cf).all() else: return None def _same_sigla(c: Client) -> bool: existing_sigla = normalize_upper(c.sigla_nazione or c.country or "IT") return existing_sigla == incoming_sigla matches: list[Client] = [] for c in candidates: if not _same_sigla(c): continue if _norm_key(_client_display_name(c)) != name_key: continue matches.append(c) if not matches: return None # If we matched by email/phone/address, any match is strong enough. if incoming_email or incoming_phone or (incoming_address and incoming_city): return matches[0] # Fallback (only piva_cf + name): skip only if the match is unambiguous. return matches[0] if len(matches) == 1 else None def _skip(reason: str): nonlocal skipped skipped += 1 skip_reasons[reason] = skip_reasons.get(reason, 0) + 1 def _row_has_selected_values(row: dict[str, str]) -> bool: for field in selected_fields: val = (row.get(field) or "").strip() if val: return True # Allow derived values: many CSVs store ISO2 in "Nazione". if field == "sigla_nazione": v = (row.get("nazione") or "").strip() if v and len(v) == 2: return True if field == "nazione": v = (row.get("sigla_nazione") or "").strip() if v and len(v) == 2: return True # If user imports only "nome", allow using "Nome (visualizzato)". if field == "nome": v = (row.get("display_name") or "").strip() if v: return True return False for r in data_rows: if not any((c or "").strip() for c in r): continue processed += 1 row: dict[str, str] = {} for idx, val in enumerate(r): if idx in header_map: row[header_map[idx]] = (val or "").strip() if not _row_has_selected_values(row): _skip("riga senza valori nei campi selezionati") continue # Precompute values used for deduplication (even if not selected for import). nome = (row.get("nome") or "").strip() cognome = (row.get("cognome") or "").strip() display_name = (row.get("display_name") or "").strip() incoming_name = ( f"{nome} {cognome}".strip() if cognome else (display_name or nome).strip() ) raw_country = _clean(row.get("nazione") or "", upper=True) if row.get("nazione") else None raw_sigla = _clean(row.get("sigla_nazione") or "", upper=True) if row.get("sigla_nazione") else None incoming_sigla = raw_sigla if not incoming_sigla and raw_country and len(raw_country) == 2: incoming_sigla = raw_country if incoming_sigla: incoming_sigla = normalize_upper(incoming_sigla) imported_piva_cf = _clean(row.get("piva_cf") or "", upper=True) if row.get("piva_cf") else None incoming_email = (row.get("email_pec") or "").strip() or None incoming_phone = (row.get("telefono") or "").strip() or None incoming_address = _clean(row.get("indirizzo") or "") if row.get("indirizzo") else None incoming_cap = _clean(row.get("cap") or "", upper=False) if row.get("cap") else None incoming_city = _clean(row.get("citta") or "") if row.get("citta") else None client = _find_client_for_csv_row(row) is_new = client is None # For foreign (non-IT) rows, prevent importing the exact same contact multiple times. # We still allow duplicate C.F./P.IVA across different foreign contacts. if is_new and incoming_sigla and incoming_sigla != "IT": try: dup = _find_duplicate_non_it( incoming_sigla=incoming_sigla, incoming_name=incoming_name, incoming_piva_cf=imported_piva_cf, incoming_email=incoming_email, incoming_phone=incoming_phone, incoming_address=incoming_address, incoming_cap=incoming_cap, incoming_city=incoming_city, ) if dup is not None: _skip("duplicato già presente") continue except Exception: # Never block import due to an unexpected deduplication error. pass if is_new: client = Client() try: if "nome" in selected_fields or "cognome" in selected_fields: if cognome and ("cognome" in selected_fields): client.first_name = to_title_case(nome) if ("nome" in selected_fields) else client.first_name client.last_name = to_title_case(cognome) client.name = None else: # Azienda/persona senza cognome -> usa name if "nome" in selected_fields and nome: # If the CSV includes a pre-composed display name (e.g. "Emanuele Liccardo"), # prefer it when user isn't importing the surname field. if display_name and ("cognome" not in selected_fields): client.name = to_title_case(display_name) else: client.name = to_title_case(nome) if "cognome" in selected_fields and not cognome: client.last_name = None client.first_name = None _ensure_country_code(incoming_sigla) if "sigla_nazione" in selected_fields: # If the CSV has only "Nazione" with ISO2 codes, accept it as sigla too. if raw_sigla: client.sigla_nazione = raw_sigla elif raw_country and len(raw_country) == 2: client.sigla_nazione = raw_country if "nazione" in selected_fields: client.country = raw_country # If CSV provides only sigla_nazione, accept it as country too. if not client.country and raw_sigla and len(raw_sigla) == 2: client.country = raw_sigla # Keep country code consistent when users store ISO codes in both fields. 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 "piva_cf" in selected_fields and row.get("piva_cf"): # Import must preserve exactly what's in the CSV (aside from trimming and uppercasing). # Do not strip country prefixes or apply heuristics here. if not imported_piva_cf: client.piva_cf = None else: # Safety guard: # - For IT rows: enforce uniqueness of C.F./P.IVA to avoid merges. # - For non-IT rows: allow duplicates (even within the same sigla), but never # overwrite existing records because we match only by explicit ID. this_sigla = normalize_upper(incoming_sigla or "IT") if this_sigla == "IT": candidates = Client.query.filter_by(piva_cf=imported_piva_cf).all() for existing in candidates: if existing.id == client.id: continue existing_sigla = normalize_upper( (existing.sigla_nazione or existing.country or "IT") ) if existing_sigla == "IT": _skip("piva_cf già presente per IT") break else: client.piva_cf = imported_piva_cf continue continue client.piva_cf = imported_piva_cf if "indirizzo" in selected_fields: client.address = _clean(row.get("indirizzo") or "") if "civico" in selected_fields: client.street_number = _clean(row.get("civico") or "") if "cap" in selected_fields: client.cap = _clean(row.get("cap") or "", upper=False) if "citta" in selected_fields: client.city = _clean(row.get("citta") or "") if "provincia" in selected_fields: client.province = _clean(row.get("provincia") or "", upper=True) if "email_pec" in selected_fields: client.email_pec = (row.get("email_pec") or "").strip() or None if "cod_univoco" in selected_fields: client.codice_univoco = _clean(row.get("cod_univoco") or "", upper=True) if "codice_destinatario" in selected_fields: client.codice_destinatario = _clean(row.get("codice_destinatario") or "", upper=True) if "telefono" in selected_fields: client.phone = (row.get("telefono") or "").strip() or None if "sito" in selected_fields: client.website = (row.get("sito") or "").strip() or None db.session.add(client) if is_new: created += 1 else: updated += 1 except Exception: _skip("errore durante import") reason_summary = "" if skipped and skip_reasons: parts = sorted(skip_reasons.items(), key=lambda kv: (-kv[1], kv[0])) reason_summary = " (" + "; ".join([f"{k}: {v}" for k, v in parts[:4]]) + ")" if created or updated: db.session.commit() log_activity(None, action="client_import_csv", entity_type="client", entity_id=None) flash( f"Import CSV completato: {processed} righe lette, {created} creati, {updated} aggiornati, {skipped} scartati{reason_summary}.", "success", ) else: db.session.rollback() flash( f"Import CSV: nessuna modifica applicata ({processed} righe lette, {skipped} scartati{reason_summary}).", "warning", ) return redirect(url_for("clients.index"))