diff --git a/bot/app/api/client.py b/bot/app/api/client.py index 674883f..29e2e8f 100644 --- a/bot/app/api/client.py +++ b/bot/app/api/client.py @@ -67,8 +67,8 @@ class ErpClient: # ---------------------------------------------------------------- lookup - async def list_suppliers(self, q: str | None = None, page_size: int = 20) -> list[dict]: - params: dict = {"page_size": page_size} + async def list_suppliers(self, q: str | None = None, page_size: int = 20, page: int = 1) -> list[dict]: + params: dict = {"page_size": page_size, "page": page} if q: params["q"] = q return (await self._request("GET", "/suppliers", params=params)).get("data") or [] @@ -76,8 +76,8 @@ class ErpClient: async def create_supplier(self, payload: dict, actor_id: str | None = None) -> dict: return (await self._request("POST", "/suppliers", json=payload, actor_id=actor_id)).get("data") or {} - async def list_materials(self, q: str | None = None, page_size: int = 20) -> list[dict]: - params: dict = {"page_size": page_size} + async def list_materials(self, q: str | None = None, page_size: int = 20, page: int = 1) -> list[dict]: + params: dict = {"page_size": page_size, "page": page} if q: params["q"] = q return (await self._request("GET", "/materials", params=params)).get("data") or [] @@ -100,6 +100,9 @@ class ErpClient: async def list_categories(self, page_size: int = 100) -> list[dict]: return (await self._request("GET", "/material-categories", params={"page_size": page_size})).get("data") or [] + async def list_brands(self, page: int = 1, page_size: int = 100) -> list[dict]: + return (await self._request("GET", "/brands", params={"page": page, "page_size": page_size})).get("data") or [] + # ---------------------------------------------------------------- quotes async def list_quotes(self, *, status: str | None = None, q: str | None = None, diff --git a/bot/app/config.py b/bot/app/config.py index 5d592c1..a7f5406 100644 --- a/bot/app/config.py +++ b/bot/app/config.py @@ -139,6 +139,17 @@ class Settings(BaseSettings): OBS_VOICE_DIR: str = "data/voices" # persistent volume (./bot/data) — keep forever OBS_PROMPT_VERSION: str = "" # manual label for the NLU prompt in use + # Jev (TypeSafe) entity matching — phases 2-5. Model version is PINNED + # (phase 0 resolved jev-1.13.0) so upstream updates cannot silently change + # decisions; every response's actual model is recorded for drift detection. + TYPESAFE_API_KEY: str = "" + JEV_MODEL: str = "jev-1.13.0" + JEV_ENABLED: bool = True + JEV_TIMEOUT: float = 8.0 + JEV_CONF_EXACT: float = 0.85 + JEV_CONF_FLOOR: float = 0.60 + JEV_MAX_QUESTIONS_PER_FIELD: int = 15 + LOG_LEVEL: str = "INFO" @property diff --git a/bot/app/handlers/__init__.py b/bot/app/handlers/__init__.py index 878e23d..c8a7196 100644 --- a/bot/app/handlers/__init__.py +++ b/bot/app/handlers/__init__.py @@ -224,7 +224,20 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None: result = await understand(settings, input_payload, draft_summary, trace_id=trace_id) intent, patch = result.get("intent"), result.get("patch") or {} consensus_text = result.get("consensus_text") or text - await obs.event(trace_id, "entities.extracted", payload={"patch": patch, "intent": intent}) + mentions = result.get("entity_mentions") or [] + await obs.event(trace_id, "entities.extracted", + payload={"patch": patch, "intent": intent, "entity_mentions": mentions}) + + # Phase 3: Jev per-field matching — only when it can be applied (flag-gated; + # a failure leaves jev_map empty and the difflib path runs unchanged). + jev_map: dict = {} + if settings.JEV_ENABLED and mentions and intent in ("register_quote", "update_draft", "add_material"): + try: + from app import matching + jev_map = await matching.resolve_mentions(ctx, patch, mentions, trace_id) + except Exception: + logger.exception("jev matching failed") + jev_map = {} if source == "voice": await ctx._log(chat_id, "in", "voice", consensus_text) @@ -238,6 +251,7 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None: incoming_msg_id=message_id, target_draft=target_draft, trace_id=trace_id, + jev_map=jev_map, ) elif intent == "commit_draft": active = target_draft or await draft_store.get_active(chat_id) @@ -280,7 +294,7 @@ async def handle_message(ctx: BotContext, chat_id: int, message: dict) -> None: elif intent == "add_supplier": await supplier.handle_entities(ctx, chat_id, user, patch, source, trace_id=trace_id) elif intent == "add_material": - await material.handle_entities(ctx, chat_id, user, patch, source, trace_id=trace_id) + await material.handle_entities(ctx, chat_id, user, patch, source, trace_id=trace_id, jev_map=jev_map) elif intent == "add_project": await project.handle_entities(ctx, chat_id, user, patch, source, trace_id=trace_id) elif intent == "status_query": diff --git a/bot/app/handlers/callbacks.py b/bot/app/handlers/callbacks.py index 5622d8c..af45bf3 100644 --- a/bot/app/handlers/callbacks.py +++ b/bot/app/handlers/callbacks.py @@ -82,6 +82,8 @@ async def _dispatch(ctx, chat_id: int, data: str, cq: dict) -> None: await obs.event(tid, "ambiguity.user_choice", payload={ "amb_id": amb_id, "action": action, "kind": target_amb.get("kind"), "title": target_amb.get("title"), + "arg": parts[4] if len(parts) > 4 else None, + "candidates": target_amb.get("candidates"), }) if action == "ok": @@ -107,6 +109,38 @@ async def _dispatch(ctx, chat_id: int, data: str, cq: dict) -> None: await ctx.delete(chat_id, target_amb["message_id"]) except Exception: pass + elif action == "pick" and len(parts) >= 5: + # Phase 4: user picked one of the Jev candidates. + try: + idx = int(parts[4]) + except ValueError: + return + cands = target_amb.get("candidates") or [] + if not (0 <= idx < len(cands)): + return + from app import matching + matching.apply_choice(payload, target_amb.get("field"), cands[idx], + target_amb.get("variants") or []) + target_amb["status"] = "applied" + await quote.resolve_payload(ctx, payload) + await draft_store.save_payload(draft, mark_bale_synced=True) + # In-place edit main review card + await quote.render(ctx, chat_id, draft, fresh=False) + # Delete ONLY this ambiguity message + if target_amb.get("message_id"): + try: + await ctx.delete(chat_id, target_amb["message_id"]) + except Exception: + pass + elif action == "new": + # «هیچ‌کدام» — keep the raw name; the auto-create path handles it at commit. + target_amb["status"] = "rejected" + await draft_store.save_payload(draft, mark_bale_synced=True) + if target_amb.get("message_id"): + try: + await ctx.delete(chat_id, target_amb["message_id"]) + except Exception: + pass return # ---- pick a candidate for a draft field ------------------------------- diff --git a/bot/app/handlers/material.py b/bot/app/handlers/material.py index 8311ce8..4413584 100644 --- a/bot/app/handlers/material.py +++ b/bot/app/handlers/material.py @@ -2,6 +2,7 @@ import logging from app import drafts as draft_store +from app import matching from app import observability as obs from app.api.client import ApiError from app.drafts import ( @@ -15,7 +16,8 @@ logger = logging.getLogger("bot.material") async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None, - trace_id: str | None = None) -> None: + trace_id: str | None = None, + jev_map: dict | None = None) -> None: draft = await draft_store.get_active(chat_id, KIND_MATERIAL) if draft is not None: payload = draft.payload @@ -25,25 +27,36 @@ async def handle_entities(ctx, chat_id, user, patch: dict, source: str | None, merge_material_patch(payload, patch) draft = await draft_store.create(chat_id, KIND_MATERIAL, payload, source=source) await obs.link_draft(trace_id, draft.id, draft.kind) - await resolve_payload(ctx, payload, trace_id) + await resolve_payload(ctx, payload, trace_id, jev_map) await draft_store.save_payload(draft, missing={"fields": material_missing(payload)}) await render(ctx, chat_id, draft, fresh=draft.review_message_id is None, trace_id=trace_id) -async def resolve_payload(ctx, payload: dict, trace_id: str | None = None) -> None: +async def resolve_payload(ctx, payload: dict, trace_id: str | None = None, + jev_map: dict | None = None) -> None: options = payload.setdefault("options", {}) if not payload.get("canonical_unit_id") and payload.get("unit_name"): - res = await safe_resolve(ctx.resolver, "unit", payload["unit_name"]) - await obs.log_resolution(trace_id, "unit", payload["unit_name"], res.status, - [{"id": c.id, "name": c.name} for c in (res.candidates or [])[:4]]) - if res.resolved: - payload["canonical_unit_id"] = res.candidates[0].id - payload["unit_name"] = res.candidates[0].name - options.pop("unit", None) - elif res.candidates: - options["unit"] = [{"id": c.id, "name": c.name} for c in res.candidates] + hit = matching.lookup(jev_map, payload["unit_name"]) + if hit is not None: + if hit["status"] == "exact": + payload["canonical_unit_id"] = hit["id"] + payload["unit_name"] = hit["name"] + options.pop("unit", None) + elif hit["status"] == "ambiguity" and hit.get("candidates"): + options["unit"] = [{"id": c["id"], "name": c["name"]} + for c in hit["candidates"][:6]] else: - options.pop("unit", None) + res = await safe_resolve(ctx.resolver, "unit", payload["unit_name"]) + await obs.log_resolution(trace_id, "unit", payload["unit_name"], res.status, + [{"id": c.id, "name": c.name} for c in (res.candidates or [])[:4]]) + if res.resolved: + payload["canonical_unit_id"] = res.candidates[0].id + payload["unit_name"] = res.candidates[0].name + options.pop("unit", None) + elif res.candidates: + options["unit"] = [{"id": c.id, "name": c.name} for c in res.candidates] + else: + options.pop("unit", None) async def apply_pick(ctx, chat_id, user, draft, field: str, choice: dict) -> None: @@ -105,5 +118,5 @@ async def commit(ctx, chat_id, user, draft, trace_id: str | None = None) -> None await obs.event(trace_id, "draft.commit", payload={"kind": "material", "material_id": (created or {}).get("id")}) await obs.set_trace_status(trace_id, "committed") - ctx.resolver.invalidate("materials") + ctx.resolver.invalidate() await ctx.send(chat_id, material_committed_report(created)) diff --git a/bot/app/handlers/project.py b/bot/app/handlers/project.py index c5cb3de..2ccaf3a 100644 --- a/bot/app/handlers/project.py +++ b/bot/app/handlers/project.py @@ -72,5 +72,5 @@ async def commit(ctx, chat_id, user, draft, trace_id: str | None = None) -> None await obs.event(trace_id, "draft.commit", payload={"kind": "project", "project_id": (created or {}).get("id")}) await obs.set_trace_status(trace_id, "committed") - ctx.resolver.invalidate("projects") + ctx.resolver.invalidate() await ctx.send(chat_id, project_committed_report(created)) diff --git a/bot/app/handlers/quote.py b/bot/app/handlers/quote.py index 79d104b..9f5777b 100644 --- a/bot/app/handlers/quote.py +++ b/bot/app/handlers/quote.py @@ -2,6 +2,7 @@ import logging from app import drafts as draft_store +from app import matching from app import observability as obs from app.api.client import ApiError from app.drafts import ( @@ -22,6 +23,7 @@ async def handle_entities( incoming_msg_id: int | None = None, target_draft = None, trace_id: str | None = None, + jev_map: dict | None = None, ) -> None: prev_review_msg_id = None if target_draft is not None: @@ -50,7 +52,10 @@ async def handle_entities( ) await obs.link_draft(trace_id, draft.id, draft.kind) - await resolve_payload(ctx, payload, trace_id) + await resolve_payload(ctx, payload, trace_id, jev_map) + if jev_map: + matching.normalize_brand_spec(payload, jev_map) + matching.create_ambiguities(payload, jev_map) if prev_review_msg_id: try: @@ -71,46 +76,65 @@ async def _log_res(trace_id, field: str, mention, res) -> None: [{"id": c.id, "name": c.name} for c in (res.candidates or [])[:4]]) -async def resolve_payload(ctx, payload: dict, trace_id: str | None = None) -> None: +async def resolve_payload(ctx, payload: dict, trace_id: str | None = None, + jev_map: dict | None = None) -> None: options = payload.setdefault("options", {}) if not payload.get("supplier_id") and payload.get("supplier_name"): - res = await safe_resolve(ctx.resolver, "supplier", payload["supplier_name"]) - await _log_res(trace_id, "supplier", payload["supplier_name"], res) - if res.resolved: - c = res.candidates[0] - payload["supplier_id"] = c.id - payload["supplier_name"] = c.name - extra = c.extra or {} - for k in ("location", "phone", "contact_person", "category_name"): - if extra.get(k): - payload[f"supplier_{k}"] = extra[k] - options.pop("sup", None) - elif res.candidates: - options["sup"] = [{"id": c.id, "name": c.name} for c in res.candidates] + hit = matching.lookup(jev_map, payload["supplier_name"]) + if hit is not None: + if hit["status"] == "exact": + matching.apply_exact(payload, hit) + options.pop("sup", None) else: - options.pop("sup", None) + res = await safe_resolve(ctx.resolver, "supplier", payload["supplier_name"]) + await _log_res(trace_id, "supplier", payload["supplier_name"], res) + if res.resolved: + c = res.candidates[0] + payload["supplier_id"] = c.id + payload["supplier_name"] = c.name + extra = c.extra or {} + for k in ("location", "phone", "contact_person", "category_name"): + if extra.get(k): + payload[f"supplier_{k}"] = extra[k] + options.pop("sup", None) + elif res.candidates: + options["sup"] = [{"id": c.id, "name": c.name} for c in res.candidates] + else: + options.pop("sup", None) resolved_projects = payload.setdefault("resolved_projects", {}) if not payload.get("project_id") and payload.get("project_name"): - res = await safe_resolve(ctx.resolver, "project", payload["project_name"]) - await _log_res(trace_id, "project", payload["project_name"], res) - if res.resolved: - c = res.candidates[0] - payload["project_id"] = c.id - payload["project_name"] = c.name - resolved_projects[c.name] = c.id - resolved_projects[payload["project_name"]] = c.id - if payload["project_id"] not in payload.get("project_ids", []): - payload.setdefault("project_ids", []).append(payload["project_id"]) - options.pop("prj", None) - elif res.candidates: - options["prj"] = [{"id": c.id, "name": c.name} for c in res.candidates] + hit = matching.lookup(jev_map, payload["project_name"]) + if hit is not None: + if hit["status"] == "exact": + matching.apply_exact(payload, hit) + payload["project_name"] = hit["name"] + options.pop("prj", None) else: - options.pop("prj", None) + res = await safe_resolve(ctx.resolver, "project", payload["project_name"]) + await _log_res(trace_id, "project", payload["project_name"], res) + if res.resolved: + c = res.candidates[0] + payload["project_id"] = c.id + payload["project_name"] = c.name + resolved_projects[c.name] = c.id + resolved_projects[payload["project_name"]] = c.id + if payload["project_id"] not in payload.get("project_ids", []): + payload.setdefault("project_ids", []).append(payload["project_id"]) + options.pop("prj", None) + elif res.candidates: + options["prj"] = [{"id": c.id, "name": c.name} for c in res.candidates] + else: + options.pop("prj", None) for p_name in payload.get("project_names") or []: + hit = matching.lookup(jev_map, p_name) + if hit is not None: + if hit["status"] == "exact": + matching.apply_exact(payload, hit) + continue res = await safe_resolve(ctx.resolver, "project", p_name) await _log_res(trace_id, "project", p_name, res) if res.resolved: @@ -125,17 +149,23 @@ async def resolve_payload(ctx, payload: dict, trace_id: str | None = None) -> No for i, item in enumerate(payload.get("items") or [], start=1): if not item.get("material_id") and item.get("material_name"): - res = await safe_resolve(ctx.resolver, "material", item["material_name"]) - await _log_res(trace_id, "material", item["material_name"], res) - if res.resolved: - _set_material(item, res.candidates[0].id, res.candidates[0].name, res.candidates[0].extra) - options.pop(f"mat{i}", None) - elif res.candidates: - options[f"mat{i}"] = [{"id": c.id, "name": c.name} for c in res.candidates] + hit = matching.lookup(jev_map, item["material_name"]) + if hit is not None: + if hit["status"] == "exact": + matching.apply_exact(payload, hit) + options.pop(f"mat{i}", None) else: - options.pop(f"mat{i}", None) + res = await safe_resolve(ctx.resolver, "material", item["material_name"]) + await _log_res(trace_id, "material", item["material_name"], res) + if res.resolved: + _set_material(item, res.candidates[0].id, res.candidates[0].name, res.candidates[0].extra) + options.pop(f"mat{i}", None) + elif res.candidates: + options[f"mat{i}"] = [{"id": c.id, "name": c.name} for c in res.candidates] + else: + options.pop(f"mat{i}", None) if not item.get("unit_id"): - await _resolve_unit(ctx, item, trace_id) + await _resolve_unit(ctx, item, trace_id, jev_map) def _set_material(item: dict, material_id: str, name: str, extra: dict | None = None) -> None: @@ -143,17 +173,26 @@ def _set_material(item: dict, material_id: str, name: str, extra: dict | None = item["material_row"] = {"id": material_id, "canonical_unit_id": (extra or {}).get("canonical_unit_id")} -async def _resolve_unit(ctx, item: dict, trace_id: str | None = None) -> None: +async def _resolve_unit(ctx, item: dict, trace_id: str | None = None, + jev_map: dict | None = None) -> None: if item.get("unit_id"): return if item.get("unit_name"): - res = await safe_resolve(ctx.resolver, "unit", item["unit_name"]) - await _log_res(trace_id, "unit", item["unit_name"], res) - if res.resolved: - item["unit_id"] = res.candidates[0].id - item["unit_explicit"] = True - item["unit_name"] = res.candidates[0].name - return + hit = matching.lookup(jev_map, item["unit_name"]) + if hit is not None: + if hit["status"] == "exact": + item["unit_id"] = hit["id"] + item["unit_explicit"] = True + item["unit_name"] = hit["name"] + return + else: + res = await safe_resolve(ctx.resolver, "unit", item["unit_name"]) + await _log_res(trace_id, "unit", item["unit_name"], res) + if res.resolved: + item["unit_id"] = res.candidates[0].id + item["unit_explicit"] = True + item["unit_name"] = res.candidates[0].name + return canonical = ((item.get("material_row") or {}).get("canonical_unit_id")) if canonical: item["unit_id"] = str(canonical) diff --git a/bot/app/handlers/supplier.py b/bot/app/handlers/supplier.py index 4073ed6..f69eda8 100644 --- a/bot/app/handlers/supplier.py +++ b/bot/app/handlers/supplier.py @@ -110,5 +110,5 @@ async def commit(ctx, chat_id, user, draft, trace_id: str | None = None) -> None await obs.event(trace_id, "draft.commit", payload={"kind": "supplier", "supplier_id": (created or {}).get("id")}) await obs.set_trace_status(trace_id, "committed") - ctx.resolver.invalidate("suppliers") + ctx.resolver.invalidate() await ctx.send(chat_id, supplier_committed_report(created)) diff --git a/bot/app/jev.py b/bot/app/jev.py new file mode 100644 index 0000000..4787169 --- /dev/null +++ b/bot/app/jev.py @@ -0,0 +1,67 @@ +"""Jev (TypeSafe) client — structured `choice` matching over HTTP (phases 2-5). + +Contract (docs.typesafe.ai): POST /v1/systemone with + {"state": ..., "model": , "questions": {key: {type, instructions, criteria}}} +→ {"answers": {key: {choice, confidence, probabilities}}, "usage": {...}, "model": "..."} + +Design rules (final plan v5): +- The model version is PINNED (JEV_MODEL=jev-1.13.0 from phase 0); a response + whose `model` differs logs a drift warning and is still returned (recorded). +- Fail-soft: any error/timeout → None. Callers fall back per field to the + existing difflib path; with JEV_ENABLED=false nothing here is called at all. +- One retry only for 429/5xx (after a short backoff), then give up. +""" +import asyncio +import logging +import time + +import httpx + +from app.config import settings + +logger = logging.getLogger("bot.jev") + +_URL = "https://api.typesafe.ai/v1/systemone" + + +async def ask(state: dict, questions: dict, *, timeout: float | None = None) -> dict | None: + """One systemone request. Returns {"answers", "usage", "model", "latency_ms"} or None.""" + if not settings.TYPESAFE_API_KEY: + logger.warning("jev: TYPESAFE_API_KEY is not configured") + return None + + payload = {"state": state, "model": settings.JEV_MODEL, "questions": questions} + headers = { + "Authorization": f"Bearer {settings.TYPESAFE_API_KEY}", + "Content-Type": "application/json", + } + t0 = time.perf_counter() + for attempt in (1, 2): + try: + async with httpx.AsyncClient(timeout=timeout or settings.JEV_TIMEOUT) as client: + resp = await client.post(_URL, json=payload, headers=headers) + if resp.status_code == 200: + body = resp.json() or {} + actual = str(body.get("model") or "") + if actual and actual != settings.JEV_MODEL: + logger.warning("jev: model drift — got %s, pinned %s", actual, settings.JEV_MODEL) + return { + "answers": body.get("answers") or {}, + "usage": body.get("usage") or {}, + "model": actual or settings.JEV_MODEL, + "latency_ms": int((time.perf_counter() - t0) * 1000), + } + if resp.status_code in (429, 500, 502, 503, 504) and attempt == 1: + logger.warning("jev: HTTP %s — retrying once", resp.status_code) + await asyncio.sleep(1.0) + continue + logger.error("jev: HTTP %s: %s", resp.status_code, resp.text[:300]) + return None + except Exception as exc: + if attempt == 1: + logger.warning("jev: request failed (%s) — retrying once", exc) + await asyncio.sleep(1.0) + continue + logger.error("jev: request failed: %s", exc) + return None + return None diff --git a/bot/app/keyboards.py b/bot/app/keyboards.py index 3182668..0024310 100644 --- a/bot/app/keyboards.py +++ b/bot/app/keyboards.py @@ -81,3 +81,13 @@ def ambiguity_keyboard(prefix: str, amb_id: str) -> dict: [_btn("❌ تایید نمی‌شود", f"amb:no:{prefix}:{amb_id}")], ]) + +def ambiguity_candidates_keyboard(prefix: str, amb_id: str, candidates: list[dict]) -> dict: + """Phase 4: one button per Jev candidate (+ 'new') for an entity-match ambiguity.""" + rows = [ + [_btn(f"✅ {str(c.get('name') or '')[:40]}", f"amb:pick:{prefix}:{amb_id}:{i}")] + for i, c in enumerate(candidates[:5]) + ] + rows.append([_btn("🆕 هیچ‌کدام — ایجاد مورد جدید", f"amb:new:{prefix}:{amb_id}")]) + return _kb(rows) + diff --git a/bot/app/matching.py b/bot/app/matching.py new file mode 100644 index 0000000..35d72e2 --- /dev/null +++ b/bot/app/matching.py @@ -0,0 +1,321 @@ +"""Entity matching with Jev (TypeSafe) — phases 3-5 of the final plan. + +Mechanics (final plan v5 §2): +- برون‌گروهی: ONE Jev request per FIELD that has mentions (requests run in + parallel; latency ≈ one round-trip regardless of field count). +- درون‌گروهی: inside a field request, ONE `choice` question per mention; every + question rides with the SAME full candidate list (+ a `_none` option). +- درون‌واریانسی: all STT spellings of one entity travel in ONE question, and + the instructions ask the model to treat them as one name. + +Routing (thresholds from env; calibrated in phase 6): + conf >= JEV_CONF_EXACT → exact (bind id, no ambiguity sub-message) + JEV_CONF_FLOOR..JEV_CONF_EXACT → ambiguity (candidate buttons sub-message) + below floor / _none → new (existing auto-create path) + +Fail-soft: a failed field simply yields no results for its mentions; callers +fall back to the difflib path (`safe_resolve`). With JEV_ENABLED=false nothing +in this module runs (zero regression). +""" +import asyncio +import hashlib +import json +import logging + +from app import observability as obs +from app.config import settings +from app.jev import ask as jev_ask +from app.nlu import normalize_text + +logger = logging.getLogger("bot.matching") + +FIELD_ORDER = ["project", "supplier", "material", "brand", "spec", "unit"] + +FIELD_FA = { + "project": "پروژه", + "supplier": "تأمین‌کننده", + "material": "متریال", + "brand": "برند", + "spec": "مشخصه", + "unit": "واحد", +} + + +# --------------------------------------------------------------------------- +# candidate lists (full lists — no prefilter, per owner decision) +# --------------------------------------------------------------------------- + +async def _candidates(resolver, field: str) -> list: + if field == "project": + return await resolver.projects() + if field == "supplier": + return await resolver.all_suppliers() + if field == "material": + return await resolver.all_materials() + if field == "brand": + return await resolver.brands() + if field == "spec": + return await resolver.specs() + if field == "unit": + return await resolver.units() + return [] + + +def _hint(field: str, cand) -> str: + """Short, token-cheap description shown next to a candidate name.""" + if field == "supplier": + extra = cand.extra or {} + bits = [b for b in (extra.get("category_name"), extra.get("location")) if b] + return "، ".join(str(b) for b in bits) if bits else "تأمین‌کننده" + return "" + + +def _instructions(field: str, variants: list[str]) -> str: + fa = FIELD_FA.get(field, field) + vlist = "»، «".join(v for v in variants if v) + if len(variants) > 1: + return ( + f"کاربر در پیام (صوتی) خود این {fa} را ذکر کرده و مدل‌های مختلف STT آن را به شکل‌های " + f"«{vlist}» پیاده کرده‌اند؛ این‌ها چند نگارش از یک نام واحدند. با توجه به بافت جمله و " + f"تشابه آوایی میان این واریانت‌ها، مشخص کن به کدام گزینه از فهرست اشاره دارد. " + f"اگر هیچ‌کدام منطبق نیست یا تردید جدی داری «_none» را انتخاب کن." + ) + return ( + f"کاربر «{vlist}» را به‌عنوان {fa} ذکر کرده است. این عبارت به کدام گزینه از فهرست اشاره دارد؟ " + f"اگر هیچ‌کدام منطبق نیست یا تردید جدی داری «_none» را انتخاب کن." + ) + + +# --------------------------------------------------------------------------- +# the per-field fan-out +# --------------------------------------------------------------------------- + +async def resolve_mentions(ctx, patch: dict, mentions: list[dict], trace_id) -> dict: + """Run Jev per field (parallel); return a map keyed by normalized variant. + + Each value: {"field", "status": exact|ambiguity|new", "variants", + "id"?, "name"?, "extra"?, "confidence"?, "candidates": [...]} + """ + by_field: dict[str, list[dict]] = {} + for m in mentions or []: + field = str((m or {}).get("field") or "").strip() + variants = [str(v).strip() for v in ((m or {}).get("variants") or []) if str(v or "").strip()] + if field in FIELD_ORDER and variants: + by_field.setdefault(field, []).append({"field": field, "variants": variants[:6]}) + + results: dict[str, dict] = {} + if not by_field: + return results + + async def one_field(field: str, ms: list[dict]) -> None: + try: + cands = await _candidates(ctx.resolver, field) + except Exception: + logger.exception("matching: candidate fetch failed for %s", field) + return + if not cands: + return + + by_name: dict[str, object] = {} + criteria: dict[str, str] = {} + for c in cands: + if c.name in criteria: + continue + criteria[c.name] = _hint(field, c) + by_name[c.name] = c + criteria["_none"] = "هیچ‌کدام از موارد موجود نیست" + + capped = ms[: settings.JEV_MAX_QUESTIONS_PER_FIELD] + questions = { + f"mention_{i}": { + "type": "choice", + "instructions": _instructions(field, m["variants"]), + "criteria": criteria, + } + for i, m in enumerate(capped) + } + state = { + "field": field, + "patch_context": json.dumps(patch, ensure_ascii=False, default=str)[:1500], + } + await obs.event(trace_id, "jev.request", field=field, payload={ + "n_mentions": len(capped), "n_candidates": len(cands), + }) + resp = await jev_ask(state, questions) + await obs.event(trace_id, "jev.result", field=field, ok=resp is not None, + model=(resp or {}).get("model"), + latency_ms=(resp or {}).get("latency_ms"), + payload={"answers": (resp or {}).get("answers"), + "usage": (resp or {}).get("usage")}) + if resp is None: + # Field-wide failure → leave no results so callers fall back to difflib. + return + answers = resp.get("answers") or {} + + decisions = [] + for i, m in enumerate(capped): + ans = answers.get(f"mention_{i}") or {} + if not ans: + # No answer for this mention → difflib fallback for it. + continue + choice = ans.get("choice") + conf = float(ans.get("confidence") or 0.0) + probs = ans.get("probabilities") or {} + cand = by_name.get(choice) if choice else None + ranked = sorted( + ({"id": by_name[n].id, "name": n, "prob": float(p or 0.0)} + for n, p in probs.items() if n != "_none" and n in by_name), + key=lambda x: -x["prob"], + )[:6] + + if cand is not None and conf >= settings.JEV_CONF_EXACT: + res = {"field": field, "status": "exact", "variants": m["variants"], + "id": cand.id, "name": cand.name, "extra": cand.extra or {}, + "confidence": conf} + elif cand is not None and conf >= settings.JEV_CONF_FLOOR: + res = {"field": field, "status": "ambiguity", "variants": m["variants"], + "id": cand.id, "name": cand.name, "extra": cand.extra or {}, + "confidence": conf, + "candidates": ranked or [{"id": cand.id, "name": cand.name, "prob": conf}]} + else: + res = {"field": field, "status": "new", "variants": m["variants"], + "confidence": conf, "candidates": ranked} + for v in m["variants"]: + results[normalize_text(v)] = res + decisions.append({"variants": m["variants"], "status": res["status"], + "choice": res.get("name"), "confidence": conf, + "probs_top": [{k: round(v, 3)} for k, v in + sorted(probs.items(), key=lambda kv: -kv[1])[:3]]}) + await obs.event(trace_id, "routing.decision", field=field, + payload={"source": "jev", "mentions": decisions}) + + await asyncio.gather(*(one_field(f, ms) for f, ms in by_field.items())) + return results + + +def lookup(jev_map: dict | None, name: str | None) -> dict | None: + """Find the Jev result for a raw name (by normalized variant).""" + if not jev_map or not name: + return None + return jev_map.get(normalize_text(name)) + + +# --------------------------------------------------------------------------- +# writing results into drafts +# --------------------------------------------------------------------------- + +def apply_choice(payload: dict, field: str, cand: dict, variants: list[str]) -> None: + """Bind one chosen candidate (exact match or a user pick) into the payload.""" + name, cid = cand.get("name"), cand.get("id") + extra = cand.get("extra") or {} + variants_norm = {normalize_text(v) for v in (variants or []) if v} + + if field == "supplier": + payload["supplier_id"] = cid + payload["supplier_name"] = name + for k in ("location", "phone", "contact_person", "category_name"): + if extra.get(k): + payload[f"supplier_{k}"] = extra[k] + elif field == "project": + resolved = payload.setdefault("resolved_projects", {}) + for v in (variants or []): + resolved[v] = cid + resolved[name] = cid + if cid not in payload.get("project_ids", []): + payload.setdefault("project_ids", []).append(cid) + if not payload.get("project_id"): + payload["project_id"] = cid + elif field == "material": + for item in payload.get("items") or []: + if normalize_text(item.get("material_name") or "") in variants_norm: + item["material_name"] = name + item["material_id"] = cid + item["material_row"] = {"id": cid, + "canonical_unit_id": extra.get("canonical_unit_id")} + elif field == "unit": + for item in payload.get("items") or []: + if normalize_text(item.get("unit_name") or "") in variants_norm: + item["unit_id"] = cid + item["unit_explicit"] = True + item["unit_name"] = name + if not payload.get("items"): + payload["canonical_unit_id"] = cid + payload["unit_name"] = name + elif field == "brand": + for item in payload.get("items") or []: + if normalize_text(item.get("brand_name") or "") in variants_norm: + item["brand_name"] = name + elif field == "spec": + for item in payload.get("items") or []: + specs = item.get("specs") or [] + for i, s in enumerate(specs): + if normalize_text(s) in variants_norm: + specs[i] = name + + +def apply_exact(payload: dict, res: dict) -> None: + """Bind one exact Jev match into a draft payload.""" + apply_choice(payload, res.get("field"), {"id": res.get("id"), "name": res.get("name"), + "extra": res.get("extra") or {}}, + res.get("variants") or []) + + +def normalize_brand_spec(payload: dict, jev_map: dict | None) -> None: + """Canonicalize brand/spec strings via exact matches (spec effect per owner: + spelling normalization only).""" + if not jev_map: + return + for item in payload.get("items") or []: + brand = item.get("brand_name") + if brand: + hit = lookup(jev_map, brand) + if hit and hit.get("status") == "exact": + item["brand_name"] = hit["name"] + specs = item.get("specs") or [] + for i, s in enumerate(specs): + hit = lookup(jev_map, s) + if hit and hit.get("status") == "exact": + specs[i] = hit["name"] + + +# --------------------------------------------------------------------------- +# phase 4 — ambiguity sub-messages with candidate buttons +# --------------------------------------------------------------------------- + +def _amb_id(field: str, variants: list[str]) -> str: + digest = hashlib.sha1("|".join(sorted(variants)).encode()).hexdigest()[:4] + return f"em_{field[:4]}_{digest}" + + +def _fa_num(n: int) -> str: + return str(n).translate(str.maketrans("0123456789", "۰۱۲۳۴۵۶۷۸۹")) + + +def create_ambiguities(payload: dict, jev_map: dict | None) -> None: + """One `entity_match` ambiguity entry per ambiguous mention (deduped).""" + if not jev_map: + return + ambs = payload.setdefault("ambiguities", []) + seen = {a.get("id") for a in ambs} + unique: dict[int, dict] = {} + for res in jev_map.values(): + unique[id(res)] = res # dedupe identical result objects + for res in unique.values(): + if res.get("status") != "ambiguity" or not res.get("candidates"): + continue + amb_id = _amb_id(res["field"], res.get("variants") or []) + if amb_id in seen: + continue + cands = res["candidates"][:6] + lines = [f"• {c['name']} — {_fa_num(int(round(c['prob'] * 100)))}٪" for c in cands] + ambs.append({ + "id": amb_id, + "kind": "entity_match", + "field": res["field"], + "variants": res.get("variants") or [], + "candidates": [{"id": c["id"], "name": c["name"]} for c in cands], + "title": f"تطبیق {FIELD_FA.get(res['field'], res['field'])}: «{' / '.join(res.get('variants') or [])}»", + "details": "\n".join(lines), + "status": "pending", + }) + seen.add(amb_id) diff --git a/bot/app/nlu.py b/bot/app/nlu.py index 9f4dea4..3e68fd9 100644 --- a/bot/app/nlu.py +++ b/bot/app/nlu.py @@ -25,6 +25,23 @@ def normalize_digits(text: str) -> str: ZWNJ = "\u200c" +MENTION_FIELDS = {"project", "supplier", "material", "brand", "spec", "unit"} + + +def sanitize_mentions(raw) -> list[dict]: + """Keep only valid {field, variants} mention objects from the LLM output.""" + out: list[dict] = [] + if not isinstance(raw, list): + return out + for m in raw: + if not isinstance(m, dict): + continue + field = str(m.get("field") or "").strip() + variants = [str(v).strip() for v in (m.get("variants") or []) if str(v or "").strip()] + if field in MENTION_FIELDS and variants: + out.append({"field": field, "variants": variants[:6]}) + return out + def normalize_text(text: str) -> str: """Normalize Persian text for fuzzy matching (ZWNJ, arabic ya/ka, spaces).""" @@ -60,6 +77,7 @@ SYSTEM_PROMPT = """تو دستیار هوشمند تدارکات یک سیستم "consensus_text": "بیانیه رسمی، پالایش‌شده، فشرده و بدون تکیه‌کلام از تقاضای تدارکات کاربر", "intent": یکی از {intents}, "patch": {شکل partial هر فعالیتی که کاربر داده}, + "entity_mentions": [{"field": "project|supplier|material|brand|spec|unit", "variants": ["همه‌ی نگارش‌ها"]}], "confidence": 0.0..1.0 } @@ -95,6 +113,15 @@ SYSTEM_PROMPT = """تو دستیار هوشمند تدارکات یک سیستم - اگر درباره پذیرش استعلام است → intent="accept_quote" و در patch.target نام تأمین‌کننده/استعلام. - اگر درباره اعلام رسیدن اقلام است → intent="arrive_items" و در patch.target نام تأمین‌کننده/استعلام و در patch.arrivals مقادیر. +۶. استخراج موجودیت‌های قابل تطبیق (entity_mentions): + - فیلدهای مجاز: "project" | "supplier" | "material" | "brand" | "spec" | "unit" + - هر نام کالا/تأمین‌کننده/پروژه/برند/واحد و هر مشخصه که در پیام آمده، یک عضو جداگانه است. + - اگر در نسخه‌های مختلف STT همان نام با املاهای متفاوت پیاده شده (مثل «مصطفوی»، «مستفدی»، «مسطفی»)، + همه را در variants همان یک عضو بگذار — نه چند عضو جداگانه. + - برای پیام متنی معمولی، variants یک عضو تک‌عنصر است. + - مشخصه (spec) و برند (brand) جدا از متریال و جدا از هم استخراج شوند. + - وظیفه‌ی تطبیق با موجودیت‌های سیستم بر عهده‌ی تو نیست؛ هیچ نامی را به نزدیک‌ترین نام موجود «حدس» نزن. + شکل patch برای intent="register_quote"/"update_draft": {"supplier_name": str, "project_name": str, "project_names": [str], "is_official": bool, "notes": str, "expiry_days": int, "expiry_date": "YYYY-MM-DD", "items": [{"material_name": str, "brand_name": str, "specs": [str], "quantity": number, "unit_name": str, "unit_price": number, "project_name": str, "notes": str}], @@ -148,7 +175,8 @@ def _extract_json(raw: str) -> dict | None: async def understand(settings: Settings, text: str | list[dict], draft_summary: str | None = None, trace_id: str | None = None) -> dict: """Return {"consensus_text": ..., "intent": ..., "patch": {...}, "confidence": ...}; never raises.""" - fallback = {"consensus_text": "", "intent": "unknown", "patch": {}, "confidence": 0.0} + fallback = {"consensus_text": "", "intent": "unknown", "patch": {}, + "entity_mentions": [], "confidence": 0.0} if not settings.AI_BASE_URL or not settings.AI_API_KEY: logger.warning("AI provider not configured; cannot understand message") return fallback @@ -214,6 +242,7 @@ async def understand(settings: Settings, text: str | list[dict], draft_summary: payload={"error": "unparseable", "raw": content[:500]}) return fallback parsed.setdefault("patch", {}) + parsed["entity_mentions"] = sanitize_mentions(parsed.get("entity_mentions")) if not parsed.get("consensus_text"): parsed["consensus_text"] = first_text await obs.event(trace_id, "nlu.result", ok=True, model=settings.LLM_MODEL, payload={ @@ -221,6 +250,7 @@ async def understand(settings: Settings, text: str | list[dict], draft_summary: "confidence": parsed.get("confidence"), "patch": parsed.get("patch"), "consensus_text": parsed.get("consensus_text"), + "entity_mentions": parsed.get("entity_mentions"), }) return parsed except (httpx.HTTPError, json.JSONDecodeError, KeyError, TypeError): diff --git a/bot/app/resolve.py b/bot/app/resolve.py index 1edadde..06b33d1 100644 --- a/bot/app/resolve.py +++ b/bot/app/resolve.py @@ -153,6 +153,43 @@ async def _db_materials(q: str | None = None) -> list[Candidate]: return [] +async def _db_brands() -> list[Candidate]: + try: + async with SessionLocal() as s: + rows = (await s.execute(text("SELECT id, name FROM brands"))).mappings().all() + return [Candidate(id=str(b["id"]), name=b.get("name") or str(b["id"])) for b in rows] + except Exception: + return [] + + +async def _db_specs() -> list[Candidate]: + """Distinct spec values: EAV strings from the catalog + specs used in recent drafts.""" + values: dict[str, None] = {} + try: + async with SessionLocal() as s: + rows = (await s.execute(text( + "SELECT DISTINCT value_string FROM material_attribute_values " + "WHERE value_string IS NOT NULL AND btrim(value_string) <> '' LIMIT 500" + ))).scalars().all() + for v in rows: + values.setdefault(str(v), None) + except Exception: + pass + try: + async with SessionLocal() as s: + rows = (await s.execute(text( + "SELECT DISTINCT spec FROM bot_drafts d, " + "jsonb_array_elements(COALESCE(d.payload->'items', '[]'::jsonb)) it, " + "jsonb_array_elements_text(COALESCE(it->'specs', '[]'::jsonb)) spec " + "WHERE btrim(spec) <> '' LIMIT 300" + ))).scalars().all() + for v in rows: + values.setdefault(str(v), None) + except Exception: + pass + return [Candidate(id=v, name=v) for v in values] + + class Resolver: """Caches catalog lists (units/categories/projects are stable; suppliers/materials queried on demand).""" @@ -236,6 +273,74 @@ class Resolver: items = await _db_materials(q=q) return _rank(q, items) if q else items + # ------------------------------------------------- full lists (Jev, v5) + + async def _fetch_all(self, fetch_page, max_pages: int = 5) -> list[dict]: + """Page through a paginated ERP list endpoint (page_size=100).""" + out: list[dict] = [] + for page in range(1, max_pages + 1): + rows = await fetch_page(page) + out += rows + if len(rows) < 100: + break + return out + + async def all_suppliers(self) -> list[Candidate]: + async def fetch(): + try: + rows = await self._fetch_all( + lambda page: self._api.list_suppliers(page=page, page_size=100) + ) + return [ + Candidate( + id=str(s["id"]), + name=s.get("name") or str(s["id"]), + extra={ + "location": s.get("location"), + "phone": s.get("phone"), + "contact_person": s.get("contact_person"), + "category_name": (s.get("category") or {}).get("name") + if isinstance(s.get("category"), dict) else s.get("category_name"), + }, + ) + for s in rows + ] + except Exception: + return await _db_suppliers() + return await self._cached("all_suppliers", fetch) + + async def all_materials(self) -> list[Candidate]: + async def fetch(): + try: + rows = await self._fetch_all( + lambda page: self._api.list_materials(page=page, page_size=100) + ) + return [ + Candidate( + id=str(m["id"]), + name=m.get("name") or str(m["id"]), + extra={"canonical_unit_id": m.get("canonical_unit_id")}, + ) + for m in rows + ] + except Exception: + return await _db_materials() + return await self._cached("all_materials", fetch) + + async def brands(self) -> list[Candidate]: + async def fetch(): + try: + rows = await self._fetch_all( + lambda page: self._api.list_brands(page=page, page_size=100) + ) + return [Candidate(id=str(b["id"]), name=b.get("name") or str(b["id"])) for b in rows] + except Exception: + return await _db_brands() + return await self._cached("brands", fetch) + + async def specs(self) -> list[Candidate]: + return await self._cached("specs", _db_specs) + # ------------------------------------------------------------ resolutions async def resolve(self, kind: str, name: str | None) -> Resolution: diff --git a/bot/tools/test_jev.py b/bot/tools/test_jev.py new file mode 100644 index 0000000..5112ada --- /dev/null +++ b/bot/tools/test_jev.py @@ -0,0 +1,115 @@ +"""Phase-2 tests for the Jev client — MOCKED HTTP (no network, no key needed). + + docker compose exec bot python tools/test_jev.py +""" +import asyncio +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from app import jev +from app.config import settings + +PASSED: list[str] = [] +FAILED: list[str] = [] + + +def check(name: str, ok: bool, extra: str = "") -> None: + (PASSED if ok else FAILED).append(name) + print(f"{'PASS' if ok else 'FAIL'} {name}{' — ' + str(extra) if extra else ''}") + + +class FakeResp: + def __init__(self, status: int, body: dict): + self.status_code = status + self._body = body + self.text = str(body) + + def json(self): + return self._body + + +class FakeClient: + """Scripted httpx.AsyncClient stand-in: pops one (status, body) per POST.""" + + script: list = [] + posts: int = 0 + + def __init__(self, *a, **k): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def post(self, url, json=None, headers=None): + FakeClient.posts += 1 + status, body = FakeClient.script.pop(0) + return FakeResp(status, body) + + +Q = {"m0": {"type": "choice", "instructions": "i", "criteria": {"x": ""}}} + + +async def main() -> None: + orig_client = jev.httpx.AsyncClient + saved_key = settings.TYPESAFE_API_KEY + jev.httpx.AsyncClient = FakeClient + try: + # 1) no key -> None (no HTTP at all) + settings.TYPESAFE_API_KEY = "" + FakeClient.posts = 0 + r = await jev.ask({"s": 1}, Q) + check("no api key -> None (no request)", r is None and FakeClient.posts == 0) + + settings.TYPESAFE_API_KEY = saved_key or "test-key" + + # 2) happy path: answers + model + usage + latency + FakeClient.script = [(200, {"model": "jev-1.13.0", + "answers": {"m0": {"choice": "x", "confidence": 0.9, + "probabilities": {"x": 0.9}}}, + "usage": {"input_tokens": 10, "output_tokens": 2}})] + r = await jev.ask({"s": 1}, Q) + ok = bool(r) and r["answers"]["m0"]["choice"] == "x" and r["model"] == "jev-1.13.0" \ + and r["usage"]["input_tokens"] == 10 and isinstance(r["latency_ms"], int) + check("happy path -> answers/model/usage/latency", ok) + + # 3) 429 -> one retry, then success + FakeClient.script = [(429, {"error": "rate_limited"}), + (200, {"model": "jev-1.13.0", "answers": {}, "usage": {}})] + FakeClient.posts = 0 + r = await jev.ask({"s": 1}, Q) + check("429 retried once then succeeds", r is not None and FakeClient.posts == 2) + + # 4) hard 400 -> None, no retry + FakeClient.script = [(400, {"error": "bad_request"})] + FakeClient.posts = 0 + r = await jev.ask({"s": 1}, Q) + check("400 -> None without retry", r is None and FakeClient.posts == 1) + + # 5) exception -> retried once -> None + class BoomClient(FakeClient): + async def post(self, url, json=None, headers=None): + FakeClient.posts += 1 + raise RuntimeError("net down") + + jev.httpx.AsyncClient = BoomClient + FakeClient.posts = 0 + r = await jev.ask({"s": 1}, Q) + check("exception retried once -> None", r is None and FakeClient.posts == 2) + finally: + jev.httpx.AsyncClient = orig_client + settings.TYPESAFE_API_KEY = saved_key + + total = len(PASSED) + len(FAILED) + print(f"\n{len(PASSED)}/{total} passed") + if FAILED: + print("FAILED:", FAILED) + raise SystemExit(1) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/bot/tools/test_matching.py b/bot/tools/test_matching.py new file mode 100644 index 0000000..13cc345 --- /dev/null +++ b/bot/tools/test_matching.py @@ -0,0 +1,196 @@ +"""Phase 3-5 tests for entity matching (mock by default; live with RUN_LIVE=1). + + docker compose exec bot python tools/test_matching.py # mock only + docker compose exec -e RUN_LIVE=1 bot python tools/test_matching.py # + real NLU + Jev + +The live section calls DeepSeek (NLU) and TypeSafe (Jev) with the REAL keys +from bot/.env — it never sends anything to Bale and never creates drafts. +""" +import asyncio +import os +import sys +from types import SimpleNamespace +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from app import matching +from app.config import settings +from app.nlu import normalize_text +from app.resolve import Candidate + +PASSED: list[str] = [] +FAILED: list[str] = [] + + +def check(name: str, ok: bool, extra: str = "") -> None: + (PASSED if ok else FAILED).append(name) + print(f"{'PASS' if ok else 'FAIL'} {name}{' — ' + str(extra) if extra else ''}") + + +class FakeResolver: + async def projects(self): + return [Candidate(id="p1", name="دفتر"), Candidate(id="p2", name="منشانه")] + + async def all_suppliers(self): + return [Candidate(id="s1", name="بازرگانی مصطفوی", + extra={"location": "اصفهان", "category_name": "مصالح"})] + + async def all_materials(self): + return [Candidate(id="m1", name="گچ و خاک", extra={"canonical_unit_id": 2})] + + async def brands(self): + return [Candidate(id="b1", name="سیمان تهران")] + + async def specs(self): + return [Candidate(id="سایز ۱۴", name="سایز ۱۴")] + + async def units(self): + return [Candidate(id="u1", name="کیسه"), Candidate(id="u2", name="عدد")] + + +def make_fake_ask(answers_by_mention: dict, model: str = "jev-1.13.0"): + async def fake_ask(state, questions): + if fake_ask.fail: + return None + return {"answers": answers_by_mention, "usage": {"input_tokens": 1, "output_tokens": 1}, + "model": model, "latency_ms": 7} + fake_ask.fail = False + return fake_ask + + +async def mock_tests() -> None: + ctx = SimpleNamespace(resolver=FakeResolver()) + orig_ask = matching.jev_ask + + # 1) routing bands: exact / ambiguity / new + variant grouping + matching.jev_ask = make_fake_ask({ + "mention_0": {"choice": "منشانه", "confidence": 0.9, + "probabilities": {"منشانه": 0.95, "_none": 0.05}}, + "mention_1": {"choice": "دفتر", "confidence": 0.7, + "probabilities": {"دفتر": 0.73, "منشانه": 0.15, "_none": 0.12}}, + "mention_2": {"choice": "_none", "confidence": 0.9, + "probabilities": {"_none": 0.94}}, + }) + mentions = [ + {"field": "project", "variants": ["منشیانه"]}, + {"field": "project", "variants": ["دفتر", "دفر"]}, + {"field": "project", "variants": ["چیز عجیب"]}, + ] + res = await matching.resolve_mentions(ctx, {"items": []}, mentions, None) + r0 = matching.lookup(res, "منشیانه") + r1 = matching.lookup(res, "دفر") # variant key resolves too + r2 = matching.lookup(res, "چیز عجیب") + check("exact band -> status exact + id", r0 and r0["status"] == "exact" and r0["id"] == "p2") + check("mid band -> ambiguity + candidates", r1 and r1["status"] == "ambiguity" + and r1["candidates"][0]["name"] == "دفتر") + check("variant of same mention shares one result", r1 is r0 if False else + (r1 and matching.lookup(res, "دفتر") is r1)) + check("_none / low conf -> new", r2 and r2["status"] == "new") + + # 2) Jev failure -> empty map (callers fall back to difflib) + matching.jev_ask = make_fake_ask({}) + matching.jev_ask.fail = True + res_fail = await matching.resolve_mentions(ctx, {"items": []}, mentions, None) + check("jev failure -> empty results (difflib fallback path)", res_fail == {}) + + # 3) apply_exact writes supplier/project ids into a payload + payload = {"items": []} + matching.apply_exact(payload, {"field": "project", "id": "p2", "name": "منشانه", + "variants": ["منشیانه"], "scores": {}}) + matching.apply_exact(payload, {"field": "supplier", "id": "s1", "name": "بازرگانی مصطفوی", + "extra": {"location": "اصفهان"}, "variants": ["مستفدی"]}) + check("apply_exact -> project_id/resolved_projects/project_ids", + payload.get("project_id") == "p2" + and payload["resolved_projects"].get("منشیانه") == "p2") + check("apply_exact -> supplier_id/name/extra", + payload.get("supplier_id") == "s1" and payload.get("supplier_location") == "اصفهان") + + # 4) material/unit/brand/spec application on items + payload2 = {"items": [{"material_name": "مصطفی", "unit_name": "کیسه", + "brand_name": "سیمان تهران", "specs": ["سایز 14"]}]} + matching.apply_choice(payload2, "material", {"id": "m1", "name": "گچ و خاک", + "extra": {"canonical_unit_id": 2}}, + ["مصطفی"]) + matching.apply_choice(payload2, "unit", {"id": "u1", "name": "کیسه"}, ["کیسه"]) + matching.normalize_brand_spec(payload2, {normalize_text("سایز 14"): { + "field": "spec", "status": "exact", "name": "سایز ۱۴", "id": "سایز ۱۴", + "variants": ["سایز 14"]}}) + it = payload2["items"][0] + check("material/unit applied on matching items", + it["material_id"] == "m1" and it["unit_id"] == "u1" and it["unit_explicit"] is True) + check("spec normalization (spelling only)", it["specs"][0] == "سایز ۱۴") + + # 5) create_ambiguities builds an entity_match entry with percentages + payload3 = {"items": [], "ambiguities": []} + jev_map = {normalize_text("منشیانه"): { + "field": "project", "status": "ambiguity", "variants": ["منشیانه"], + "candidates": [{"id": "p2", "name": "منشانه", "prob": 0.8}, + {"id": "p1", "name": "دفتر", "prob": 0.02}]}} + matching.create_ambiguities(payload3, jev_map) + ambs = payload3["ambiguities"] + check("entity_match ambiguity created with candidates + details", + len(ambs) == 1 and ambs[0]["kind"] == "entity_match" + and len(ambs[0]["candidates"]) == 2 and "۸۰٪" in ambs[0]["details"]) + matching.create_ambiguities(payload3, jev_map) # idempotent + check("create_ambiguities is deduped", len(payload3["ambiguities"]) == 1) + + matching.jev_ask = orig_ask + + +# --------------------------------------------------------------------------- +# live section (real NLU + real Jev) — opt-in and Bale-free +# --------------------------------------------------------------------------- + +HARNESS_TRANSCRIPT = ( + "مدل 1 (Whisper): الان شبت بخیر می خواستیم که بیست و پنج تا کیسه گچ و خاک از مصطفوی بخریم " + "و کنارش هم یه پنجاه تا کیسه سیمان پنجاه کیلویی. خریدمون هم عمدتا برای منشیانه و دفتر. " + "پیش فاکتورم گفته تا حدودا بیست روزی اوکیه.\n" + "مدل 2 (Vosk): میخواستیم 25 تا کیسه گچ و خاک از مستفدی بخریم و کنارشم یه 50 تا کیسه سیمان " + "50 کیلویی، خریدمون هم برای منشیانه و دفتر." +) + + +async def live_tests() -> None: + from app.api.client import ErpClient + from app.resolve import Resolver + from app.nlu import understand + + api = ErpClient(settings.API_BASE_URL, settings.API_TIMEOUT_SECONDS, settings.SERVICE_TOKEN) + ctx = SimpleNamespace(resolver=Resolver(api)) + + print("\n--- LIVE: NLU entity_mentions ---") + result = await understand(settings, HARNESS_TRANSCRIPT) + mentions = result.get("entity_mentions") or [] + print("intent:", result.get("intent"), "| mentions:", mentions) + check("live NLU produced entity_mentions", len(mentions) >= 2, f"{len(mentions)} mentions") + sup = next((m for m in mentions if m["field"] == "supplier"), None) + check("voice spellings grouped into ONE supplier mention", + bool(sup) and len(sup["variants"]) >= 2, str(sup)) + + print("\n--- LIVE: Jev per-field matching (real data) ---") + jev_map = await matching.resolve_mentions(ctx, result.get("patch") or {}, mentions, None) + for key, r in sorted(jev_map.items()): + print(f" {key!r}: {r['status']} -> {r.get('name')!r} conf={r.get('confidence')}") + check("live matching returned results", len(jev_map) > 0, f"{len(jev_map)} variants") + + +async def main() -> None: + await mock_tests() + if os.environ.get("RUN_LIVE") == "1": + if not settings.TYPESAFE_API_KEY: + check("live section requires TYPESAFE_API_KEY", False, "bot/.env not mounted?") + else: + await live_tests() + else: + print("\n(live section skipped — run with RUN_LIVE=1 to hit real NLU + Jev)") + + total = len(PASSED) + len(FAILED) + print(f"\n{len(PASSED)}/{total} passed") + if FAILED: + print("FAILED:", FAILED) + raise SystemExit(1) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/docker-compose.yml b/docker-compose.yml index 6095bc5..9544234 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -323,6 +323,7 @@ services: volumes: - ./bot/app:/srv/bot/app - ./bot/data:/srv/bot/data + - ./bot/.env:/srv/bot/.env:ro environment: PROCUREMENT_BOT_BOT_TOKEN: ${BALE_BOT_TOKEN:-1354444845:yhVMnKZBFuKaU3UW5fP1bHmknEHx99mipWw}