"""Domínio comercial ClientFlow. Guarda clientes locais, documentos Jasmin (orçamentos/faturas) e envios externos de forma normalizada. Esta camada evita transformar operation_links num armazenamento principal e permite vários orçamentos/faturas por cliente e por oportunidade. """ from __future__ import annotations import json import re import uuid from decimal import Decimal from typing import Any, Dict, Iterable, List, Optional from sqlalchemy import text from app.db import engine _SCHEMA_READY = False def _json(value: Any) -> str: return json.dumps(value or {}, ensure_ascii=False, default=str) def _clean(value: Any) -> str: return str(value or "").strip() def normalize_tax_id(value: Any) -> str: """Normaliza NIF/VAT PT para uso local e Jasmin. Nos testes reais Jasmin, getCustomerByCompanyTaxId aceitou 504931946 e não encontrou PT504931946. Por isso guardamos sem prefixo PT. """ tax_id = _clean(value).upper().replace(" ", "").replace("-", "") if tax_id.startswith("PT"): tax_id = tax_id[2:] return tax_id def normalize_fiscal_name(value: Any) -> str: """Normaliza nome fiscal para evitar duplicados quando falta NIF. No domínio ClientFlow o nome fiscal de empresa é tratado como chave de identidade forte. Esta normalização é deliberadamente simples e estável: remove pontuação/ruído de espaçamento, preservando a designação legal. """ name = _clean(value).casefold() name = re.sub(r"[^0-9a-záàâãéêíóôõúç]+", " ", name, flags=re.IGNORECASE) return re.sub(r"\s+", " ", name).strip() def normalize_email(value: Any) -> str: """Normaliza email para identidade lógica do cliente. A criação de clientes deve tratar ``CLIENTE@EXEMPLO.COM`` e `` cliente@exemplo.com `` como a mesma ficha comercial. """ email = _clean(value).casefold() email = re.sub(r"\s+", "", email) return email def _validate_email_or_empty(value: Any) -> str: email = normalize_email(value) if not email: return "" # Validação deliberadamente simples: evita lixo óbvio sem bloquear emails válidos raros. if not re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", email): raise ValueError("Email inválido.") return email def ensure_commercial_schema() -> None: global _SCHEMA_READY if _SCHEMA_READY: return with engine.begin() as conn: conn.execute(text(""" CREATE TABLE IF NOT EXISTS customers ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name TEXT NOT NULL, tax_id TEXT, email TEXT, phone TEXT, street_name TEXT, postal_zone TEXT, city_name TEXT, country TEXT NOT NULL DEFAULT 'PT', jasmin_customer_party_key TEXT, jasmin_customer_id TEXT, metadata JSONB NOT NULL DEFAULT '{}'::jsonb, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ) """)) conn.execute(text(""" CREATE UNIQUE INDEX IF NOT EXISTS ux_customers_tax_id ON customers(tax_id) WHERE tax_id IS NOT NULL AND tax_id <> '' """)) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_customers_name ON customers(name)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_customers_email_normalized ON customers((lower(trim(email)))) WHERE email IS NOT NULL AND trim(email) <> ''")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_customers_jasmin_key ON customers(jasmin_customer_party_key)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_customers_external_identity ON customers USING gin (metadata)")) conn.execute(text(""" CREATE TABLE IF NOT EXISTS commercial_documents ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), customer_id UUID REFERENCES customers(id) ON DELETE SET NULL, opportunity_id UUID REFERENCES opportunities(id) ON DELETE SET NULL, system TEXT NOT NULL DEFAULT 'jasmin', document_kind TEXT NOT NULL, external_id TEXT, external_url TEXT, company TEXT, document_type TEXT, serie TEXT, series_number INTEGER, document_number TEXT, customer_party_key TEXT, status TEXT NOT NULL DEFAULT 'draft', amount NUMERIC(12,2), tax_amount NUMERIC(12,2), total_amount NUMERIC(12,2), currency TEXT DEFAULT 'EUR', parent_document_id UUID REFERENCES commercial_documents(id) ON DELETE SET NULL, version_number INTEGER, role TEXT NOT NULL DEFAULT 'current', is_primary BOOLEAN NOT NULL DEFAULT TRUE, is_active BOOLEAN NOT NULL DEFAULT TRUE, document_date DATE, due_date DATE, payload JSONB NOT NULL DEFAULT '{}'::jsonb, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ) """)) # Upgrades aditivos para bases que já tenham a tabela. for stmt in [ "ALTER TABLE commercial_documents ADD COLUMN IF NOT EXISTS external_url TEXT", "ALTER TABLE commercial_documents ADD COLUMN IF NOT EXISTS parent_document_id UUID REFERENCES commercial_documents(id) ON DELETE SET NULL", "ALTER TABLE commercial_documents ADD COLUMN IF NOT EXISTS version_number INTEGER", "ALTER TABLE commercial_documents ADD COLUMN IF NOT EXISTS role TEXT NOT NULL DEFAULT 'current'", "ALTER TABLE commercial_documents ADD COLUMN IF NOT EXISTS is_primary BOOLEAN NOT NULL DEFAULT TRUE", "ALTER TABLE commercial_documents ADD COLUMN IF NOT EXISTS is_active BOOLEAN NOT NULL DEFAULT TRUE", "ALTER TABLE commercial_documents ADD COLUMN IF NOT EXISTS tax_amount NUMERIC(12,2)", "ALTER TABLE commercial_documents ADD COLUMN IF NOT EXISTS total_amount NUMERIC(12,2)", ]: conn.execute(text(stmt)) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_commercial_documents_customer ON commercial_documents(customer_id)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_commercial_documents_opportunity ON commercial_documents(opportunity_id)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_commercial_documents_external ON commercial_documents(system, external_id)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_commercial_documents_kind ON commercial_documents(document_kind, status)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_commercial_documents_role ON commercial_documents(opportunity_id, document_kind, role, is_primary)")) # Hotfix v4927.1: em bases já usadas podem existir vários documentos # Jasmin/Odoo ligados à mesma oportunidade e ao mesmo tipo. Ao adicionar # role/is_primary com DEFAULT current/TRUE, todos passariam a ser # “atuais”, fazendo o índice único falhar no arranque e deixando o # serviço indisponível atrás do nginx. Antes de criar o índice, mantemos # só o documento mais recente como primário e movemos os restantes para # histórico. conn.execute(text(""" WITH ranked AS ( SELECT id, ROW_NUMBER() OVER ( PARTITION BY opportunity_id, system, document_kind, COALESCE(role, 'current') ORDER BY COALESCE(updated_at, created_at) DESC, created_at DESC, id DESC ) AS rn FROM commercial_documents WHERE opportunity_id IS NOT NULL AND COALESCE(is_primary, TRUE) = TRUE AND COALESCE(role, 'current') IN ('current', 'accepted') ) UPDATE commercial_documents AS d SET role = CASE WHEN COALESCE(d.role, 'current') IN ('current', 'accepted') THEN 'historical' ELSE COALESCE(d.role, 'historical') END, is_primary = FALSE, is_active = FALSE, updated_at = now() FROM ranked AS r WHERE d.id = r.id AND r.rn > 1 """)) # v4927.2: não criar índice UNIQUE durante o arranque. # Em produção podem existir duplicados históricos que ainda não foram # reconciliados; se o índice único falhar, toda a app fica indisponível # e o nginx devolve 502. A regra de “um principal por fase” fica # aplicada pela normalização acima e pelos serviços que promovem um # documento a atual. O índice de apoio é não único e seguro em bases # reais com dados legados. conn.execute(text("DROP INDEX IF EXISTS ux_commercial_documents_primary_role")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_commercial_documents_primary_role ON commercial_documents(opportunity_id, system, document_kind, role) WHERE opportunity_id IS NOT NULL AND COALESCE(is_primary, TRUE) = TRUE AND COALESCE(role, 'current') IN ('current', 'accepted')")) # v4928.1: migração leve de registos antigos/reconstruídos. # Não apaga documentos nem altera fases. Apenas evita que orçamentos e # pró-formas legados continuem a aparecer como documento principal quando # já existe fatura Jasmin atual/aceite para a mesma oportunidade. conn.execute(text(""" WITH primary_invoice AS ( SELECT DISTINCT ON (opportunity_id, system) id, opportunity_id, system, COALESCE(document_date, created_at::date) AS invoice_date, created_at FROM commercial_documents WHERE opportunity_id IS NOT NULL AND system = 'jasmin' AND document_kind = 'invoice' AND COALESCE(is_primary, TRUE) = TRUE AND COALESCE(role, 'current') IN ('current', 'accepted') ORDER BY opportunity_id, system, COALESCE(document_date, created_at::date) DESC, created_at DESC ) UPDATE commercial_documents AS d SET role = 'historical', is_primary = FALSE, is_active = FALSE, payload = COALESCE(d.payload, '{}'::jsonb) || jsonb_build_object('legacy_reason', 'superseded_by_invoice_v4928_1'), updated_at = now() FROM primary_invoice AS inv WHERE d.opportunity_id = inv.opportunity_id AND d.system = inv.system AND d.document_kind IN ('quotation', 'proforma') AND COALESCE(d.role, 'current') IN ('current', 'accepted') AND COALESCE(d.is_primary, TRUE) = TRUE AND COALESCE(d.document_date, d.created_at::date) <= inv.invoice_date """)) # Marca oportunidades reconstruídas com fatura e sem task pendente para a UI # poder explicar que são processos antigos/auditoria, não um funil limpo. conn.execute(text(""" UPDATE opportunities AS o SET metadata = COALESCE(o.metadata, '{}'::jsonb) || jsonb_build_object( 'clientflow_record_mode', COALESCE(o.metadata->>'clientflow_record_mode', 'reconstructed_invoice_review'), 'clientflow_legacy_migrated_by', 'v4928_1' ), updated_at = now() WHERE EXISTS ( SELECT 1 FROM commercial_documents d WHERE d.opportunity_id = o.id AND d.system = 'jasmin' AND d.document_kind = 'invoice' AND COALESCE(d.role, 'current') IN ('current', 'accepted') AND COALESCE(d.is_primary, TRUE) = TRUE ) AND NOT EXISTS ( SELECT 1 FROM tasks t WHERE t.opportunity_id = o.id AND t.status = 'pending' ) AND COALESCE(o.metadata->>'clientflow_record_mode', '') = '' """)) conn.execute(text(""" CREATE TABLE IF NOT EXISTS commercial_document_lines ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), document_id UUID NOT NULL REFERENCES commercial_documents(id) ON DELETE CASCADE, opportunity_item_id UUID REFERENCES opportunity_items(id) ON DELETE SET NULL, line_index INTEGER NOT NULL DEFAULT 0, local_product_id UUID, jasmin_sales_item TEXT, description TEXT NOT NULL, quantity NUMERIC(12,3) NOT NULL DEFAULT 1, unit TEXT DEFAULT 'UN', unit_price NUMERIC(12,2) NOT NULL DEFAULT 0, tax_schema TEXT DEFAULT 'NORMAL', total_amount NUMERIC(12,2), payload JSONB NOT NULL DEFAULT '{}'::jsonb, created_at TIMESTAMPTZ NOT NULL DEFAULT now() ) """)) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_commercial_document_lines_document ON commercial_document_lines(document_id)")) conn.execute(text(""" CREATE TABLE IF NOT EXISTS shipments ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), customer_id UUID REFERENCES customers(id) ON DELETE SET NULL, opportunity_id UUID REFERENCES opportunities(id) ON DELETE SET NULL, system TEXT NOT NULL DEFAULT 'packlink', external_reference TEXT, carrier TEXT, service_id TEXT, service_name TEXT, status TEXT NOT NULL DEFAULT 'draft', tracking_code TEXT, tracking_url TEXT, label_url TEXT, price NUMERIC(12,2), currency TEXT DEFAULT 'EUR', payload JSONB NOT NULL DEFAULT '{}'::jsonb, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ) """)) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_shipments_opportunity ON shipments(opportunity_id)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_shipments_external ON shipments(system, external_reference)")) # Relação explícita oportunidade -> cliente local. Mantém o campo antigo # opportunities.customer_id intacto para compatibilidade com Chatwoot/legado. conn.execute(text("ALTER TABLE opportunities ADD COLUMN IF NOT EXISTS local_customer_id UUID REFERENCES customers(id) ON DELETE SET NULL")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_opportunities_local_customer ON opportunities(local_customer_id)")) _SCHEMA_READY = True def get_customer_by_tax_id(tax_id: str) -> Optional[Dict[str, Any]]: ensure_commercial_schema() normalized = normalize_tax_id(tax_id) if not normalized: return None with engine.begin() as conn: row = conn.execute(text(""" SELECT id::text, name, tax_id, email, phone, street_name, postal_zone, city_name, country, jasmin_customer_party_key, jasmin_customer_id, metadata, created_at, updated_at FROM customers WHERE tax_id = :tax_id LIMIT 1 """), {"tax_id": normalized}).mappings().first() return dict(row) if row else None def get_customer(customer_id: str) -> Optional[Dict[str, Any]]: ensure_commercial_schema() with engine.begin() as conn: row = conn.execute(text(""" SELECT id::text, name, tax_id, email, phone, street_name, postal_zone, city_name, country, jasmin_customer_party_key, jasmin_customer_id, metadata, created_at, updated_at FROM customers WHERE id = CAST(:id AS UUID) LIMIT 1 """), {"id": customer_id}).mappings().first() return dict(row) if row else None def upsert_customer(data: Dict[str, Any]) -> Dict[str, Any]: ensure_commercial_schema() tax_id = normalize_tax_id(data.get("tax_id") or data.get("companyTaxID")) name = _clean(data.get("name") or data.get("customerName") or data.get("customer_name")) if not name: raise ValueError("Nome do cliente é obrigatório") params = { "name": name, "tax_id": tax_id or None, "email": _validate_email_or_empty(data.get("email") or data.get("electronicMail")) or None, "phone": _clean(data.get("phone") or data.get("telephone")) or None, "street_name": _clean(data.get("street_name") or data.get("streetName")) or None, "postal_zone": _clean(data.get("postal_zone") or data.get("postalZone")) or None, "city_name": _clean(data.get("city_name") or data.get("cityName")) or None, "country": _clean(data.get("country") or "PT") or "PT", "jasmin_customer_party_key": _clean(data.get("jasmin_customer_party_key") or data.get("customerPartyKey")) or None, "jasmin_customer_id": _clean(data.get("jasmin_customer_id") or data.get("jasmin_id") or data.get("id")) or None, "metadata": _json(data.get("metadata") or {}), } with engine.begin() as conn: # Identidade forte por email quando não há NIF. Evita fichas duplicadas # por maiúsculas/espaços no email. O NIF continua a ter prioridade. if not params["tax_id"] and params["email"]: existing_by_email = conn.execute(text(""" SELECT id::text, name, tax_id, email, phone, street_name, postal_zone, city_name, country, jasmin_customer_party_key, jasmin_customer_id, metadata, created_at, updated_at FROM customers WHERE lower(trim(email)) = :email ORDER BY updated_at DESC LIMIT 1 """), {"email": params["email"]}).mappings().first() if existing_by_email: row = conn.execute(text(""" UPDATE customers SET name = COALESCE(NULLIF(:name, ''), name), phone = COALESCE(:phone, phone), street_name = COALESCE(:street_name, street_name), postal_zone = COALESCE(:postal_zone, postal_zone), city_name = COALESCE(:city_name, city_name), country = COALESCE(:country, country), jasmin_customer_party_key = COALESCE(:jasmin_customer_party_key, jasmin_customer_party_key), jasmin_customer_id = COALESCE(:jasmin_customer_id, jasmin_customer_id), metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:metadata AS JSONB), updated_at = now() WHERE id = CAST(:id AS UUID) RETURNING id::text, name, tax_id, email, phone, street_name, postal_zone, city_name, country, jasmin_customer_party_key, jasmin_customer_id, metadata, created_at, updated_at """), {**params, "id": existing_by_email["id"]}).mappings().first() return dict(row or {}) if params["tax_id"]: row = conn.execute(text(""" INSERT INTO customers ( name, tax_id, email, phone, street_name, postal_zone, city_name, country, jasmin_customer_party_key, jasmin_customer_id, metadata, updated_at ) VALUES ( :name, :tax_id, :email, :phone, :street_name, :postal_zone, :city_name, :country, :jasmin_customer_party_key, :jasmin_customer_id, CAST(:metadata AS JSONB), now() ) ON CONFLICT (tax_id) WHERE tax_id IS NOT NULL AND tax_id <> '' DO UPDATE SET name = EXCLUDED.name, email = COALESCE(EXCLUDED.email, customers.email), phone = COALESCE(EXCLUDED.phone, customers.phone), street_name = COALESCE(EXCLUDED.street_name, customers.street_name), postal_zone = COALESCE(EXCLUDED.postal_zone, customers.postal_zone), city_name = COALESCE(EXCLUDED.city_name, customers.city_name), country = COALESCE(EXCLUDED.country, customers.country), jasmin_customer_party_key = COALESCE(EXCLUDED.jasmin_customer_party_key, customers.jasmin_customer_party_key), jasmin_customer_id = COALESCE(EXCLUDED.jasmin_customer_id, customers.jasmin_customer_id), metadata = customers.metadata || EXCLUDED.metadata, updated_at = now() RETURNING id::text, name, tax_id, email, phone, street_name, postal_zone, city_name, country, jasmin_customer_party_key, jasmin_customer_id, metadata, created_at, updated_at """), params).mappings().first() else: normalized_name = normalize_fiscal_name(params["name"]) row = conn.execute(text(""" SELECT id::text, name, tax_id, email, phone, street_name, postal_zone, city_name, country, jasmin_customer_party_key, jasmin_customer_id, metadata, created_at, updated_at FROM customers WHERE COALESCE(tax_id, '') = '' AND trim(regexp_replace(regexp_replace(lower(name), '[^0-9a-záàâãéêíóôõúç]+', ' ', 'g'), '\\s+', ' ', 'g')) = :normalized_name LIMIT 1 """), {"normalized_name": normalized_name}).mappings().first() if row: row = conn.execute(text(""" UPDATE customers SET email = COALESCE(:email, email), phone = COALESCE(:phone, phone), street_name = COALESCE(:street_name, street_name), postal_zone = COALESCE(:postal_zone, postal_zone), city_name = COALESCE(:city_name, city_name), country = COALESCE(:country, country), jasmin_customer_party_key = COALESCE(:jasmin_customer_party_key, jasmin_customer_party_key), jasmin_customer_id = COALESCE(:jasmin_customer_id, jasmin_customer_id), metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:metadata AS JSONB), updated_at = now() WHERE id = CAST(:id AS UUID) RETURNING id::text, name, tax_id, email, phone, street_name, postal_zone, city_name, country, jasmin_customer_party_key, jasmin_customer_id, metadata, created_at, updated_at """), {**params, "id": row["id"]}).mappings().first() else: row = conn.execute(text(""" INSERT INTO customers ( name, email, phone, street_name, postal_zone, city_name, country, jasmin_customer_party_key, jasmin_customer_id, metadata, updated_at ) VALUES ( :name, :email, :phone, :street_name, :postal_zone, :city_name, :country, :jasmin_customer_party_key, :jasmin_customer_id, CAST(:metadata AS JSONB), now() ) RETURNING id::text, name, tax_id, email, phone, street_name, postal_zone, city_name, country, jasmin_customer_party_key, jasmin_customer_id, metadata, created_at, updated_at """), params).mappings().first() return dict(row or {}) def _next_document_version(conn, opportunity_id: Optional[str], document_kind: str) -> int: if not opportunity_id: return 1 value = conn.execute(text(""" SELECT COALESCE(MAX(version_number), 0) + 1 FROM commercial_documents WHERE opportunity_id = CAST(:opportunity_id AS UUID) AND document_kind = :document_kind """), {"opportunity_id": opportunity_id, "document_kind": document_kind}).scalar() return int(value or 1) def create_commercial_document( *, document_kind: str, customer_id: Optional[str] = None, opportunity_id: Optional[str] = None, system: str = "jasmin", external_id: Optional[str] = None, external_url: Optional[str] = None, company: Optional[str] = None, document_type: Optional[str] = None, serie: Optional[str] = None, series_number: Optional[int] = None, document_number: Optional[str] = None, customer_party_key: Optional[str] = None, status: str = "created", amount: Any = None, tax_amount: Any = None, total_amount: Any = None, currency: str = "EUR", parent_document_id: Optional[str] = None, payload: Optional[Dict[str, Any]] = None, document_date: Optional[str] = None, due_date: Optional[str] = None, ) -> Dict[str, Any]: ensure_commercial_schema() document_kind = _clean(document_kind) if not document_kind: raise ValueError("document_kind é obrigatório") with engine.begin() as conn: version_number = _next_document_version(conn, opportunity_id, document_kind) if document_kind == "quotation" and opportunity_id: conn.execute(text(""" UPDATE commercial_documents SET status = CASE WHEN status IN ('created','sent','draft') THEN 'superseded' ELSE status END, is_active = FALSE, updated_at = now() WHERE opportunity_id = CAST(:opportunity_id AS UUID) AND document_kind = 'quotation' AND is_active = TRUE """), {"opportunity_id": opportunity_id}) row = conn.execute(text(""" INSERT INTO commercial_documents ( customer_id, opportunity_id, system, document_kind, external_id, external_url, company, document_type, serie, series_number, document_number, customer_party_key, status, amount, tax_amount, total_amount, currency, parent_document_id, version_number, is_active, document_date, due_date, payload, updated_at ) VALUES ( CAST(:customer_id AS UUID), CAST(:opportunity_id AS UUID), :system, :document_kind, :external_id, :external_url, :company, :document_type, :serie, :series_number, :document_number, :customer_party_key, :status, :amount, :tax_amount, :total_amount, :currency, CAST(:parent_document_id AS UUID), :version_number, TRUE, CAST(:document_date AS DATE), CAST(:due_date AS DATE), CAST(:payload AS JSONB), now() ) RETURNING id::text, customer_id::text, opportunity_id::text, system, document_kind, external_id, external_url, company, document_type, serie, series_number, document_number, customer_party_key, status, amount, tax_amount, total_amount, currency, parent_document_id::text, version_number, is_active, document_date, due_date, payload, created_at, updated_at """), { "customer_id": customer_id, "opportunity_id": opportunity_id, "system": system, "document_kind": document_kind, "external_id": external_id, "external_url": external_url, "company": company, "document_type": document_type, "serie": serie, "series_number": series_number, "document_number": document_number, "customer_party_key": customer_party_key, "status": status, "amount": amount, "tax_amount": tax_amount, "total_amount": total_amount if total_amount is not None else amount, "currency": currency, "parent_document_id": parent_document_id, "version_number": version_number, "document_date": document_date, "due_date": due_date, "payload": _json(payload), }).mappings().first() return dict(row or {}) def add_document_lines(document_id: str, lines: Iterable[Dict[str, Any]]) -> None: ensure_commercial_schema() with engine.begin() as conn: for idx, line in enumerate(lines): conn.execute(text(""" INSERT INTO commercial_document_lines ( document_id, opportunity_item_id, line_index, local_product_id, jasmin_sales_item, description, quantity, unit, unit_price, tax_schema, total_amount, payload ) VALUES ( CAST(:document_id AS UUID), CAST(:opportunity_item_id AS UUID), :line_index, CAST(:local_product_id AS UUID), :jasmin_sales_item, :description, :quantity, :unit, :unit_price, :tax_schema, :total_amount, CAST(:payload AS JSONB) ) """), { "document_id": document_id, "opportunity_item_id": line.get("opportunity_item_id"), "line_index": int(line.get("line_index") if line.get("line_index") is not None else idx), "local_product_id": line.get("local_product_id"), "jasmin_sales_item": line.get("jasmin_sales_item"), "description": line.get("description") or "Linha", "quantity": line.get("quantity") or 1, "unit": line.get("unit") or "UN", "unit_price": line.get("unit_price") or 0, "tax_schema": line.get("tax_schema") or "NORMAL", "total_amount": line.get("total_amount"), "payload": _json(line.get("payload") or {}), }) def list_commercial_documents(opportunity_id: Optional[str] = None, customer_id: Optional[str] = None, limit: int = 100) -> List[Dict[str, Any]]: ensure_commercial_schema() filters = [] params: Dict[str, Any] = {"limit": int(limit)} if opportunity_id: filters.append("cd.opportunity_id = CAST(:opportunity_id AS UUID)") params["opportunity_id"] = opportunity_id if customer_id: filters.append("cd.customer_id = CAST(:customer_id AS UUID)") params["customer_id"] = customer_id where = "WHERE " + " AND ".join(filters) if filters else "" with engine.begin() as conn: rows = conn.execute(text(f""" SELECT cd.id::text, cd.customer_id::text, cd.opportunity_id::text, cd.system, cd.document_kind, cd.external_id, cd.external_url, cd.company, cd.document_type, cd.serie, cd.series_number, cd.document_number, cd.customer_party_key, cd.status, cd.amount, cd.tax_amount, cd.total_amount, cd.currency, cd.parent_document_id::text, cd.version_number, cd.role, cd.is_primary, cd.is_active, cd.document_date, cd.due_date, cd.payload, cd.created_at, cd.updated_at, c.name AS customer_name, c.tax_id AS customer_tax_id FROM commercial_documents cd LEFT JOIN customers c ON c.id = cd.customer_id {where} ORDER BY CASE COALESCE(cd.role, 'current') WHEN 'current' THEN 0 WHEN 'accepted' THEN 1 WHEN 'related' THEN 2 WHEN 'historical' THEN 3 ELSE 4 END, COALESCE(cd.is_primary, FALSE) DESC, cd.created_at DESC LIMIT :limit """), params).mappings().all() return [dict(r) for r in rows] def get_commercial_document(document_id: str) -> Optional[Dict[str, Any]]: ensure_commercial_schema() with engine.begin() as conn: row = conn.execute(text(""" SELECT cd.id::text, cd.customer_id::text, cd.opportunity_id::text, cd.system, cd.document_kind, cd.external_id, cd.external_url, cd.company, cd.document_type, cd.serie, cd.series_number, cd.document_number, cd.customer_party_key, cd.status, cd.amount, cd.tax_amount, cd.total_amount, cd.currency, cd.parent_document_id::text, cd.version_number, cd.role, cd.is_primary, cd.is_active, cd.document_date, cd.due_date, cd.payload, cd.created_at, cd.updated_at, c.name AS customer_name, c.tax_id AS customer_tax_id FROM commercial_documents cd LEFT JOIN customers c ON c.id = cd.customer_id WHERE cd.id = CAST(:id AS UUID) LIMIT 1 """), {"id": document_id}).mappings().first() return dict(row) if row else None def get_latest_active_quotation(opportunity_id: str) -> Optional[Dict[str, Any]]: ensure_commercial_schema() with engine.begin() as conn: row = conn.execute(text(""" SELECT id::text, customer_id::text, opportunity_id::text, system, document_kind, external_id, external_url, company, document_type, serie, series_number, document_number, customer_party_key, status, amount, total_amount, currency, parent_document_id::text, version_number, is_active, payload, created_at, updated_at FROM commercial_documents WHERE opportunity_id = CAST(:opportunity_id AS UUID) AND document_kind = 'quotation' AND status NOT IN ('cancelled','failed') ORDER BY is_active DESC, created_at DESC LIMIT 1 """), {"opportunity_id": opportunity_id}).mappings().first() return dict(row) if row else None def find_invoice_for_parent(parent_document_id: str) -> Optional[Dict[str, Any]]: ensure_commercial_schema() with engine.begin() as conn: row = conn.execute(text(""" SELECT id::text, external_id, document_number, status, created_at FROM commercial_documents WHERE parent_document_id = CAST(:parent_document_id AS UUID) AND document_kind = 'invoice' ORDER BY created_at DESC LIMIT 1 """), {"parent_document_id": parent_document_id}).mappings().first() return dict(row) if row else None def mark_document_status(document_id: str, status: str, payload: Optional[Dict[str, Any]] = None) -> None: ensure_commercial_schema() with engine.begin() as conn: conn.execute(text(""" UPDATE commercial_documents SET status = :status, payload = payload || CAST(:payload AS JSONB), updated_at = now() WHERE id = CAST(:id AS UUID) """), {"id": document_id, "status": status, "payload": _json(payload or {})}) def update_commercial_document_details(document_id: str, details: Dict[str, Any]) -> Optional[Dict[str, Any]]: """Atualiza campos normalizados depois de consultar o documento no sistema externo. Usado sobretudo para Jasmin, porque o POST devolve apenas o UUID e o número/série precisam de um GET subsequente. Campos ausentes não apagam dados já existentes. """ ensure_commercial_schema() payload = details.get("payload") or {} with engine.begin() as conn: row = conn.execute(text(""" UPDATE commercial_documents SET company = COALESCE(:company, company), document_type = COALESCE(:document_type, document_type), serie = COALESCE(:serie, serie), series_number = COALESCE(:series_number, series_number), document_number = COALESCE(:document_number, document_number), amount = COALESCE(:amount, amount), tax_amount = COALESCE(:tax_amount, tax_amount), total_amount = COALESCE(:total_amount, total_amount), currency = COALESCE(:currency, currency), document_date = COALESCE(CAST(:document_date AS DATE), document_date), due_date = COALESCE(CAST(:due_date AS DATE), due_date), payload = payload || CAST(:payload AS JSONB), updated_at = now() WHERE id = CAST(:id AS UUID) RETURNING id::text, customer_id::text, opportunity_id::text, system, document_kind, external_id, external_url, company, document_type, serie, series_number, document_number, customer_party_key, status, amount, tax_amount, total_amount, currency, parent_document_id::text, version_number, is_active, document_date, due_date, payload, created_at, updated_at """), { "id": document_id, "company": details.get("company"), "document_type": details.get("document_type"), "serie": details.get("serie"), "series_number": details.get("series_number"), "document_number": details.get("document_number"), "amount": details.get("amount"), "tax_amount": details.get("tax_amount"), "total_amount": details.get("total_amount"), "currency": details.get("currency"), "document_date": details.get("document_date"), "due_date": details.get("due_date"), "payload": _json(payload), }).mappings().first() return dict(row) if row else None def upsert_shipment_record(data: Dict[str, Any]) -> Dict[str, Any]: ensure_commercial_schema() params = { "customer_id": data.get("customer_id"), "opportunity_id": data.get("opportunity_id"), "system": data.get("system") or "packlink", "external_reference": data.get("external_reference") or data.get("reference"), "carrier": data.get("carrier"), "service_id": data.get("service_id"), "service_name": data.get("service_name"), "status": data.get("status") or "created", "tracking_code": data.get("tracking_code"), "tracking_url": data.get("tracking_url"), "label_url": data.get("label_url"), "price": data.get("price"), "currency": data.get("currency") or "EUR", "payload": _json(data.get("payload") or {}), } with engine.begin() as conn: row = conn.execute(text(""" INSERT INTO shipments ( customer_id, opportunity_id, system, external_reference, carrier, service_id, service_name, status, tracking_code, tracking_url, label_url, price, currency, payload, updated_at ) VALUES ( CAST(:customer_id AS UUID), CAST(:opportunity_id AS UUID), :system, :external_reference, :carrier, :service_id, :service_name, :status, :tracking_code, :tracking_url, :label_url, :price, :currency, CAST(:payload AS JSONB), now() ) RETURNING id::text, customer_id::text, opportunity_id::text, system, external_reference, carrier, service_id, service_name, status, tracking_code, tracking_url, label_url, price, currency, payload, created_at, updated_at """), params).mappings().first() return dict(row or {}) def list_customers(q: Optional[str] = None, limit: int = 200) -> List[Dict[str, Any]]: """Lista clientes locais normalizados. A página Customers passa a ser a fonte principal dos dados fiscais. """ ensure_commercial_schema() params: Dict[str, Any] = {"limit": int(limit)} where = "" if q: params["q"] = f"%{str(q).strip()}%" where = """ WHERE name ILIKE :q OR COALESCE(tax_id, '') ILIKE :q OR COALESCE(email, '') ILIKE :q OR COALESCE(phone, '') ILIKE :q OR COALESCE(jasmin_customer_party_key, '') ILIKE :q """ with engine.begin() as conn: rows = conn.execute(text(f""" SELECT c.id::text, c.name, c.tax_id, c.email, c.phone, c.street_name, c.postal_zone, c.city_name, c.country, c.jasmin_customer_party_key, c.jasmin_customer_id, c.metadata, c.created_at, c.updated_at, COUNT(DISTINCT o.id) AS opportunity_count, COUNT(DISTINCT cd.id) FILTER (WHERE cd.document_kind = 'quotation') AS quotation_count, COUNT(DISTINCT cd.id) FILTER (WHERE cd.document_kind = 'invoice') AS invoice_count FROM customers c LEFT JOIN opportunities o ON COALESCE(o.local_customer_id::text, '') = c.id::text LEFT JOIN commercial_documents cd ON cd.customer_id = c.id {where} GROUP BY c.id ORDER BY c.updated_at DESC, c.created_at DESC LIMIT :limit """), params).mappings().all() return [dict(r) for r in rows] def update_customer(customer_id: str, data: Dict[str, Any]) -> Dict[str, Any]: """Atualiza uma ficha de cliente local. Não chama Jasmin diretamente; a sincronização é feita por jasmin_service/outbox. """ ensure_commercial_schema() params = { "id": customer_id, "name": _clean(data.get("name")), "tax_id": normalize_tax_id(data.get("tax_id")), "email": _validate_email_or_empty(data.get("email")) or None, "phone": _clean(data.get("phone")) or None, "street_name": _clean(data.get("street_name")) or None, "postal_zone": _clean(data.get("postal_zone")) or None, "city_name": _clean(data.get("city_name")) or None, "country": _clean(data.get("country") or "PT") or "PT", "jasmin_customer_party_key": _clean(data.get("jasmin_customer_party_key")) or None, "jasmin_customer_id": _clean(data.get("jasmin_customer_id")) or None, "metadata": _json(data.get("metadata") or {}), } if not params["name"]: raise ValueError("Nome do cliente é obrigatório") with engine.begin() as conn: if params["tax_id"]: duplicate = conn.execute(text(""" SELECT id::text, name, tax_id, email, phone, street_name, postal_zone, city_name, country, jasmin_customer_party_key, jasmin_customer_id, metadata, created_at, updated_at FROM customers WHERE tax_id = :tax_id AND id <> CAST(:id AS UUID) LIMIT 1 """), {"tax_id": params["tax_id"], "id": params["id"]}).mappings().first() if duplicate: raise DuplicateCustomerTaxIdError(params["tax_id"], dict(duplicate)) row = conn.execute(text(""" UPDATE customers SET name = :name, tax_id = NULLIF(:tax_id, ''), email = :email, phone = :phone, street_name = :street_name, postal_zone = :postal_zone, city_name = :city_name, country = :country, jasmin_customer_party_key = :jasmin_customer_party_key, jasmin_customer_id = :jasmin_customer_id, metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:metadata AS JSONB), updated_at = now() WHERE id = CAST(:id AS UUID) RETURNING id::text, name, tax_id, email, phone, street_name, postal_zone, city_name, country, jasmin_customer_party_key, jasmin_customer_id, metadata, created_at, updated_at """), params).mappings().first() if not row: raise ValueError("Cliente não encontrado") return dict(row) def link_customer_to_opportunity(customer_id: str, opportunity_id: str) -> None: """Liga um cliente local a uma oportunidade sem duplicar dados fiscais.""" ensure_commercial_schema() with engine.begin() as conn: conn.execute(text(""" UPDATE opportunities SET local_customer_id = CAST(:customer_id AS UUID), updated_at = now() WHERE id = CAST(:opportunity_id AS UUID) """), {"customer_id": customer_id, "opportunity_id": opportunity_id}) def unlink_customer_from_opportunity(opportunity_id: str) -> None: ensure_commercial_schema() with engine.begin() as conn: conn.execute(text(""" UPDATE opportunities SET local_customer_id = NULL, updated_at = now() WHERE id = CAST(:opportunity_id AS UUID) """), {"opportunity_id": opportunity_id}) def get_customer_for_opportunity(opportunity_id: str) -> Optional[Dict[str, Any]]: ensure_commercial_schema() with engine.begin() as conn: row = conn.execute(text(""" SELECT c.id::text, c.name, c.tax_id, c.email, c.phone, c.street_name, c.postal_zone, c.city_name, c.country, c.jasmin_customer_party_key, c.jasmin_customer_id, c.metadata, c.created_at, c.updated_at FROM opportunities o JOIN customers c ON c.id = o.local_customer_id WHERE o.id = CAST(:opportunity_id AS UUID) LIMIT 1 """), {"opportunity_id": opportunity_id}).mappings().first() return dict(row) if row else None def list_opportunities_for_customer(customer_id: str, limit: int = 50) -> List[Dict[str, Any]]: ensure_commercial_schema() with engine.begin() as conn: rows = conn.execute(text(""" SELECT id::text, title, stage, status, customer_name, customer_email, product_interest, value_amount, currency, updated_at, created_at FROM opportunities WHERE local_customer_id = CAST(:customer_id AS UUID) ORDER BY updated_at DESC LIMIT :limit """), {"customer_id": customer_id, "limit": int(limit)}).mappings().all() return [dict(r) for r in rows] def list_shipments(opportunity_id: Optional[str] = None, customer_id: Optional[str] = None, limit: int = 100) -> List[Dict[str, Any]]: ensure_commercial_schema() filters = [] params: Dict[str, Any] = {"limit": int(limit)} if opportunity_id: filters.append("s.opportunity_id = CAST(:opportunity_id AS UUID)") params["opportunity_id"] = opportunity_id if customer_id: filters.append("s.customer_id = CAST(:customer_id AS UUID)") params["customer_id"] = customer_id where = "WHERE " + " AND ".join(filters) if filters else "" with engine.begin() as conn: rows = conn.execute(text(f""" SELECT s.id::text, s.customer_id::text, s.opportunity_id::text, s.system, s.external_reference, s.carrier, s.service_id, s.service_name, s.status, s.tracking_code, s.tracking_url, s.label_url, s.price, s.currency, s.payload, s.created_at, s.updated_at, c.name AS customer_name FROM shipments s LEFT JOIN customers c ON c.id = s.customer_id {where} ORDER BY s.created_at DESC LIMIT :limit """), params).mappings().all() return [dict(r) for r in rows]