Autoparts-DB/pos/services/sku_matcher.py

# /home/Autopartes/pos/services/sku_matcher.py
"""Inventory <-> supplier-catalog SKU matching.

Tries to populate ``inventory.catalog_part_id`` (today 0% linked) and register
the matched catalog SKU as an alias in ``inventory_sku_aliases``.

Matching strategy (first hit wins), all over a *normalized* key (uppercase,
dashes/spaces/dots stripped):

  1. exact  — inventory.part_number == supplier_catalog.sku
  2. xref   — inventory.part_number == supplier_catalog_interchange.part_number
  3. name   — fuzzy name similarity (difflib, no sklearn available) against the
              supplier-catalog names, blocked by first significant token.
  4. semantic — optional (opt-in) cosine similarity over QWEN embeddings
              (``semantic_search.embed``), used to fill the gaps difflib
              leaves behind (same part, different wording/abbreviations).
              Catalog-name embeddings are cached in-memory per ``CatalogIndex``
              so a long run embeds each catalog name at most once.  With no
              network/provider the step degrades to a no-op.

``supplier_catalog`` / ``supplier_catalog_interchange`` live in the MASTER DB;
``inventory`` / ``inventory_sku_aliases`` live in the tenant DB.  Because the
two databases are queried with separate connections, the join happens in
Python (the catalog is small: ~17k items / ~18k distinct cross-refs).

Public API:
    match_inventory(tenant_conn, master_conn, limit=None, ...) -> summary dict
    load_catalog_index(master_conn) -> CatalogIndex
"""

import re
import time
import unicodedata
from difflib import SequenceMatcher

_SKU_BLOCK_RE = re.compile(r"[^A-Z0-9]+")


def norm_sku(s):
    """Normalize a SKU/part number for exact matching."""
    s = unicodedata.normalize("NFD", str(s or ""))
    s = s.encode("ascii", "ignore").decode("ascii")
    return _SKU_BLOCK_RE.sub("", s.upper())


def _name_tokens(s):
    s = unicodedata.normalize("NFD", str(s or ""))
    s = s.encode("ascii", "ignore").decode("ascii")
    return [t for t in re.split(r"[^A-Za-z0-9]+", s.upper()) if t]


def _name_norm(s):
    return " ".join(_name_tokens(s))


def _load_semantic_search():
    """Import ``semantic_search`` whether pos/ is on sys.path or not."""
    try:
        from services import semantic_search as ss
    except Exception:
        try:
            import semantic_search as ss
        except Exception:
            return None
    return ss


def _embed_texts(texts, batch=64, timeout=None, retries=1):
    """Embed ``texts`` in batches via ``semantic_search.embed``.

    Returns the list of vectors in input order, or ``None`` when the provider
    is unavailable / errors out (never raises).  A single retry covers the
    transient 429/timeout seen on the shared QWEN provider.
    """
    items = list(texts or [])
    if not items:
        return []
    ss = _load_semantic_search()
    if ss is None:
        return None
    vectors = []
    for i in range(0, len(items), max(int(batch), 1)):
        chunk = items[i:i + max(int(batch), 1)]
        out = None
        for attempt in range(int(retries) + 1):
            try:
                out = ss.embed(chunk, timeout=timeout)
            except Exception:
                out = None
            if out and len(out) == len(chunk):
                break
            if attempt < int(retries):
                time.sleep(2.0)
        if not out or len(out) != len(chunk):
            return None
        vectors.extend(out)
    return vectors


class CatalogIndex:
    """In-memory index over supplier_catalog (+ interchange)."""

    def __init__(self):
        self.by_sku = {}          # norm sku -> list of catalog dicts
        self.by_xref = {}         # norm part_number -> list of catalog dicts
        self.by_token = {}        # first significant name token -> list of dicts
        self.total = 0
        self._emb_cache = {}      # name_norm -> embedding vector (lazy)
        self.emb_requests = 0     # provider calls used by the semantic matcher

    def add(self, cat_id, supplier, sku, name, category):
        rec = {
            "catalog_id": cat_id, "supplier": supplier, "sku": sku,
            "name": name, "category": category, "name_norm": _name_norm(name),
        }
        self.total += 1
        ns = norm_sku(sku)
        if ns:
            self.by_sku.setdefault(ns, []).append(rec)
        tokens = _name_tokens(name)
        if tokens:
            self.by_token.setdefault(tokens[0], []).append(rec)
        return rec

    def add_xref(self, norm_pn, rec):
        if norm_pn:
            self.by_xref.setdefault(norm_pn, []).append(rec)

    def match_sku(self, norm_pn):
        return self.by_sku.get(norm_pn)

    def match_xref(self, norm_pn):
        return self.by_xref.get(norm_pn)

    def match_name(self, name, threshold=0.82):
        """Best fuzzy name match (blocked by first token). Returns (rec, score)."""
        nt = _name_norm(name)
        tokens = _name_tokens(name)
        if not tokens or len(nt) < 4:
            return None, 0.0
        candidates = self.by_token.get(tokens[0])
        if not candidates:
            return None, 0.0
        best, best_score = None, 0.0
        for rec in candidates:
            if not rec["name_norm"]:
                continue
            score = SequenceMatcher(None, nt, rec["name_norm"]).ratio()
            if score > best_score:
                best, best_score = rec, score
        if best and best_score >= threshold:
            return best, best_score
        return None, best_score

    def _embed_names(self, names, batch=64, timeout=None):
        """Return {name_norm: vector}, embedding only the uncached names.

        Catalog-side cache: each distinct ``name_norm`` is embedded once per
        index.  Returns ``None`` (degraded) when the provider is unavailable.
        """
        missing = [n for n in dict.fromkeys(names) if n and n not in self._emb_cache]
        if missing:
            self.emb_requests += 1
            vectors = _embed_texts(missing, batch=batch, timeout=timeout)
            if vectors is None:
                return None
            for n, v in zip(missing, vectors):
                self._emb_cache[n] = v
        return self._emb_cache

    def match_name_semantic(self, name, threshold=0.86, top_k=200,
                            batch=64, timeout=None):
        """Best semantic name match (QWEN embeddings). Returns (rec, score).

        Only the closest ``top_k`` candidates (blocked by first token, then
        pre-ranked with difflib) are embedded, keeping cost bounded.  Degrades
        to ``(None, 0.0)`` when embeddings are unavailable.
        """
        nt = _name_norm(name)
        tokens = _name_tokens(name)
        if not tokens or len(nt) < 4:
            return None, 0.0
        candidates = self.by_token.get(tokens[0]) or []
        if not candidates:
            return None, 0.0

        ss = _load_semantic_search()
        if ss is None:
            return None, 0.0
        try:
            if hasattr(ss, 'is_configured') and not ss.is_configured():
                return None, 0.0
        except Exception:
            pass

        pool = [(SequenceMatcher(None, nt, r["name_norm"]).ratio(), r)
                for r in candidates if r["name_norm"]]
        pool.sort(key=lambda t: t[0], reverse=True)
        pool = [r for _, r in pool[:max(int(top_k), 1)]]
        if not pool:
            return None, 0.0

        vecs = self._embed_names([r["name_norm"] for r in pool],
                                 batch=batch, timeout=timeout)
        if not vecs:
            return None, 0.0
        query_vecs = _embed_texts([nt], batch=1, timeout=timeout)
        if not query_vecs:
            return None, 0.0
        query_vec = query_vecs[0]

        best, best_sim = None, 0.0
        for rec in pool:
            vec = vecs.get(rec["name_norm"])
            if not vec:
                continue
            try:
                sim = ss.cosine_similarity(query_vec, vec)
            except Exception:
                sim = None
            if sim is not None and sim > best_sim:
                best, best_sim = rec, sim
        if best and best_sim >= threshold:
            return best, best_sim
        return None, best_sim


def load_catalog_index(master_conn, supplier=None):
    """Build a CatalogIndex from the master supplier_catalog tables."""
    idx = CatalogIndex()
    cur = master_conn.cursor()
    try:
        if supplier:
            cur.execute(
                "SELECT id, supplier_name, sku, name, category FROM supplier_catalog "
                "WHERE is_active = true AND supplier_name = %s",
                (supplier,),
            )
        else:
            cur.execute(
                "SELECT id, supplier_name, sku, name, category FROM supplier_catalog "
                "WHERE is_active = true"
            )
        recs = {}
        for cat_id, sup, sku, name, category in cur.fetchall():
            recs[cat_id] = idx.add(cat_id, sup, sku, name, category)

        # Interchange cross-refs, deduped (406k rows / ~18k distinct part numbers)
        cur.execute(
            "SELECT MIN(catalog_id), UPPER(part_number) FROM supplier_catalog_interchange "
            "GROUP BY UPPER(part_number)"
        )
        for catalog_id, part_number in cur.fetchall():
            rec = recs.get(catalog_id)
            if rec is None:
                continue
            idx.add_xref(norm_sku(part_number), rec)
    finally:
        cur.close()
    return idx


def match_inventory(tenant_conn, master_conn, limit=None, offset=0,
                    only_unmatched=True, create_aliases=True, use_names=True,
                    name_threshold=0.82, apply=True, supplier=None,
                    after_id=None, sample_limit=20, use_semantic=False,
                    semantic_threshold=0.86, semantic_top_k=200,
                    semantic_batch=64, semantic_timeout=20.0,
                    semantic_prefer_below=None, collect=False, index=None,
                    only_linked=False):
    """Match tenant inventory items against the supplier catalog.

    Args:
        only_unmatched: only consider items whose catalog_part_id is NULL/0.
        create_aliases: insert matched catalog SKUs into inventory_sku_aliases.
        use_names: enable the fuzzy-name fallback.
        apply: persist changes (False = dry run).
        use_semantic: enable the embedding-based name fallback (opt-in; it
            needs the QWEN provider and is slower, so pair it with ``limit``).
        semantic_threshold: minimum cosine similarity (qwen3-embedding scores
            unrelated names low and near-synonyms high; 0.86 is conservative).
        semantic_prefer_below: when set, also run the semantic matcher for
            items whose fuzzy score is below this value and keep whichever
            scores higher (lets semantic *correct* weak difflib hits).

    Returns a summary dict with counts by method and a sample of matches.
    """
    idx = index if index is not None else load_catalog_index(master_conn, supplier=supplier)

    cur = tenant_conn.cursor()
    where = ["is_active = true"]
    params = []
    if only_linked:
        where.append("(catalog_part_id IS NOT NULL AND catalog_part_id <> 0)")
    elif only_unmatched:
        where.append("(catalog_part_id IS NULL OR catalog_part_id = 0)")
    if after_id is not None:
        where.append("id > %s")
        params.append(after_id)
    sql = (
        "SELECT id, part_number, name, catalog_part_id FROM inventory "
        "WHERE " + " AND ".join(where) + " ORDER BY id"
    )
    if limit is not None:
        sql += " LIMIT %s OFFSET %s"
        params.extend([int(limit), int(offset)])
    cur.execute(sql, params)
    rows = cur.fetchall()
    cur.close()

    by_method = {"sku": 0, "sku_prefix": 0, "interchange": 0,
                 "interchange_prefix": 0, "name": 0, "name_semantic": 0}
    updated = 0
    aliases = 0
    skipped_existing = 0
    sample = []
    all_results = []
    last_id = after_id

    def _lookup(norm_pn):
        """(rec, method, score) for an exact/cross-ref key, else (None, None, 0).

        Cross-refs on very short numeric keys are skipped: a bare "3341" can
        collide with unrelated candidates in the 406k interchange table.
        """
        if not norm_pn:
            return None, None, 0.0
        hit = idx.match_sku(norm_pn)
        if hit:
            return hit[0], "sku", 1.0
        if len(norm_pn) >= 5 or any(c.isalpha() for c in norm_pn):
            hit = idx.match_xref(norm_pn)
            if hit:
                return hit[0], "interchange", 0.95
        return None, None, 0.0

    for inv_id, part_number, name, current_cpid in rows:
        last_id = inv_id
        norm_pn = norm_sku(part_number)

        match, method, score = _lookup(norm_pn)

        # Many tenant SKUs are "<vendor>-<manufacturer_code>"; retry stripped.
        if match is None and part_number and "-" in str(part_number):
            head, rest = str(part_number).split("-", 1)
            if head.isalpha() and 2 <= len(head) <= 4 and rest.strip():
                match, method, score = _lookup(norm_sku(rest))
                if match:
                    method = "sku_prefix" if method == "sku" else "interchange_prefix"
                    score = 0.96 if method == "sku_prefix" else 0.9

        if match is None and use_names and name:
            match, score = idx.match_name(name, threshold=name_threshold)
            if match:
                method = "name"

        needs_semantic = (
            use_semantic and name and
            (match is None or
             (semantic_prefer_below is not None and score < semantic_prefer_below))
        )
        if needs_semantic:
            s_rec, s_score = idx.match_name_semantic(
                name, threshold=semantic_threshold, top_k=semantic_top_k,
                batch=semantic_batch, timeout=semantic_timeout,
            )
            if s_rec and (match is None or s_score > score):
                match, method, score = s_rec, "name_semantic", s_score

        if match is None:
            continue

        by_method[method] += 1

        if apply:
            cur = tenant_conn.cursor()
            try:
                cur.execute(
                    "UPDATE inventory SET catalog_part_id = %s "
                    "WHERE id = %s AND (catalog_part_id IS NULL OR catalog_part_id = 0)",
                    (match["catalog_id"], inv_id),
                )
                if cur.rowcount:
                    updated += 1
                else:
                    skipped_existing += 1

                if create_aliases and match["sku"]:
                    cur.execute(
                        "INSERT INTO inventory_sku_aliases (inventory_id, sku, label) "
                        "VALUES (%s, %s, %s) ON CONFLICT (inventory_id, sku) DO NOTHING",
                        (inv_id, match["sku"], f"Catálogo:{match['supplier'] or 'proveedor'}"),
                    )
                    aliases += cur.rowcount
                tenant_conn.commit()
            finally:
                cur.close()

        if collect:
            all_results.append({
                "inventory_id": inv_id, "part_number": part_number, "name": name,
                "existing_catalog_id": current_cpid,
                "matched_catalog_id": match["catalog_id"],
                "matched_sku": match["sku"], "method": method,
                "score": round(score, 3),
            })

        if len(sample) < sample_limit:
            sample.append({
                "inventory_id": inv_id, "part_number": part_number, "name": name,
                "matched_catalog_id": match["catalog_id"],
                "matched_sku": match["sku"], "matched_name": match["name"],
                "supplier": match["supplier"], "method": method,
                "score": round(score, 3),
            })

    return {
        "scanned": len(rows),
        "matched": sum(by_method.values()),
        "by_method": by_method,
        "updated": updated,
        "aliases_created": aliases,
        "skipped_existing": skipped_existing,
        "catalog_size": idx.total,
        "sample": sample,
        "last_id": last_id,
        "results": all_results,
        "semantic": {
            "enabled": bool(use_semantic),
            "threshold": semantic_threshold,
            "prefer_below": semantic_prefer_below,
            "catalog_embeddings_cached": len(idx._emb_cache),
            "embed_requests": idx.emb_requests,
        },
    }


def count_links(tenant_conn):
    """Return (total_items, linked_items) for progress reporting."""
    cur = tenant_conn.cursor()
    try:
        cur.execute(
            "SELECT count(*), count(*) FILTER "
            "(WHERE catalog_part_id IS NOT NULL AND catalog_part_id <> 0) FROM inventory"
        )
        return cur.fetchone()
    finally:
        cur.close()