diff --git a/app/commercial_service.py b/app/commercial_service.py
index 144fabb..94ccf29 100644
--- a/app/commercial_service.py
+++ b/app/commercial_service.py
@@ -718,7 +718,6 @@ def list_customers(q: Optional[str] = None, limit: int = 200) -> List[Dict[str,
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:
diff --git a/app/email_identity_extraction_service.py b/app/email_identity_extraction_service.py
index 73617f2..3f9c7ea 100644
--- a/app/email_identity_extraction_service.py
+++ b/app/email_identity_extraction_service.py
@@ -776,7 +776,6 @@ def _normalize_stored_identity(identity: Optional[Dict[str, Any]]) -> Optional[D
def latest_identity_for_opportunity(opportunity_id: str) -> Optional[Dict[str, Any]]:
- ensure_email_identity_schema()
with engine.begin() as conn:
row = conn.execute(text("""
SELECT id::text, opportunity_id::text, task_id::text, message_id::text,
diff --git a/app/fiscal_enrichment_service.py b/app/fiscal_enrichment_service.py
index 543f5e0..fd3d452 100644
--- a/app/fiscal_enrichment_service.py
+++ b/app/fiscal_enrichment_service.py
@@ -36,7 +36,11 @@ from app.commercial_service import (
upsert_customer,
)
from app.opportunity_service import ensure_opportunity_schema, get_opportunity, list_opportunities
-from app.email_identity_extraction_service import extract_identity_for_opportunity, is_plausible_company_mention
+from app.email_identity_extraction_service import (
+ extract_identity_for_opportunity,
+ is_plausible_company_mention,
+ latest_identity_for_opportunity,
+)
CACHE_SOURCE = "informa_pipeline_api"
@@ -823,7 +827,10 @@ def _identity_conflicts_with_linked_customer(identity: Optional[Dict[str, Any]],
return not _identity_mentions_match_name(identity, linked_name)
-def email_identity_review_for_opportunity(opportunity_id: str, *, refresh: bool = False) -> Dict[str, Any]:
+def email_identity_review_for_opportunity(
+ opportunity_id: str, *, refresh: bool = False,
+ opportunity: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
"""Return the operator-facing identity review for an opportunity.
This is read-only except when refresh=True, where it stores a new extraction.
@@ -831,16 +838,18 @@ def email_identity_review_for_opportunity(opportunity_id: str, *, refresh: bool
can show cases like: email mentions Dietimport S.A. but the opportunity is
linked to Fmrl - Imobiliária S.A.
"""
- ensure_fiscal_enrichment_schema()
- opportunity = get_opportunity(opportunity_id)
+ opportunity = opportunity or get_opportunity(opportunity_id)
if not opportunity:
return {"ok": False, "reason": "opportunity_not_found"}
try:
- identity = extract_identity_for_opportunity(
- opportunity_id,
- refresh=refresh,
- use_llm=bool(getattr(settings, "email_identity_extraction_use_llm", True)),
- )
+ if refresh:
+ identity = extract_identity_for_opportunity(
+ opportunity_id,
+ refresh=True,
+ use_llm=bool(getattr(settings, "email_identity_extraction_use_llm", True)),
+ )
+ else:
+ identity = latest_identity_for_opportunity(opportunity_id)
except Exception as exc:
return {"ok": False, "reason": f"email_identity_extraction_failed: {exc}"}
@@ -1266,7 +1275,6 @@ def reject_fiscal_suggestion(suggestion_id: str, *, actor: str = "operator", rea
def list_fiscal_suggestions_for_opportunity(opportunity_id: str, *, limit: int = 5) -> List[Dict[str, Any]]:
- ensure_fiscal_enrichment_schema()
with engine.begin() as conn:
rows = conn.execute(text("""
SELECT
diff --git a/app/jasmin_backfill_service.py b/app/jasmin_backfill_service.py
index 2261688..d2318f0 100644
--- a/app/jasmin_backfill_service.py
+++ b/app/jasmin_backfill_service.py
@@ -437,7 +437,11 @@ def _email_domain(email: Any) -> str:
return text_value.rsplit("@", 1)[-1].strip()
-def find_jasmin_document_candidates_for_opportunity(opportunity_id: str, *, limit: int = 5) -> List[Dict[str, Any]]:
+def find_jasmin_document_candidates_for_opportunity(
+ opportunity_id: str, *, limit: int = 5,
+ opportunity: Optional[Dict[str, Any]] = None,
+ current_documents: Optional[List[Dict[str, Any]]] = None,
+) -> List[Dict[str, Any]]:
"""Find Jasmin documents that probably belong to an opportunity but are not imported yet.
This is deliberately conservative and prioritizes fiscal identity (NIF/customer_id)
@@ -446,7 +450,7 @@ def find_jasmin_document_candidates_for_opportunity(opportunity_id: str, *, limi
quotation instead of duplicating it.
"""
with engine.begin() as conn:
- opp = conn.execute(text("""
+ opp = opportunity or conn.execute(text("""
SELECT
o.id::text,
o.customer_name,
@@ -470,7 +474,8 @@ def find_jasmin_document_candidates_for_opportunity(opportunity_id: str, *, limi
raw_domain = _email_domain(fiscal_email)
domain = "" if is_public_email_domain(raw_domain) else raw_domain
- current_docs = conn.execute(text("""
+ if current_documents is None:
+ current_docs = conn.execute(text("""
SELECT
id::text, document_number, external_id, document_kind, document_date,
total_amount, amount, payload, created_at, updated_at
@@ -479,7 +484,13 @@ def find_jasmin_document_candidates_for_opportunity(opportunity_id: str, *, limi
AND system = 'jasmin'
AND document_kind IN ('quotation', 'proforma')
ORDER BY document_date DESC NULLS LAST, created_at DESC
- """), {"opportunity_id": opportunity_id}).mappings().all()
+ """), {"opportunity_id": opportunity_id}).mappings().all()
+ else:
+ current_docs = [
+ row for row in current_documents
+ if str(row.get("system") or "") == "jasmin"
+ and str(row.get("document_kind") or "") in {"quotation", "proforma"}
+ ]
candidate_columns = """
ri.id::text,
diff --git a/app/jasmin_fiscal_sync_service.py b/app/jasmin_fiscal_sync_service.py
index 9091ef5..d6389fc 100644
--- a/app/jasmin_fiscal_sync_service.py
+++ b/app/jasmin_fiscal_sync_service.py
@@ -182,8 +182,6 @@ def _select_candidate_doc(conn: Any, opportunity_id: str) -> Optional[Dict[str,
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,
diff --git a/app/operation_service.py b/app/operation_service.py
index c303078..ec45326 100644
--- a/app/operation_service.py
+++ b/app/operation_service.py
@@ -86,7 +86,9 @@ def get_operation_links(opportunity_id: str) -> List[Dict[str, Any]]:
"""), {"opportunity_id": opportunity_id}).mappings().all()
return [dict(r) for r in rows]
-def _commercial_document_operation_fallbacks(opportunity_id: str) -> Dict[tuple, Dict[str, Any]]:
+def _commercial_document_operation_fallbacks(
+ opportunity_id: str, resolved_documents: Optional[List[Dict[str, Any]]] = None,
+) -> Dict[tuple, Dict[str, Any]]:
"""Infer operation cards from linked commercial_documents when operation_links lag.
Reconstructed opportunities often have commercial_documents imported from
@@ -95,7 +97,7 @@ def _commercial_document_operation_fallbacks(opportunity_id: str) -> Dict[tuple,
"""
try:
from app.document_reconciliation_service import resolve_document_links, select_valid_primary
- resolved = resolve_document_links(opportunity_id)
+ resolved = resolved_documents if resolved_documents is not None else resolve_document_links(opportunity_id)
rows = [row for kind in ("invoice", "quotation", "quote", "proforma")
if (row := select_valid_primary(resolved, kind)) is not None]
except Exception:
@@ -132,10 +134,13 @@ def _commercial_document_operation_fallbacks(opportunity_id: str) -> Dict[tuple,
return fallbacks
-def get_operation_snapshot(opportunity_id: str) -> Dict[str, Any]:
- links = get_operation_links(opportunity_id)
+def get_operation_snapshot(
+ opportunity_id: str, *, links: Optional[List[Dict[str, Any]]] = None,
+ resolved_documents: Optional[List[Dict[str, Any]]] = None,
+) -> Dict[str, Any]:
+ links = list(links) if links is not None else get_operation_links(opportunity_id)
by_key = {(x["system"], x["external_type"]): x for x in links}
- commercial_fallbacks = _commercial_document_operation_fallbacks(opportunity_id)
+ commercial_fallbacks = _commercial_document_operation_fallbacks(opportunity_id, resolved_documents)
cards = []
for card in OPERATION_CARDS:
link = by_key.get((card["system"], card["external_type"])) or commercial_fallbacks.get((card["system"], card["external_type"]))
diff --git a/app/opportunity_next_action_service.py b/app/opportunity_next_action_service.py
index 4010cb4..653720b 100644
--- a/app/opportunity_next_action_service.py
+++ b/app/opportunity_next_action_service.py
@@ -55,10 +55,13 @@ def _operation_snapshot_safe(opportunity_id: str) -> dict[str, Any]:
return {"cards": [], "links": []}
-def _build_db_evidence(opportunity_id: str) -> OpportunityEvidence | None:
+def _build_db_evidence(
+ opportunity_id: str, preloaded: Optional[Dict[str, Any]] = None,
+) -> OpportunityEvidence | None:
+ preloaded = preloaded or {}
params = {"opportunity_id": opportunity_id}
with engine.begin() as conn:
- opp = _first_row(conn, """
+ opp = preloaded.get("opportunity") or _first_row(conn, """
SELECT
id::text,
stage,
@@ -73,7 +76,9 @@ def _build_db_evidence(opportunity_id: str) -> OpportunityEvidence | None:
if not opp:
return None
- tasks = _rows(conn, """
+ tasks = preloaded.get("tasks")
+ if tasks is None:
+ tasks = _rows(conn, """
SELECT id::text, action_code, action, note, priority, route, status, due_at, created_at, metadata
FROM tasks
WHERE opportunity_id = CAST(:opportunity_id AS UUID)
@@ -88,14 +93,16 @@ def _build_db_evidence(opportunity_id: str) -> OpportunityEvidence | None:
# SELECT id::text, external_id, document_kind
# The SQL now qualifies these fields because links and documents both
# have ids; relationship comes exclusively from the canonical link.
- from app.document_reconciliation_service import resolve_document_links
- docs = [row for row in resolve_document_links(opportunity_id, conn=conn)
- if row.get("system") == "jasmin" and row.get("relationship") in
+ docs = preloaded.get("resolved_documents")
+ if docs is None:
+ from app.document_reconciliation_service import resolve_document_links
+ docs = resolve_document_links(opportunity_id, conn=conn)
+ docs = [row for row in docs if row.get("system") == "jasmin" and row.get("relationship") in
{"PRIMARY", "SECONDARY", "HISTORICAL"}]
- linked_customer = None
+ linked_customer = preloaded.get("customer")
customer_id = opp.get("fiscal_customer_id") or opp.get("customer_id")
- if customer_id:
+ if customer_id and linked_customer is None:
linked_customer = _first_row(conn, """
SELECT
id::text,
@@ -120,7 +127,7 @@ def _build_db_evidence(opportunity_id: str) -> OpportunityEvidence | None:
LIMIT 1
""", params)
- snapshot = _operation_snapshot_safe(opportunity_id)
+ snapshot = preloaded.get("operation_snapshot") or _operation_snapshot_safe(opportunity_id)
fiscal_complete = False
if linked_customer:
fiscal_complete = bool(
@@ -144,14 +151,17 @@ def _build_db_evidence(opportunity_id: str) -> OpportunityEvidence | None:
)
-def get_opportunity_next_action(opportunity_id: str) -> Dict[str, Any]:
+def get_opportunity_next_action(
+ opportunity_id: str, *, preloaded: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
"""Return the recommended operator action for one opportunity.
This remains a read-only service and returns the legacy dict shape, but the
decision is now produced by the company workflow engine.
"""
- evidence = _build_db_evidence(opportunity_id)
+ evidence = (_build_db_evidence(opportunity_id, preloaded=preloaded)
+ if preloaded is not None else _build_db_evidence(opportunity_id))
if evidence is None:
return OpportunityNextAction(
action_code="NOT_FOUND",
diff --git a/tests/test_email_identity_review_read_semantics.py b/tests/test_email_identity_review_read_semantics.py
new file mode 100644
index 0000000..df20286
--- /dev/null
+++ b/tests/test_email_identity_review_read_semantics.py
@@ -0,0 +1,84 @@
+from inspect import getsource
+
+import app.email_identity_extraction_service as extraction_service
+import app.fiscal_enrichment_service as service
+from app.admin_ui.pages import opportunities as opportunity_page
+
+
+OPPORTUNITY = {
+ "id": "11111111-1111-1111-1111-111111111111",
+ "local_customer_id": None,
+ "linked_customer_id": None,
+ "linked_customer_name": None,
+ "linked_customer_tax_id": None,
+}
+
+
+def test_read_review_uses_latest_only_and_never_upserts(monkeypatch):
+ calls = {"latest": 0, "extract": 0, "upsert": 0}
+
+ def latest(opportunity_id):
+ calls["latest"] += 1
+ return {"company_mentions": ["Example Lda"], "confidence": 0.9}
+
+ monkeypatch.setattr(service, "latest_identity_for_opportunity", latest)
+ monkeypatch.setattr(service, "extract_identity_for_opportunity", lambda *a, **k: calls.__setitem__("extract", calls["extract"] + 1))
+ monkeypatch.setattr(extraction_service, "upsert_identity_extraction_for_opportunity", lambda *a, **k: calls.__setitem__("upsert", calls["upsert"] + 1))
+ monkeypatch.setattr(service, "_find_internal_customer_by_identity", lambda *a, **k: None)
+
+ result = service.email_identity_review_for_opportunity(
+ OPPORTUNITY["id"], refresh=False, opportunity=dict(OPPORTUNITY),
+ )
+
+ assert result["ok"] is True
+ assert result["identity"]["company_mentions"] == ["Example Lda"]
+ assert calls == {"latest": 1, "extract": 0, "upsert": 0}
+
+
+def test_read_review_without_stored_identity_returns_identity_none(monkeypatch):
+ monkeypatch.setattr(service, "latest_identity_for_opportunity", lambda opportunity_id: None)
+ monkeypatch.setattr(
+ service, "extract_identity_for_opportunity",
+ lambda *a, **k: (_ for _ in ()).throw(AssertionError("read path extracted identity")),
+ )
+
+ result = service.email_identity_review_for_opportunity(
+ OPPORTUNITY["id"], refresh=False, opportunity=dict(OPPORTUNITY),
+ )
+
+ assert result["ok"] is True
+ assert result["identity"] is None
+ assert result["valid_company_mentions"] == []
+ assert result["conflict"] is False
+
+
+def test_refresh_review_still_performs_extraction(monkeypatch):
+ calls = []
+
+ def extract(opportunity_id, **kwargs):
+ calls.append((opportunity_id, kwargs))
+ return {"company_mentions": [], "confidence": 0.5}
+
+ monkeypatch.setattr(service, "extract_identity_for_opportunity", extract)
+ monkeypatch.setattr(
+ service, "latest_identity_for_opportunity",
+ lambda *a, **k: (_ for _ in ()).throw(AssertionError("refresh used latest-only path")),
+ )
+
+ result = service.email_identity_review_for_opportunity(
+ OPPORTUNITY["id"], refresh=True, opportunity=dict(OPPORTUNITY),
+ )
+
+ assert result["ok"] is True
+ assert calls and calls[0][1]["refresh"] is True
+
+
+def test_opportunity_detail_get_has_no_identity_write_or_schema_initialization():
+ review_source = getsource(service.email_identity_review_for_opportunity)
+ detail_source = getsource(opportunity_page.opportunity_detail_page)
+
+ assert "latest_identity_for_opportunity(opportunity_id)" in review_source
+ assert "ensure_email_identity_schema" not in review_source
+ assert "upsert_identity_extraction_for_opportunity" not in review_source
+ assert "ensure_pending_task_for_next_action" not in detail_source
+ assert "materialized_task" not in detail_source
diff --git a/tests/test_opportunity_detail_context.py b/tests/test_opportunity_detail_context.py
new file mode 100644
index 0000000..f7dd0fa
--- /dev/null
+++ b/tests/test_opportunity_detail_context.py
@@ -0,0 +1,100 @@
+from inspect import getsource
+
+import app.admin_dashboard as dashboard
+import app.admin_ui.pages.opportunities as page
+import app.commercial_service as commercial
+import app.integration_outbox_service as outbox
+import app.jasmin_backfill_service as backfill
+import app.operation_service as operation
+
+
+def test_detail_context_loads_reusable_groups_once(monkeypatch):
+ calls = {}
+
+ def once(name, value):
+ def load(*args, **kwargs):
+ calls[name] = calls.get(name, 0) + 1
+ return value
+ return load
+
+ opportunity = {"id": "opp", "local_customer_id": "customer", "status": "open", "stage": "NEW_LEAD"}
+ monkeypatch.setattr(page, "get_opportunity", once("opportunity", opportunity))
+ monkeypatch.setattr(page, "list_opportunity_tasks", once("tasks", []))
+ monkeypatch.setattr(page, "list_opportunity_events", once("events", []))
+ monkeypatch.setattr(page, "list_opportunity_items", once("items", []))
+ monkeypatch.setattr(page, "list_products", once("products", []))
+ monkeypatch.setattr(page, "list_communications_for_opportunity", once("communications", []))
+ monkeypatch.setattr(page, "_opportunity_odoo_rows", once("odoo_candidates", ([], [])))
+ monkeypatch.setattr(commercial, "get_customer_for_opportunity", once("customer", {"id": "customer"}))
+ monkeypatch.setattr(commercial, "list_commercial_documents", once("documents", []))
+ monkeypatch.setattr(operation, "get_operation_links", once("operation_links", []))
+ monkeypatch.setattr(operation, "get_operation_snapshot", once("operation_snapshot", {"cards": [], "links": []}))
+ monkeypatch.setattr(outbox, "list_outbox", once("outbox", []))
+ monkeypatch.setattr(backfill, "find_jasmin_document_candidates_for_opportunity", once("jasmin_candidates", []))
+
+ context = page.load_opportunity_detail_context("opp")
+
+ assert context is not None
+ assert calls == {
+ "opportunity": 1, "tasks": 1, "events": 1, "items": 1, "products": 1,
+ "communications": 1, "odoo_candidates": 1, "customer": 1, "documents": 1,
+ "operation_links": 1, "operation_snapshot": 1, "outbox": 1,
+ "jasmin_candidates": 1,
+ }
+
+
+def test_products_panel_preloaded_render_matches_standalone(monkeypatch):
+ items = [{"id": "i1", "product_name": "Produto", "quantity": 1, "unit_price": 10,
+ "total_price": 10, "status": "ACTIVE", "jasmin_sales_item": "A1", "metadata": {}}]
+ products = [{"id": "p1", "sku": "SKU", "jasmin_sales_item": "A1", "name": "Produto"}]
+ monkeypatch.setattr(dashboard, "list_opportunity_items", lambda opportunity_id: items)
+ monkeypatch.setattr(dashboard, "list_products", lambda **kwargs: products)
+
+ standalone = dashboard.opportunity_products_panel_html("opp")
+ preloaded = dashboard.opportunity_products_panel_html(
+ "opp", preloaded_items=items, preloaded_products=products,
+ )
+ assert preloaded == standalone
+
+
+def test_detail_route_threads_context_into_repeated_panels():
+ source = getsource(page.opportunity_detail_page)
+ assert "load_opportunity_detail_context(opportunity_id)" in source
+ assert "preloaded_documents=context.resolved_documents" in source
+ assert "preloaded_items=context.opportunity_items" in source
+ assert "preloaded_products=context.active_products" in source
+ assert "preloaded_candidates=context.odoo_candidates" in source
+ assert "preloaded_items=context.outbox_items" in source
+ assert "preloaded=context.next_action_preloaded()" in source
+
+
+def test_detail_read_path_contains_no_schema_initialization_or_ddl():
+ readers = (
+ page.load_opportunity_detail_context,
+ commercial.get_customer_for_opportunity,
+ commercial.list_commercial_documents,
+ commercial.list_customers,
+ operation.get_operation_links,
+ )
+ forbidden = ("ensure_", "CREATE TABLE", "CREATE INDEX", "ALTER TABLE")
+ for reader in readers:
+ source = getsource(reader)
+ for token in forbidden:
+ assert token not in source, (reader.__name__, token)
+
+ route_source = getsource(page.opportunity_detail_page)
+ assert "ensure_pending_task_for_next_action" not in route_source
+ assert "materialized_task" not in route_source
+
+
+def test_context_next_action_customer_aliases_preserve_evidence_contract():
+ context = page.OpportunityDetailContext(
+ opportunity_id="opp", opportunity={"id": "opp", "local_customer_id": "customer"},
+ customer={"id": "customer", "email": "a@b.pt", "street_name": "Rua",
+ "postal_zone": "1000-001", "city_name": "Lisboa"},
+ )
+ customer = context.next_action_preloaded()["customer"]
+ assert customer["billing_email"] == "a@b.pt"
+ assert customer["address"] == "Rua"
+ assert customer["postal_code"] == "1000-001"
+ assert customer["city"] == "Lisboa"
diff --git a/tests/test_v4928_1_5_37_odoo_sync_panel_static.py b/tests/test_v4928_1_5_37_odoo_sync_panel_static.py
index 4bdd5c2..f4037d2 100644
--- a/tests/test_v4928_1_5_37_odoo_sync_panel_static.py
+++ b/tests/test_v4928_1_5_37_odoo_sync_panel_static.py
@@ -6,7 +6,9 @@ ROOT = Path(__file__).resolve().parents[1]
def test_odoo_status_panel_added_to_opportunity_detail():
page = (ROOT / "app/admin_ui/pages/opportunities.py").read_text()
assert 'Odoo' in page
- assert '
{odoo_status_panel_html(opportunity_id)}
' in page
+ assert '
{odoo_status_panel_html(' in page
+ assert 'preloaded_links=' in page
+ assert 'preloaded_candidates=context.odoo_candidates' in page
assert '@router.get("/opportunities/{opportunity_id}/partials/odoo-status"' in page
assert '@router.post("/opportunities/{opportunity_id}/odoo/sync-status")' in page
assert '@router.post("/opportunities/{opportunity_id}/odoo/link-candidate/{item_id}")' in page
diff --git a/tests/test_v4928_1_5_96_materialize_send_invoice_task_static.py b/tests/test_v4928_1_5_96_materialize_send_invoice_task_static.py
index 11ecbde..11614ea 100644
--- a/tests/test_v4928_1_5_96_materialize_send_invoice_task_static.py
+++ b/tests/test_v4928_1_5_96_materialize_send_invoice_task_static.py
@@ -9,11 +9,12 @@ def test_materializer_limits_to_send_invoice_and_is_idempotent():
assert 'ON CONFLICT (idempotency_key) DO NOTHING' in text
-def test_opportunity_page_materializes_next_action_task_and_refreshes_tasks():
+def test_opportunity_page_is_read_only_and_does_not_materialize_tasks():
text = Path('app/admin_ui/pages/opportunities.py').read_text()
- assert 'ensure_pending_task_for_next_action' in text
- assert 'materialized_task.get("created")' in text
- assert 'tasks = list_opportunity_tasks(opportunity_id, limit=100)' in text
+ detail = text.split('async def opportunity_detail_page', 1)[1].split('@router.', 1)[0]
+ assert 'ensure_pending_task_for_next_action' not in detail
+ assert 'materialized_task' not in detail
+ assert 'GET detail is strictly read-only' in detail
def test_repair_script_uses_central_next_action():