feat(web): paritate editor mapare import-preview cu pagina /mapari

Panoul inline "Operatii de mapat la cod RAR" din preview-ul de import folosea
doar sugestii fuzzy si nu arata sursa sugestiei. Acum are paritate 1:1 cu pagina
/mapari: aceeasi sugestie_principala (GOLD partajat > SILVER > embeddings k-NN)
si acelasi badge sursa (confirmat / similar / non-operatie).

- _collect_unmapped_ops primeste `conn`: ataseaza sugestie_principala +
  surse_sugestie via enrich_suggestions, cu ensure_embeddings_corpus o data
  inainte de bucla (replica pattern-ul din pending_unmapped). Init default pe
  fiecare entry (inclusiv conn=None) -> contract template identic.
  SUGGESTION-ONLY: nu atinge resolve_prestatii/load_mapping (#13).
- _web_compute_preview paseaza conn=conn la _collect_unmapped_ops.
- _preview_import.html: preselect din sugestie_principala > fuzzy>=60 + badge
  sursa (clase .sugg-sursa--{confirmat,similar,nul} deja existente in base.html).
- Test de paritate TARE: seed embeddings + GOLD/SILVER/NUL, batch import cu
  needs_mapping, verifica _web_compute_preview()["unmapped_ops"] ==
  pending_unmapped(conn, account) pe sugestie_principala + surse_sugestie, cate
  un caz per sursa (gold/silver/embedding/nul).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

app/web/routes.py

@@ -2694,12 +2694,21 @@ def post_sterge_format_coloane(
 # Toate rutele /_import/* returneaza fragmente HTML (target #import-section).  #
 # =========================================================================== #
 
-def _collect_unmapped_ops(preview_rows: list[dict], nomenclator: list[dict]) -> list[dict]:
+def _collect_unmapped_ops(preview_rows: list[dict], nomenclator: list[dict], conn=None) -> list[dict]:
-    """Operatii distincte nemapate dintr-un preview de import (staging), cu sugestii fuzzy.
+    """Operatii distincte nemapate dintr-un preview de import (staging), cu sugestii fuzzy + enriched.
 
-    Echivalentul lui pending_unmapped() dar pe randuri de PREVIEW (import in staging,
+    Echivalentul lui `pending_unmapped()` dar pe randuri de PREVIEW (import in staging,
     inca neexistente ca submissions). Aduna doar prestatiile fara cod_prestatie
     (cele cu auto_send=0 au deja cod -> nu apar aici). Sortare: cele mai blocate intai.
+
+    L14-S6 / paritate editor: cand `conn` e dat, ataseaza `sugestie_principala`
+    (GOLD partajat > SILVER > embeddings k-NN) si `surse_sugestie` din `enrich_suggestions`,
+    exact ca `pending_unmapped` — asa panoul inline al preview-ului are aceeasi sugestie
+    principala + badge sursa ca pagina /mapari. SUGGESTION-ONLY: nu atinge
+    `resolve_prestatii`/`load_mapping` (#13). Degradare gratioasa pe embeddings (#16b).
+
+    `sugestie_principala`/`surse_sugestie` se initializeaza pe FIECARE entry (chiar cand
+    conn=None) ca contractul catre template sa fie identic indiferent de call-site.
     """
     agg: dict[str, dict[str, Any]] = {}
     for row in preview_rows:
@@ -2715,9 +2724,22 @@ def _collect_unmapped_ops(preview_rows: list[dict], nomenclator: list[dict]) ->
             if not entry["denumire"] and item.get("denumire"):
                 entry["denumire"] = item.get("denumire")
             entry["blocked"] += 1
+
+    # Indexeaza corpusul embeddings o data inainte de bucla (no-op cand flagul e off).
+    if conn is not None:
+        ensure_embeddings_corpus(conn, nomenclator)
+
     out: list[dict] = []
     for entry in agg.values():
         entry["suggestions"] = suggest_codes(entry["denumire"], nomenclator, limit=5)
+        # Init default pe FIECARE entry -> contract template identic (conn=None inclus).
+        entry["sugestie_principala"] = None
+        entry["surse_sugestie"] = {"gold_partajat": None, "silver": None, "embedding": None, "nul": False}
+        # L14-S6: imbogatire cu GOLD partajat > SILVER > embeddings (SUGGESTION-ONLY, #13)
+        if conn is not None:
+            enriched = enrich_suggestions(conn, entry["denumire"])
+            entry["sugestie_principala"] = enriched["sugestie_principala"]
+            entry["surse_sugestie"] = enriched["surse"]
         out.append(entry)
     out.sort(key=lambda e: (-e["blocked"], e["cod_op_service"]))
     return out
@@ -2940,7 +2962,7 @@ def _web_compute_preview(
         "summary": summary,
         "total": len(preview_rows),
         "filename": batch["filename"],
-        "unmapped_ops": _collect_unmapped_ops(preview_rows, nomenclator),
+        "unmapped_ops": _collect_unmapped_ops(preview_rows, nomenclator, conn=conn),
         "nomenclator": nomenclator,
     }
 

app/web/templates/_preview_import.html

@@ -85,7 +85,8 @@
         <input type="hidden" name="csrf_token" value="{{ csrf_token or '' }}">
         {% for e in unmapped_ops %}
         {%- set top = e.suggestions[0] if e.suggestions else None -%}
-        {%- set preselect = top.cod_prestatie if (top and top.score >= 60) else '' -%}
+        {# L14-S6: pre-selectare din sugestie_principala (GOLD/SILVER/embedding) > fuzzy>=60 #}
+        {%- set preselect = e.sugestie_principala.cod_prestatie if e.sugestie_principala else (top.cod_prestatie if (top and top.score >= 60) else '') -%}
         <div class="maprow" style="align-items:flex-end; margin-bottom:10px;">
           <input type="hidden" name="cod_op_service" value="{{ e.cod_op_service }}">
           <div class="mapcol grow">
@@ -94,9 +95,20 @@
             {% if e.denumire and e.denumire != e.cod_op_service %}
             <div class="muted">{{ e.denumire }}</div>
             {% endif %}
-            {% if e.suggestions %}
+            {% if e.suggestions or e.sugestie_principala or (e.surse_sugestie and e.surse_sugestie.nul) %}
             <div class="muted" style="font-size:12px; margin-top:4px;">
               sugestii:
+              {# 5.18 US-007: badge sursa pe sugestia sistemului — confirmat (GOLD) / similar
+                 (SILVER+embedding k-NN) / non-operatie (pre-filtru NUL). Suggestion-only. #}
+              {% if e.sugestie_principala %}
+                {% if e.sugestie_principala.sursa == 'gold_partajat' %}
+                <span class="sugg-sursa sugg-sursa--confirmat" title="cod confirmat de un operator">confirmat</span>
+                {% else %}
+                <span class="sugg-sursa sugg-sursa--similar" title="operatie similara deja vazuta (k-NN/exact)">similar</span>
+                {% endif %}
+              {% elif e.surse_sugestie and e.surse_sugestie.nul %}
+                <span class="sugg-sursa sugg-sursa--nul" title="pare non-operatie (ITP/plata/discount...)">non-operatie</span>
+              {% endif %}
               {% for s in e.suggestions[:3] %}
                 <span class="sugg">{{ s.cod_prestatie }} ({{ s.score|round|int }}%)</span>{% if not loop.last %}, {% endif %}
               {% endfor %}

tests/test_web_preview_paritate_mapari.py

@@ -0,0 +1,221 @@
+"""Paritate editor mapare: panoul inline din preview-ul de import == pagina /mapari.
+
+Panoul "Operatii de mapat la cod RAR" din preview (`_collect_unmapped_ops` via
+`_web_compute_preview`) trebuie sa produca ACEEASI `sugestie_principala` +
+`surse_sugestie` ca `pending_unmapped` (functia care randeaza /mapari), pentru
+aceeasi denumire de operatie. Fara asta, cele doua editoare diverg (preview arata
+doar fuzzy, /mapari arata GOLD partajat > SILVER > embeddings k-NN + badge sursa).
+
+Test TARE (Eng finding F6): NU reproba doar determinismul lui enrich (a chema enrich
+de doua ori), ci probeaza WIRING-ul real — `conn` pasat din `_web_compute_preview`,
+corpusul indexat o data, campurile atasate — construind un batch de import cu randuri
+needs_mapping si comparand rezultatul cu `pending_unmapped`, cate un caz per sursa
+(gold / silver / embedding / nul).
+
+Suggestion-only (#13): enrichment NU intra in resolve_prestatii/load_mapping.
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+import json
+import os
+import re
+import tempfile
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+# --------------------------------------------------------------------------- #
+# Cazuri: cate o operatie per sursa de sugestie.                                #
+# (op, denumire, cod_asteptat, sursa_asteptata)                                 #
+# --------------------------------------------------------------------------- #
+_CAZ_GOLD = ("OP-GOLD", "Revizie gold speciala", "OE-1", "gold_partajat")
+_CAZ_SILVER = ("OP-SILVER", "Reparatie motor silver", "OE-2", "silver")
+_CAZ_EMB = ("OP-EMB", "Diagnoza semantica embedding", "OE-3", "embedding")
+_CAZ_NUL = ("OP-ITP", "ITP CT 99 XYZ", None, None)  # pre-filtru NUL -> fara cod
+
+_CAZURI = [_CAZ_GOLD, _CAZ_SILVER, _CAZ_EMB, _CAZ_NUL]
+
+
+@pytest.fixture()
+def env(monkeypatch):
+    tmp = tempfile.mkdtemp()
+    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "paritate_preview.db"))
+    monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "false")
+    # Embeddings ON ca sursa "embedding" sa fie exercitata; modulul e mock-uit mai jos
+    # (fara lazy-load al modelului ~230MB).
+    monkeypatch.setenv("AUTOPASS_EMBEDDINGS_ENABLED", "true")
+    from app.config import get_settings
+    get_settings.cache_clear()
+    from app.crypto import reset_cache
+    reset_cache()
+    from app.db import init_db
+    init_db()
+    yield monkeypatch
+    get_settings.cache_clear()
+    reset_cache()
+
+
+@pytest.fixture()
+def mock_emb(monkeypatch):
+    """Mock modulul embeddings: has_corpus True, suggest_nearest da OE-3 doar pt textul
+    care contine EMBEDDING, index_corpus/corpus_signature inofensive (fara model real)."""
+    import app.embeddings as emb
+
+    def _suggest(text, top_k=1):
+        # text = denumire NORMALIZATA (upper, fara diacritice) — enrich normalizeaza.
+        if "EMBEDDING" in (text or ""):
+            return [{"cod": "OE-3", "is_nul": False, "similaritate": 0.99}]
+        return []
+
+    monkeypatch.setattr(emb, "has_corpus", lambda: True)
+    monkeypatch.setattr(emb, "suggest_nearest", _suggest)
+    monkeypatch.setattr(emb, "corpus_signature", lambda: "")
+    monkeypatch.setattr(emb, "index_corpus", lambda items, signature=None: None)
+    return emb
+
+
+@pytest.fixture()
+def client(env):
+    from app.main import app
+    with TestClient(app) as c:
+        yield c
+
+
+def _csv_bytes(rows: list[dict]) -> bytes:
+    buf = io.StringIO()
+    writer = csv.DictWriter(buf, fieldnames=list(rows[0].keys()), delimiter=";")
+    writer.writeheader()
+    writer.writerows(rows)
+    return buf.getvalue().encode("utf-8")
+
+
+def _seed_surse(conn):
+    """Semeaza nomenclator + GOLD + SILVER pentru cazurile de test."""
+    from app.shared_store import record_human_validation, seed_suggestions
+
+    conn.executemany(
+        "INSERT OR IGNORE INTO nomenclator_rar (cod_prestatie, nume_prestatie) VALUES (?, ?)",
+        [("OE-1", "REVIZIE"), ("OE-2", "REPARATIE MOTOR"), ("OE-3", "DIAGNOZA")],
+    )
+    # GOLD partajat (shared_mappings) — NU intra in operations_mapping, deci op ramane needs_mapping.
+    record_human_validation(conn, _CAZ_GOLD[1], _CAZ_GOLD[2])
+    # SILVER (mapping_suggestions).
+    seed_suggestions(conn, [
+        {"denumire": _CAZ_SILVER[1], "cod_prestatie": _CAZ_SILVER[2], "source": "llm", "confidence": 0.9},
+    ])
+    conn.commit()
+
+
+def _insert_submissions_needs_mapping(conn):
+    """Insereaza cate un submission needs_mapping per caz, cu ACEEASI (op, denumire)
+    ca randurile de import — ca `pending_unmapped` sa vada aceleasi operatii."""
+    for i, (op, den, _cod, _sursa) in enumerate(_CAZURI):
+        conn.execute(
+            "INSERT INTO submissions (account_id, status, payload_json, idempotency_key) "
+            "VALUES (1, 'needs_mapping', ?, ?)",
+            (
+                json.dumps({
+                    "vin": f"WVWZZZ1KZAW00{i:04d}",
+                    "prestatii": [{"cod_op_service": op, "denumire": den}],
+                }),
+                f"paritate-sub-{i}",
+            ),
+        )
+    conn.commit()
+
+
+def _upload_batch(client: TestClient) -> int:
+    """Upload CSV cu cele 4 operatii nemapate + salveaza maparea de coloane. -> import_id."""
+    rows = [
+        {
+            "VIN": f"WVWZZZ1KZAW01{i:04d}",
+            "Nr": f"B{i:03d}TST",
+            "Data": "2026-06-15",
+            "KM": str(100000 + i),
+            "Operatie": op,
+            "Denumire": den,
+        }
+        for i, (op, den, _c, _s) in enumerate(_CAZURI)
+    ]
+    data = _csv_bytes(rows)
+    r = client.post("/_import/upload", files={"file": ("t.csv", io.BytesIO(data), "text/csv")})
+    assert r.status_code == 200, r.text
+    m = re.search(r"/_import/(\d+)/", r.text)
+    assert m, r.text[:400]
+    iid = int(m.group(1))
+    if f"/_import/{iid}/mapare-coloane" in r.text:
+        r2 = client.post(
+            f"/_import/{iid}/mapare-coloane",
+            data={
+                "colname": ["VIN", "Nr", "Data", "KM", "Operatie", "Denumire"],
+                "canon": ["vin", "nr_inmatriculare", "data_prestatie", "odometru_final",
+                          "operatie", "denumire_op"],
+                "format_data": "YYYY-MM-DD",
+            },
+        )
+        assert r2.status_code == 200, r2.text
+    return iid
+
+
+def test_preview_unmapped_ops_paritate_cu_pending_unmapped(client, mock_emb):
+    """`_web_compute_preview(...)["unmapped_ops"]` == `pending_unmapped(conn, account)`
+    pe `sugestie_principala` + `surse_sugestie`, per sursa (gold/silver/embedding/nul)."""
+    from app.db import get_connection
+    from app.mapping import pending_unmapped
+    from app.web.routes import _web_compute_preview
+
+    conn = get_connection()
+    try:
+        _seed_surse(conn)
+        iid = _upload_batch(client)
+        _insert_submissions_needs_mapping(conn)
+
+        # Calea PREVIEW (panou inline import) — foloseste conn (wiring nou).
+        preview = _web_compute_preview(conn, iid, 1)
+        assert isinstance(preview, dict), preview
+        preview_ops = {e["cod_op_service"]: e for e in preview["unmapped_ops"]}
+
+        # Calea /mapari (functia canonica de randare a editorului).
+        pending_ops = {e["cod_op_service"]: e for e in pending_unmapped(conn, 1)}
+
+        # Ambele cai trebuie sa vada exact aceleasi operatii.
+        assert set(preview_ops) == set(pending_ops) == {c[0] for c in _CAZURI}
+
+        # Paritate 1:1 pe sugestia principala + sursele, per operatie.
+        for op in preview_ops:
+            assert preview_ops[op]["sugestie_principala"] == pending_ops[op]["sugestie_principala"], op
+            assert preview_ops[op]["surse_sugestie"] == pending_ops[op]["surse_sugestie"], op
+
+        # Corectitudine per sursa (nu doar egalitate reciproca): fiecare caz da ce trebuie.
+        for op, _den, cod, sursa in _CAZURI:
+            sp = preview_ops[op]["sugestie_principala"]
+            surse = preview_ops[op]["surse_sugestie"]
+            if sursa is None:
+                # NUL: fara cod, badge non-operatie.
+                assert sp is None, op
+                assert surse["nul"] is True, op
+            else:
+                assert sp == {"cod_prestatie": cod, "sursa": sursa}, op
+                assert surse[sursa] == cod, op
+    finally:
+        conn.close()
+
+
+def test_collect_unmapped_ops_conn_none_contract_template(env):
+    """Fara conn, `_collect_unmapped_ops` init-eaza totusi `sugestie_principala`=None +
+    `surse_sugestie` default -> contractul catre template ramane identic (fara KeyError)."""
+    from app.web.routes import _collect_unmapped_ops
+
+    preview_rows = [{
+        "resolved_status": "needs_mapping",
+        "resolved": {"prestatii": [{"cod_op_service": "OP-X", "denumire": "Ceva"}]},
+    }]
+    out = _collect_unmapped_ops(preview_rows, [], conn=None)
+    assert len(out) == 1
+    e = out[0]
+    assert e["sugestie_principala"] is None
+    assert e["surse_sugestie"] == {"gold_partajat": None, "silver": None, "embedding": None, "nul": False}