289 lines
11 KiB
Python
289 lines
11 KiB
Python
"""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 []
|
|
|
|
|
|
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
|
|
|
|
# ------------------------------------------------------------ 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", [])
|