382 lines
19 KiB
Python
382 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.db import engine
|
|
from app.commercial_service import ensure_commercial_schema, normalize_tax_id
|
|
from app.opportunity_service import ensure_opportunity_schema
|
|
|
|
|
|
def _clean(value: Any, max_len: int = 260) -> str:
|
|
if value is None:
|
|
return ""
|
|
text_value = re.sub(r"\s+", " ", str(value).replace("\n", " ")).strip()
|
|
return text_value[:max_len]
|
|
|
|
|
|
def _json(value: Any) -> str:
|
|
return json.dumps(value or {}, ensure_ascii=False, default=str)
|
|
|
|
|
|
def _as_dict(value: Any) -> Dict[str, Any]:
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def _first_from(data: Dict[str, Any], *keys: str) -> str:
|
|
for key in keys:
|
|
if key in data and _clean(data.get(key)):
|
|
return _clean(data.get(key))
|
|
return ""
|
|
|
|
|
|
def _deep_first(data: Dict[str, Any], *keys: str) -> str:
|
|
stack: List[Any] = [data]
|
|
seen = 0
|
|
while stack and seen < 100:
|
|
seen += 1
|
|
current = stack.pop(0)
|
|
if isinstance(current, dict):
|
|
direct = _first_from(current, *keys)
|
|
if direct:
|
|
return direct
|
|
for value in current.values():
|
|
if isinstance(value, (dict, list)):
|
|
stack.append(value)
|
|
elif isinstance(current, list):
|
|
stack.extend(v for v in current if isinstance(v, (dict, list)))
|
|
return ""
|
|
|
|
|
|
def _split_pt_address(address: str) -> Tuple[str, str, str]:
|
|
clean = _clean(address, 400)
|
|
if not clean:
|
|
return "", "", ""
|
|
match = re.search(r"\b(\d{4}-\d{3}|\d{4})\b\s*(.*)$", clean)
|
|
if not match:
|
|
return clean, "", ""
|
|
return clean[: match.start()].strip(" ,-"), match.group(1), _clean(match.group(2), 90)
|
|
|
|
|
|
def _normalize_customer_payload(doc: Dict[str, Any]) -> Dict[str, Any]:
|
|
payload = _as_dict(doc.get("payload"))
|
|
record = _as_dict(payload.get("record") or payload.get("jasmin_details") or payload.get("jasmin_payload") or payload)
|
|
nested_customer: Dict[str, Any] = {}
|
|
for key in ("customer", "customerParty", "buyerCustomerParty", "party", "billing", "invoiceTo", "shipTo"):
|
|
if isinstance(record.get(key), dict):
|
|
nested_customer.update(record.get(key) or {})
|
|
merged = {**record, **nested_customer}
|
|
|
|
name = _deep_first(merged, "customerName", "partyName", "buyerCustomerPartyName", "name", "companyName") or _clean(doc.get("company"))
|
|
tax_id = normalize_tax_id(_deep_first(merged, "customerTaxId", "customerTaxID", "companyTaxID", "companyTaxId", "partyTaxID", "taxID", "taxId", "vat", "nif"))
|
|
email = _deep_first(merged, "customerEmail", "electronicMail", "email", "billingEmail")
|
|
phone = _deep_first(merged, "phone", "telephone", "mobile", "phoneNumber")
|
|
street = _deep_first(merged, "streetName", "street", "address", "billingAddress", "addressLine1", "line1")
|
|
parsed_street, parsed_zip, parsed_city = _split_pt_address(street)
|
|
street = parsed_street or street
|
|
postal = _deep_first(merged, "postalZone", "postalCode", "zip", "postcode") or parsed_zip
|
|
city = _deep_first(merged, "cityName", "city", "locality") or parsed_city
|
|
country = _deep_first(merged, "country", "countryCode", "countryKey") or "PT"
|
|
party_key = _deep_first(merged, "customerPartyKey", "buyerCustomerParty", "partyKey", "naturalKey") or _clean(doc.get("customer_party_key"))
|
|
|
|
return {k: v for k, v in {
|
|
"name": name,
|
|
"tax_id": tax_id,
|
|
"email": email,
|
|
"phone": phone,
|
|
"street_name": street,
|
|
"postal_zone": postal,
|
|
"city_name": city,
|
|
"country": country or "PT",
|
|
"jasmin_customer_party_key": party_key,
|
|
"jasmin_customer_id": _deep_first(merged, "customerId", "id", "uuid"),
|
|
}.items() if _clean(v)}
|
|
|
|
|
|
def _missing_fields(customer: Dict[str, Any]) -> List[str]:
|
|
labels = []
|
|
for key, label in [
|
|
("email", "email de faturação"),
|
|
("street_name", "morada fiscal"),
|
|
("postal_zone", "código postal"),
|
|
("city_name", "localidade"),
|
|
]:
|
|
if not _clean(customer.get(key)):
|
|
labels.append(label)
|
|
return labels
|
|
|
|
|
|
def _select_candidate_doc(conn: Any, opportunity_id: str) -> Optional[Dict[str, Any]]:
|
|
rows = conn.execute(text("""
|
|
SELECT
|
|
d.id::text,
|
|
d.customer_id::text,
|
|
d.document_kind,
|
|
d.document_number,
|
|
d.customer_party_key,
|
|
d.payload,
|
|
d.company,
|
|
d.created_at,
|
|
d.updated_at,
|
|
c.id::text AS doc_customer_id,
|
|
c.name AS doc_customer_name,
|
|
c.tax_id AS doc_customer_tax_id,
|
|
c.email AS doc_customer_email,
|
|
c.phone AS doc_customer_phone,
|
|
c.street_name AS doc_customer_street_name,
|
|
c.postal_zone AS doc_customer_postal_zone,
|
|
c.city_name AS doc_customer_city_name,
|
|
c.country AS doc_customer_country,
|
|
c.jasmin_customer_party_key AS doc_customer_party_key,
|
|
c.jasmin_customer_id AS doc_customer_jasmin_id
|
|
FROM commercial_documents d
|
|
LEFT JOIN customers c ON c.id = d.customer_id
|
|
WHERE d.opportunity_id = CAST(:opportunity_id AS UUID)
|
|
AND d.system = 'jasmin'
|
|
AND d.document_kind IN ('quotation', 'invoice', 'proforma')
|
|
AND COALESCE(d.is_active, TRUE) = TRUE
|
|
ORDER BY
|
|
CASE d.document_kind WHEN 'invoice' THEN 1 WHEN 'quotation' THEN 2 WHEN 'proforma' THEN 3 ELSE 4 END,
|
|
COALESCE(d.document_date, d.created_at::date) DESC,
|
|
d.created_at DESC
|
|
LIMIT 5
|
|
"""), {"opportunity_id": opportunity_id}).mappings().all()
|
|
best = None
|
|
best_score = -1
|
|
for row in rows:
|
|
doc = dict(row)
|
|
candidate = _normalize_customer_payload(doc)
|
|
if doc.get("doc_customer_id"):
|
|
candidate = {
|
|
**candidate,
|
|
"id": doc.get("doc_customer_id"),
|
|
"name": candidate.get("name") or doc.get("doc_customer_name"),
|
|
"tax_id": candidate.get("tax_id") or doc.get("doc_customer_tax_id"),
|
|
"email": candidate.get("email") or doc.get("doc_customer_email"),
|
|
"phone": candidate.get("phone") or doc.get("doc_customer_phone"),
|
|
"street_name": candidate.get("street_name") or doc.get("doc_customer_street_name"),
|
|
"postal_zone": candidate.get("postal_zone") or doc.get("doc_customer_postal_zone"),
|
|
"city_name": candidate.get("city_name") or doc.get("doc_customer_city_name"),
|
|
"country": candidate.get("country") or doc.get("doc_customer_country") or "PT",
|
|
"jasmin_customer_party_key": candidate.get("jasmin_customer_party_key") or doc.get("doc_customer_party_key"),
|
|
"jasmin_customer_id": candidate.get("jasmin_customer_id") or doc.get("doc_customer_jasmin_id"),
|
|
}
|
|
score = sum(1 for key in ("name", "tax_id", "email", "street_name", "postal_zone", "city_name") if _clean(candidate.get(key)))
|
|
if score > best_score:
|
|
best_score = score
|
|
best = {"document": doc, "candidate": candidate}
|
|
return best
|
|
|
|
|
|
def get_jasmin_fiscal_sync_preview(opportunity_id: str) -> Dict[str, Any]:
|
|
"""Read-only preview for fiscal data available in linked Jasmin documents."""
|
|
ensure_opportunity_schema()
|
|
ensure_commercial_schema()
|
|
with engine.begin() as conn:
|
|
opp = conn.execute(text("""
|
|
SELECT o.id::text, o.local_customer_id::text, o.title,
|
|
c.id::text AS customer_id, 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
|
|
FROM opportunities o
|
|
LEFT 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()
|
|
if not opp:
|
|
return {"available": False, "reason": "opportunity_not_found"}
|
|
selected = _select_candidate_doc(conn, opportunity_id)
|
|
if not selected:
|
|
return {"available": False, "reason": "no_jasmin_document"}
|
|
candidate = selected["candidate"]
|
|
if not candidate.get("id") and candidate.get("tax_id"):
|
|
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
|
|
FROM customers
|
|
WHERE tax_id = :tax_id
|
|
LIMIT 1
|
|
"""), {"tax_id": candidate.get("tax_id")}).mappings().first()
|
|
if row:
|
|
c = dict(row)
|
|
candidate = {**candidate, "id": c.get("id")}
|
|
for key in ("name", "tax_id", "email", "phone", "street_name", "postal_zone", "city_name", "country", "jasmin_customer_party_key", "jasmin_customer_id"):
|
|
candidate[key] = candidate.get(key) or c.get(key)
|
|
linked = dict(opp)
|
|
linked_tax = normalize_tax_id(linked.get("tax_id"))
|
|
cand_tax = normalize_tax_id(candidate.get("tax_id"))
|
|
conflict = bool(linked.get("customer_id") and linked_tax and cand_tax and linked_tax != cand_tax)
|
|
missing_now = _missing_fields(linked if linked.get("customer_id") else {})
|
|
fillable = [key for key in ("email", "phone", "street_name", "postal_zone", "city_name", "country") if not _clean(linked.get(key)) and _clean(candidate.get(key))]
|
|
if not linked.get("customer_id"):
|
|
fillable = [key for key in ("name", "tax_id", "email", "phone", "street_name", "postal_zone", "city_name", "country") if _clean(candidate.get(key))]
|
|
return {
|
|
"available": bool(candidate.get("name") or candidate.get("tax_id")),
|
|
"conflict": conflict,
|
|
"linked_customer_id": linked.get("customer_id"),
|
|
"candidate_customer_id": candidate.get("id"),
|
|
"candidate": candidate,
|
|
"document": {k: selected["document"].get(k) for k in ("id", "document_kind", "document_number")},
|
|
"fillable_fields": fillable,
|
|
"missing_fields": missing_now,
|
|
"reason": "nif_divergent" if conflict else "ok",
|
|
}
|
|
|
|
|
|
def apply_jasmin_fiscal_sync(opportunity_id: str, *, actor: str = "operator") -> Dict[str, Any]:
|
|
"""Fill only empty customer fields from Jasmin and link/create when safe."""
|
|
ensure_opportunity_schema()
|
|
ensure_commercial_schema()
|
|
preview = get_jasmin_fiscal_sync_preview(opportunity_id)
|
|
if not preview.get("available"):
|
|
raise ValueError("Não foram encontrados dados fiscais Jasmin associados a esta oportunidade.")
|
|
if preview.get("conflict"):
|
|
raise ValueError("NIF divergente entre cliente fiscal local e dados Jasmin; rever associação antes de importar.")
|
|
candidate = dict(preview.get("candidate") or {})
|
|
if not (candidate.get("name") or candidate.get("tax_id")):
|
|
raise ValueError("Dados Jasmin insuficientes para criar/associar cliente fiscal.")
|
|
|
|
with engine.begin() as conn:
|
|
opp = conn.execute(text("""
|
|
SELECT o.id::text, o.local_customer_id::text
|
|
FROM opportunities o
|
|
WHERE o.id = CAST(:opportunity_id AS UUID)
|
|
LIMIT 1
|
|
"""), {"opportunity_id": opportunity_id}).mappings().first()
|
|
linked_customer_id = str(opp.get("local_customer_id") or "") if opp else ""
|
|
target_customer_id = linked_customer_id or str(preview.get("candidate_customer_id") or "")
|
|
created = False
|
|
if target_customer_id:
|
|
existing = conn.execute(text("SELECT id::text FROM customers WHERE id = CAST(:id AS UUID) LIMIT 1"), {"id": target_customer_id}).mappings().first()
|
|
if not existing:
|
|
target_customer_id = ""
|
|
if not target_customer_id:
|
|
target_customer_id = 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, NULLIF(:tax_id,''), NULLIF(:email,''), NULLIF(:phone,''), NULLIF(:street_name,''),
|
|
NULLIF(:postal_zone,''), NULLIF(:city_name,''), COALESCE(NULLIF(:country,''), 'PT'),
|
|
NULLIF(:party_key,''), NULLIF(:jasmin_id,''), CAST(:metadata AS JSONB), now()
|
|
)
|
|
RETURNING id::text
|
|
"""), {
|
|
"name": _clean(candidate.get("name")) or "Cliente Jasmin",
|
|
"tax_id": normalize_tax_id(candidate.get("tax_id")),
|
|
"email": _clean(candidate.get("email")),
|
|
"phone": _clean(candidate.get("phone")),
|
|
"street_name": _clean(candidate.get("street_name")),
|
|
"postal_zone": _clean(candidate.get("postal_zone")),
|
|
"city_name": _clean(candidate.get("city_name")),
|
|
"country": _clean(candidate.get("country") or "PT"),
|
|
"party_key": _clean(candidate.get("jasmin_customer_party_key")),
|
|
"jasmin_id": _clean(candidate.get("jasmin_customer_id")),
|
|
"metadata": _json({"source": "jasmin_fiscal_sync", "opportunity_id": opportunity_id}),
|
|
}).scalar()
|
|
created = True
|
|
before = 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
|
|
FROM customers WHERE id = CAST(:id AS UUID)
|
|
"""), {"id": target_customer_id}).mappings().first()
|
|
before_dict = dict(before or {})
|
|
local_tax = normalize_tax_id(before_dict.get("tax_id"))
|
|
jasmin_tax = normalize_tax_id(candidate.get("tax_id"))
|
|
if local_tax and jasmin_tax and local_tax != jasmin_tax:
|
|
raise ValueError("NIF divergente entre cliente local e Jasmin; importação bloqueada.")
|
|
params = {
|
|
"id": target_customer_id,
|
|
"tax_id": jasmin_tax,
|
|
"email": _clean(candidate.get("email")),
|
|
"phone": _clean(candidate.get("phone")),
|
|
"street_name": _clean(candidate.get("street_name")),
|
|
"postal_zone": _clean(candidate.get("postal_zone")),
|
|
"city_name": _clean(candidate.get("city_name")),
|
|
"country": _clean(candidate.get("country") or "PT"),
|
|
"party_key": _clean(candidate.get("jasmin_customer_party_key")),
|
|
"jasmin_id": _clean(candidate.get("jasmin_customer_id")),
|
|
"metadata": _json({"jasmin_fiscal_sync": {"opportunity_id": opportunity_id, "document": preview.get("document"), "actor": actor}}),
|
|
}
|
|
row = conn.execute(text("""
|
|
UPDATE customers
|
|
SET tax_id = COALESCE(NULLIF(tax_id,''), NULLIF(:tax_id,'')),
|
|
email = COALESCE(NULLIF(email,''), NULLIF(:email,'')),
|
|
phone = COALESCE(NULLIF(phone,''), NULLIF(:phone,'')),
|
|
street_name = COALESCE(NULLIF(street_name,''), NULLIF(:street_name,'')),
|
|
postal_zone = COALESCE(NULLIF(postal_zone,''), NULLIF(:postal_zone,'')),
|
|
city_name = COALESCE(NULLIF(city_name,''), NULLIF(:city_name,'')),
|
|
country = COALESCE(NULLIF(country,''), NULLIF(:country,''), 'PT'),
|
|
jasmin_customer_party_key = COALESCE(NULLIF(jasmin_customer_party_key,''), NULLIF(:party_key,'')),
|
|
jasmin_customer_id = COALESCE(NULLIF(jasmin_customer_id,''), NULLIF(:jasmin_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
|
|
"""), params).mappings().first()
|
|
conn.execute(text("""
|
|
UPDATE opportunities
|
|
SET local_customer_id = CAST(:customer_id AS UUID), updated_at = now()
|
|
WHERE id = CAST(:opportunity_id AS UUID)
|
|
AND (local_customer_id IS NULL OR local_customer_id = CAST(:customer_id AS UUID))
|
|
"""), {"customer_id": target_customer_id, "opportunity_id": opportunity_id})
|
|
conn.execute(text("""
|
|
UPDATE commercial_documents
|
|
SET customer_id = COALESCE(customer_id, CAST(:customer_id AS UUID)), updated_at = now()
|
|
WHERE opportunity_id = CAST(:opportunity_id AS UUID)
|
|
AND system = 'jasmin'
|
|
"""), {"customer_id": target_customer_id, "opportunity_id": opportunity_id})
|
|
after_dict = dict(row or {})
|
|
filled = [key for key in ("tax_id", "email", "phone", "street_name", "postal_zone", "city_name", "country") if not _clean(before_dict.get(key)) and _clean(after_dict.get(key))]
|
|
conn.execute(text("""
|
|
INSERT INTO opportunity_events (id, opportunity_id, event_type, note, payload, created_by)
|
|
VALUES (gen_random_uuid(), CAST(:opportunity_id AS UUID), 'jasmin_fiscal_sync', :note, CAST(:payload AS JSONB), :actor)
|
|
"""), {
|
|
"opportunity_id": opportunity_id,
|
|
"note": "Dados fiscais completados a partir do Jasmin." if filled or created else "Cliente fiscal associado a partir do Jasmin.",
|
|
"payload": _json({"customer_id": target_customer_id, "created_customer": created, "filled_fields": filled, "document": preview.get("document"), "candidate": candidate}),
|
|
"actor": actor,
|
|
})
|
|
return {"customer_id": target_customer_id, "created_customer": created, "filled_fields": filled, "customer": after_dict}
|
|
|
|
|
|
def audit_jasmin_fiscal_gaps(limit: int = 200) -> List[Dict[str, Any]]:
|
|
ensure_opportunity_schema()
|
|
ensure_commercial_schema()
|
|
findings: List[Dict[str, Any]] = []
|
|
with engine.begin() as conn:
|
|
rows = conn.execute(text("""
|
|
SELECT DISTINCT o.id::text, o.title, o.customer_name
|
|
FROM opportunities o
|
|
JOIN commercial_documents d ON d.opportunity_id = o.id AND d.system = 'jasmin'
|
|
LEFT JOIN customers c ON c.id = o.local_customer_id
|
|
WHERE COALESCE(o.status, 'open') <> 'closed'
|
|
AND (
|
|
o.local_customer_id IS NULL
|
|
OR COALESCE(c.email,'') = ''
|
|
OR COALESCE(c.street_name,'') = ''
|
|
OR COALESCE(c.postal_zone,'') = ''
|
|
OR COALESCE(c.city_name,'') = ''
|
|
)
|
|
ORDER BY o.id
|
|
LIMIT :limit
|
|
"""), {"limit": int(limit)}).mappings().all()
|
|
for row in rows:
|
|
preview = get_jasmin_fiscal_sync_preview(str(row.get("id")))
|
|
if preview.get("available") and (preview.get("fillable_fields") or not preview.get("linked_customer_id")):
|
|
cand = preview.get("candidate") or {}
|
|
findings.append({
|
|
"severity": "médio" if not preview.get("linked_customer_id") else "baixo",
|
|
"code": "jasmin_fiscal_data_available",
|
|
"title": row.get("title") or row.get("customer_name") or row.get("id"),
|
|
"detail": "Dados fiscais vazios no ClientFlow mas disponíveis no Jasmin.",
|
|
"url": f"/opportunities/{row.get('id')}",
|
|
"candidate_name": cand.get("name"),
|
|
"candidate_tax_id": cand.get("tax_id"),
|
|
})
|
|
return findings
|