ERP/ai procurement/app/resolve.py

"""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", [])