"""Name → ERP id resolution with fuzzy matching (Persian-aware) + tiny caches. The ERP search endpoints are ilike-based; we fetch candidates with `q=` and rank them with difflib so "میلگرد A3 14" still matches "میلگرد A3 سایز ۱۴". """ import difflib import time from dataclasses import dataclass, field from sqlalchemy import text from app.api.client import ApiError, ErpClient from app.db import SessionLocal from app.nlu import normalize_text @dataclass class Candidate: id: str name: str extra: dict = field(default_factory=dict) @dataclass class Resolution: status: str # "one" | "many" | "none" candidates: list[Candidate] @property def resolved(self) -> bool: return self.status == "one" def _rank(query: str, items: list[Candidate], limit: int = 6) -> list[Candidate]: q = normalize_text(query) if not q: return items[:limit] scored = [] for item in items: name = normalize_text(item.name) if q == name: ratio = 1.0 elif q in name and len(q) >= 3: ratio = 0.85 + 0.15 * (len(q) / len(name)) elif name in q and len(name) >= 3: ratio = 0.85 + 0.15 * (len(name) / len(q)) else: ratio = difflib.SequenceMatcher(None, q, name).ratio() if ratio >= 0.72: scored.append((ratio, item)) scored.sort(key=lambda pair: (-pair[0], pair[1].name)) return [item for ratio, item in scored[:limit]] async def _db_units() -> list[Candidate]: try: async with SessionLocal() as s: rows = (await s.execute(text("SELECT id, code, name_fa FROM units"))).mappings().all() return [ Candidate( id=str(u["id"]), name=u.get("name_fa") or u.get("code") or str(u["id"]), extra={"code": u.get("code")}, ) for u in rows ] except Exception: return [] async def _db_categories() -> list[Candidate]: try: async with SessionLocal() as s: rows = (await s.execute(text("SELECT id, name, slug FROM material_categories"))).mappings().all() return [ Candidate( id=str(c["id"]), name=c.get("name") or c.get("slug") or str(c["id"]), ) for c in rows ] except Exception: return [] async def _db_projects() -> list[Candidate]: try: async with SessionLocal() as s: rows = (await s.execute(text("SELECT id, name FROM projects"))).mappings().all() return [ Candidate( id=str(p["id"]), name=p.get("name") or str(p["id"]), ) for p in rows ] except Exception: return [] async def _db_suppliers(q: str | None = None) -> list[Candidate]: try: async with SessionLocal() as s: if q: rows = (await s.execute( text("SELECT s.id, s.name, s.location, s.phone, s.contact_person, c.name as category_name " "FROM suppliers s LEFT JOIN material_categories c ON s.category_id = c.id " "WHERE s.name ILIKE :q LIMIT 20"), {"q": f"%{q}%"} )).mappings().all() else: rows = (await s.execute( text("SELECT s.id, s.name, s.location, s.phone, s.contact_person, c.name as category_name " "FROM suppliers s LEFT JOIN material_categories c ON s.category_id = c.id LIMIT 20") )).mappings().all() return [ Candidate( id=str(r["id"]), name=r.get("name") or str(r["id"]), extra={ "location": r.get("location"), "phone": r.get("phone"), "contact_person": r.get("contact_person"), "category_name": r.get("category_name"), }, ) for r in rows ] except Exception: return [] async def _db_materials(q: str | None = None) -> list[Candidate]: try: async with SessionLocal() as s: if q: rows = (await s.execute( text("SELECT id, name, topic, canonical_unit_id, status FROM materials WHERE name ILIKE :q LIMIT 20"), {"q": f"%{q}%"} )).mappings().all() else: rows = (await s.execute( text("SELECT id, name, topic, canonical_unit_id, status FROM materials LIMIT 20") )).mappings().all() return [ Candidate( id=str(r["id"]), name=r.get("name") or str(r["id"]), extra=dict(r), ) for r in rows ] except Exception: 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).""" def __init__(self, api: ErpClient, ttl_seconds: float = 1800): self._api = api self._ttl = ttl_seconds self._cache: dict[str, tuple[float, list[Candidate]]] = {} async def _cached(self, key: str, fetch) -> list[Candidate]: entry = self._cache.get(key) now = time.monotonic() if entry and now - entry[0] < self._ttl: return entry[1] items = await fetch() self._cache[key] = (now, items) return items def invalidate(self, key: str | None = None) -> None: if key: self._cache.pop(key, None) else: self._cache.clear() # ---------------------------------------------------------------- sources async def units(self) -> list[Candidate]: async def fetch(): try: rows = await self._api.list_units() return [Candidate(id=str(u["id"]), name=u.get("name_fa") or u.get("code") or str(u["id"]), extra={"code": u.get("code")}) for u in rows] except Exception: return await _db_units() return await self._cached("units", fetch) async def categories(self) -> list[Candidate]: async def fetch(): try: rows = await self._api.list_categories() return [Candidate(id=str(c["id"]), name=c.get("name") or c.get("slug") or str(c["id"])) for c in rows] except Exception: return await _db_categories() return await self._cached("categories", fetch) async def projects(self) -> list[Candidate]: async def fetch(): try: rows = await self._api.list_projects() return [Candidate(id=str(p["id"]), name=p.get("name") or str(p["id"])) for p in rows] except Exception: return await _db_projects() return await self._cached("projects", fetch) async def suppliers(self, q: str | None = None) -> list[Candidate]: try: rows = await self._api.list_suppliers(q=q, page_size=20) items = [ 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 None, }, ) for s in rows ] except Exception: items = await _db_suppliers(q=q) return _rank(q, items) if q else items async def materials(self, q: str | None = None) -> list[Candidate]: try: rows = await self._api.list_materials(q=q, page_size=20) extra = {str(m["id"]): m for m in rows} items = [Candidate(id=str(m["id"]), name=m.get("name") or str(m["id"]), extra=extra.get(str(m["id"]), {})) for m in rows] except Exception: 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: """Resolve by name; exact hit → "one", several close → "many", none → "none".""" if not name or not name.strip(): return Resolution("none", []) if kind == "supplier": candidates = await self.suppliers(name) elif kind == "material": candidates = await self.materials(name) elif kind == "project": candidates = _rank(name, await self.projects()) elif kind == "unit": candidates = _rank(name, await self.units()) elif kind == "category": candidates = _rank(name, await self.categories()) else: raise ValueError(f"unknown kind: {kind}") if not candidates: return Resolution("none", []) norm_name = normalize_text(name) exact = [c for c in candidates if normalize_text(c.name) == norm_name] if len(exact) == 1: return Resolution("one", exact) top = candidates[0] top_norm = normalize_text(top.name) ratio = difflib.SequenceMatcher(None, norm_name, top_norm).ratio() if (norm_name in top_norm or top_norm in norm_name) and len(norm_name) >= 3: ratio = max(ratio, 0.90) if ratio >= 0.88 and (len(candidates) == 1 or ratio > 0.95): return Resolution("one", [top]) return Resolution("many", candidates) async def unit_for_material(self, unit_name: str | None, material: dict | None) -> tuple[str | None, bool]: """Return (unit_id, explicitly_resolved). Falls back to the material's canonical unit when the spoken unit isn't found.""" if unit_name: res = await self.resolve("unit", unit_name) if res.resolved: return res.candidates[0].id, True if material and material.get("canonical_unit_id"): return str(material["canonical_unit_id"]), False return None, False async def safe_resolve(resolver: Resolver, kind: str, name: str | None) -> Resolution: try: return await resolver.resolve(kind, name) except Exception: return Resolution("none", [])