feat(agent): integrate procurement AI agent with Bale bot, real-time draft sync, and git history bundle

ai procurement/app/resolve.py

@@ -0,0 +1,151 @@
+"""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 app.api.client import ApiError, ErpClient
+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)
+        ratio = difflib.SequenceMatcher(None, q, name).ratio()
+        if q == name:
+            ratio = 1.0
+        elif q in name:
+            ratio += 0.35
+        scored.append((ratio, item))
+    scored.sort(key=lambda pair: (-pair[0], pair[1].name))
+    return [item for ratio, item in scored[:limit] if ratio >= 0.40]
+
+
+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():
+            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]
+        return await self._cached("units", fetch)
+
+    async def categories(self) -> list[Candidate]:
+        async def fetch():
+            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]
+        return await self._cached("categories", fetch)
+
+    async def projects(self) -> list[Candidate]:
+        async def fetch():
+            rows = await self._api.list_projects()
+            return [Candidate(id=str(p["id"]), name=p.get("name") or str(p["id"])) for p in rows]
+        return await self._cached("projects", fetch)
+
+    async def suppliers(self, q: str | None = None) -> list[Candidate]:
+        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"])) for s in rows]
+        return _rank(q, items) if q else items
+
+    async def materials(self, q: str | None = None) -> list[Candidate]:
+        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]
+        return _rank(q, items) if q else items
+
+    # ------------------------------------------------------------ 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 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 ApiError:
+        return Resolution("none", [])