Function _canonicalize_topics

scripts/drift_engine.py:86–98

(values: List[Any])

def _canonicalize_topics(values: List[Any]) -> List[str]:
    if direction_lexicon and hasattr(direction_lexicon, "canonicalize_direction_terms"):
        try:
            return direction_lexicon.canonicalize_direction_terms(values, keep_unknown=True)
        except Exception:
            pass

    canonical_topics: List[str] = []
    for value in values:
        cleaned = str(value or "").strip().lower().replace("_", "-").replace(" ", "-")
        if cleaned and cleaned not in canonical_topics:
            canonical_topics.append(cleaned)
    return canonical_topics


def _direction_match_terms(direction: str) -> List[str]:
    cleaned = str(direction or "").strip()