gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
54 lines
2.1 KiB
Python
54 lines
2.1 KiB
Python
"""Deterministic candidate generation from the ACTIVE official datasets.
|
|
|
|
The AI never receives an open question: for every task type the Hub first builds
|
|
the official candidate list (NCM top-K by description, CEST links for the NCM,
|
|
TIPI EX rows for the NCM). Providers may only pick from this list.
|
|
"""
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..models import CestEntry, CestNcmEntry, Dataset, NcmEntry, TipiEntry
|
|
from .fingerprint import token_stems
|
|
|
|
|
|
def _active_dataset(session: Session, kind: str):
|
|
return session.scalar(select(Dataset).where(Dataset.kind == kind, Dataset.status == "ACTIVE"))
|
|
|
|
|
|
def ncm_top_k(session: Session, description, k: int = 10):
|
|
"""Top-K official NCM codes by (stemmed) token overlap with the product description."""
|
|
dataset = _active_dataset(session, "NCM")
|
|
if dataset is None:
|
|
return []
|
|
stems = token_stems(description)
|
|
if not stems:
|
|
return []
|
|
scored = []
|
|
for entry in session.scalars(select(NcmEntry).where(NcmEntry.dataset_id == dataset.id)).all():
|
|
score = len(stems & token_stems(entry.description)) / len(stems)
|
|
if score > 0:
|
|
scored.append((entry.code, entry.description, round(score, 4)))
|
|
scored.sort(key=lambda row: (-row[2], row[0]))
|
|
return scored[:k]
|
|
|
|
|
|
def cest_candidates_for_ncm(session: Session, ncm: str):
|
|
dataset = _active_dataset(session, "CEST")
|
|
if dataset is None:
|
|
return []
|
|
rows = session.execute(
|
|
select(CestEntry, CestNcmEntry.ncm_code)
|
|
.join(CestNcmEntry, CestNcmEntry.cest_entry_id == CestEntry.id)
|
|
.where(CestEntry.dataset_id == dataset.id)
|
|
).all()
|
|
return list(dict.fromkeys(entry for entry, ncm_code in rows if ncm.startswith(ncm_code)))
|
|
|
|
|
|
def ex_tipi_candidates_for_ncm(session: Session, ncm: str):
|
|
dataset = _active_dataset(session, "TIPI_IPI")
|
|
if dataset is None:
|
|
return []
|
|
return list(dict.fromkeys(
|
|
entry for entry in session.scalars(select(TipiEntry).where(TipiEntry.dataset_id == dataset.id, TipiEntry.ncm == ncm, TipiEntry.ex.is_not(None))).all()
|
|
))
|