Rifqi Hafizuddin
[NOTICKET] feat(knowledge_extraction): paid extraction stage + validate, diff, queue
ab5ea78 | """Classify each candidate against the ACTIVE glossary version: new, duplicate | |
| or conflicting. | |
| The prototype diffed against the file it then overwrote, so every entry came | |
| back `new` and the interesting paths never ran. The baseline must therefore be | |
| supplied explicitly — an approved, versioned set — rather than read from | |
| wherever the last run happened to write. | |
| """ | |
| from __future__ import annotations | |
| from ..models import DiffStatus | |
| from ..settings import DUPLICATE_OVERLAP_THRESHOLD | |
| from ..validate.conflict import overlap | |
| def classify(entry: dict, existing_by_term: dict[str, dict]) -> DiffStatus: | |
| prior = existing_by_term.get((entry.get("term") or "").casefold()) | |
| if prior is None: | |
| return "new" | |
| a = (entry.get("definition") or "").strip() | |
| b = (prior.get("definition") or "").strip() | |
| if a and a == b: | |
| return "duplicate" | |
| if not a or not b: | |
| # One side abstained: not a contradiction, just less information. | |
| return "new" | |
| return "duplicate" if overlap(a, b) >= DUPLICATE_OVERLAP_THRESHOLD else "conflicting" | |
| def diff_glossary(entries: list[dict], active: list[dict]) -> list[dict]: | |
| existing_by_term = {(e.get("term") or "").casefold(): e for e in active} | |
| return [{**entry, "diff_status": classify(entry, existing_by_term)} for entry in entries] | |