"""Select bounded, attributed coach knowledge for a finalized check-in.""" from __future__ import annotations import json import re from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Final from unicodedata import normalize from urllib.parse import urlparse _MANIFEST_NAME: Final[str] = "runtime-evidence.json" _APPROVED_MANIFEST_NAME: Final[str] = "legacy-corpus-approved.json" _SCHEMA_VERSION: Final[str] = "1.0" _MAX_EVIDENCE: Final[int] = 5 _MAX_PER_TOPIC: Final[int] = 2 _TOPIC_LIMITS: Final[dict[str, int]] = { "훈련·수행": 1, "안전·의료·약물": 1, "시합·피크위크": 1, } _MAX_TEXT_LENGTH: Final[int] = 600 _TOKEN: Final[re.Pattern[str]] = re.compile(r"[가-힣A-Za-z0-9]{2,}") _ID: Final[re.Pattern[str]] = re.compile(r"^[A-Za-z][A-Za-z0-9_.:-]{1,127}$") _APPROVED_CLUSTER_ID: Final[re.Pattern[str]] = re.compile(r"^approved-[0-9a-f]{16}$") _SAFE_REWRITE_ID: Final[re.Pattern[str]] = re.compile(r"^safe-rewrite-[0-9a-f]{16}$") _SOURCE_ID: Final[re.Pattern[str]] = re.compile(r"^(instagram|naver|youtube):\S+$", re.IGNORECASE) _HANGUL: Final[re.Pattern[str]] = re.compile(r"[가-힣]") _KOREAN_SUFFIXES: Final[tuple[str, ...]] = ( "으로", "에서", "에게", "처럼", "까지", "부터", "와", "과", "은", "는", "이", "가", "을", "를", "에", "의", "도", "로", "만", ) _RUNTIME_EXCLUDED_TOPICS: Final[frozenset[str]] = frozenset({ "안전·의료·약물", "시합·피크위크", }) _CURATED_ALLOWED_TOPICS: Final[frozenset[str]] = frozenset({ "소화·식단 반응", "훈련·수행", "회복·수면", "코칭 철학·기타", "안전·의료·약물", "시합·피크위크", }) _TOPIC_REQUIRED_TERMS: Final[dict[str, frozenset[str]]] = { "소화·식단 반응": frozenset({"소화", "장", "복부", "변비", "설사", "식욕", "fodmap"}), "증량·벌크": frozenset({"증량", "벌크", "린매스"}), "회복·수면": frozenset({"수면", "잠", "숙면", "회복", "피로", "컨디션"}), "훈련·수행": frozenset({ "훈련", "운동", "수행", "퍼포먼스", "성능", "볼륨", "부하", "중량", "가동범위", "기술", "회복", }), "안전·의료·약물": frozenset({ "안전", "통증", "증상", "부상", "의료", "진료", "상담", }), "시합·피크위크": frozenset({"시합", "대회", "피크위크"}), } _FIELD_TERMS: Final[dict[str, tuple[str, ...]]] = { "bodyweight": ( "체중", "몸무게", "체성분", "체지방", "감량", "증량", "정체", "플래토", "plateau", "수분", "글리코겐", "추세", "trend", "weight", ), "calories": ( "칼로리", "열량", "섭취", "식단", "영양", "탄수화물", "단백질", "지방", "감량", "벌크", "증량", "calorie", "kcal", ), "digestion": ( "소화", "장", "복부", "변비", "설사", "식욕", "fodmap", "음식", "식단", ), "sleep_duration": ( "수면", "잠", "숙면", "수면시간", "회복", "피로", "컨디션", "sleep", ), "sleep_quality": ( "수면", "잠", "숙면", "수면의 질", "회복", "피로", "컨디션", "sleep", ), "condition": ( "컨디션", "피로", "회복", "수면", "수행", "퍼포먼스", "readiness", ), "pain": ( "통증", "아픔", "관절", "부상", "불편", "증상", "회복", "pain", ), "training_plan": ( "훈련", "운동", "수행", "퍼포먼스", "성능", "볼륨", "부하", "중량", "가동범위", "기술", "회복", "training", "performance", ), "training_summary": ( "훈련", "운동", "수행", "퍼포먼스", "성능", "볼륨", "부하", "중량", "가동범위", "기술", "회복", "training", "performance", ), "workout_quality": ( "훈련", "운동", "수행", "퍼포먼스", "성능", "볼륨", "부하", "중량", "가동범위", "기술", "회복", "training", "performance", ), } _NO_FIELD_EXPANSION: Final[frozenset[str]] = frozenset({ "training_plan", "training_summary", "workout_quality", "optional_note", }) _EVIDENCE_GRADE_RANK: Final[dict[str, int]] = { "primary_source_extracted": 3, "A": 3, "B": 2, "C": 1, } _RUNTIME_SAFETY_CLASSES: Final[frozenset[str]] = frozenset({ "general_training", "general_nutrition", "recovery", "digestive_observation", }) _REJECTED_ROW_STATUSES: Final[frozenset[str]] = frozenset({ "raw", "unavailable", "title", "title-only", "title_only", "pending", "pending_review", "context-only", "context_only", "fragment", "insufficient_fragment", "quarantined", "quarantine", }) _CURATED_CLAIM_TYPES: Final[frozenset[str]] = frozenset({ "author_viewpoint", "general_guidance", }) _CURATED_PROVENANCE_KINDS: Final[frozenset[str]] = frozenset({ "independent_review", "semantic_extraction", }) _APPROVED_INVARIANTS: Final[frozenset[str]] = frozenset({ "input_rows_accounted", "canonical_identities_accounted", "semantic_records_accounted", "all_flagged_reviewed", "all_safety_reviewed", "approved_quarantine_disjoint", "raw_and_runtime_not_modified", "quarantined_originals_remain_excluded", "safe_replacements_are_separate_from_source_approval", }) _CURATED_KIND_PRIORITY: Final[dict[str, int]] = { "exact_source_excerpt": 2, "curated_bounded_principle": 1, "approved_safe_rewrite": 1, } _CURATED_TOPIC_SAFETY: Final[dict[str, str]] = { "소화·식단 반응": "digestive_observation", "회복·수면": "recovery", "훈련·수행": "general_training", "코칭 철학·기타": "general_training", "안전·의료·약물": "recovery", "시합·피크위크": "general_training", } _FORBIDDEN_CURATED_KEYS: Final[frozenset[str]] = frozenset({ "context_only", "context-only", "raw_excerpt", "raw_text", "original_text", "original_claim", "quarantined_original", "terminal_identity", }) @dataclass(frozen=True, slots=True) class RetrievedKnowledge: """One attributed, bounded evidence item selected for model reference.""" source: str topic: str source_url: str text: str evidence_id: str = "" evidence_kind: str = "" confidence: str = "" @dataclass(frozen=True, slots=True) class _KnowledgeEntry: canonical_source_id: str evidence_id: str evidence_kind: str source: str source_platforms: frozenset[str] topic: str source_url: str text: str source_digest: str extracted_at_kst: str reviewed_at_kst: str evidence_grade: str safety_class: str truncated: bool confidence: str source_priority: int tokens: frozenset[str] def retrieve_for_checkin(profile_root: Path, answers: dict[str, str]) -> tuple[RetrievedKnowledge, ...]: """Return bounded, relevant evidence from independent reviewed layers.""" query_tokens = _query_tokens(answers if isinstance(answers, dict) else {}) if not query_tokens: return () entries = _merge_entry_layers( _load_entries(profile_root / "knowledge" / _MANIFEST_NAME), _load_curated_entries(profile_root / "knowledge" / _APPROVED_MANIFEST_NAME), ) remaining = [ (len(query_tokens & entry.tokens), entry) for entry in entries if _is_relevant(entry, query_tokens) ] selected: list[RetrievedKnowledge] = [] selected_topics: dict[str, int] = {} selected_sources: set[str] = set() selected_wordings: set[str] = set() while remaining and len(selected) < _MAX_EVIDENCE: eligible_indexes = [ index for index, (_, entry) in enumerate(remaining) if selected_topics.get(entry.topic, 0) < _TOPIC_LIMITS.get(entry.topic, _MAX_PER_TOPIC) and _selection_dedupe_key(entry) not in selected_wordings ] if not eligible_indexes: break best_index = min( eligible_indexes, key=lambda index: _selection_key(remaining[index], selected_sources), ) _, entry = remaining.pop(best_index) selected.append(_as_retrieved(entry)) selected_topics[entry.topic] = selected_topics.get(entry.topic, 0) + 1 selected_sources.update(entry.source_platforms) selected_wordings.add(_selection_dedupe_key(entry)) return tuple(selected) def _as_retrieved(entry: _KnowledgeEntry) -> RetrievedKnowledge: return RetrievedKnowledge( source=entry.source, topic=entry.topic, source_url=entry.source_url, text=entry.text, evidence_id=entry.evidence_id, evidence_kind=entry.evidence_kind, confidence=entry.confidence, ) def _query_tokens(answers: dict[str, str]) -> frozenset[str]: """Expand structured check-in fields into Korean coaching concepts.""" terms: list[str] = [] for field, value in answers.items(): if not isinstance(value, str): continue if isinstance(field, str) and field not in _NO_FIELD_EXPANSION: terms.extend(_FIELD_TERMS.get(field, ())) terms.append(value) return _token_set(" ".join(terms)) def _is_relevant(entry: _KnowledgeEntry, query_tokens: frozenset[str]) -> bool: """Require wording overlap and preserve stricter boundaries for exact excerpts.""" if not query_tokens & entry.tokens: return False if entry.evidence_kind == "exact_source_excerpt" and entry.topic in _RUNTIME_EXCLUDED_TOPICS: return False required = _TOPIC_REQUIRED_TERMS.get(entry.topic) return required is None or bool(query_tokens & required) def _selection_key( scored_entry: tuple[int, _KnowledgeEntry], selected_sources: set[str], ) -> tuple[float, int, int, int, float, float, str]: relevance, entry = scored_entry source_seen = bool(entry.source_platforms & selected_sources) return ( -relevance, -entry.source_priority, -_EVIDENCE_GRADE_RANK[entry.evidence_grade], 1 if source_seen else 0, -_recency_score(entry.reviewed_at_kst), -_recency_score(entry.extracted_at_kst), entry.evidence_id, ) def _load_entries(path: Path) -> tuple[_KnowledgeEntry, ...]: """Load only a valid, reviewed runtime manifest; never fall back to raw catalogs.""" try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError): return () if not isinstance(payload, dict): return () if payload.get("schema_version") != _SCHEMA_VERSION: return () generated_at = payload.get("generated_at_kst") raw_entries = payload.get("entries") if not isinstance(generated_at, str) or not generated_at.strip() or not isinstance(raw_entries, list): return () by_canonical_id: dict[str, _KnowledgeEntry] = {} for raw in raw_entries: entry = _parse_entry(raw) if entry is None: continue previous = by_canonical_id.get(entry.canonical_source_id) if previous is None or _entry_quality_key(entry) > _entry_quality_key(previous): by_canonical_id[entry.canonical_source_id] = entry return tuple(by_canonical_id[canonical_id] for canonical_id in sorted(by_canonical_id)) def _load_curated_entries(path: Path) -> tuple[_KnowledgeEntry, ...]: """Load only approved bounded clusters and rewrites from the curated artifact.""" try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError): return () if not _valid_approved_payload(payload): return () generated_at = payload["generated_at_kst"] entries: list[_KnowledgeEntry] = [] for raw in payload["approved_clusters"]: entry = _parse_curated_cluster(raw, generated_at) if entry is not None: entries.append(entry) for raw in payload["approved_safe_replacements"]: entry = _parse_safe_replacement(raw, generated_at) if entry is not None: entries.append(entry) return _deduplicate_entries(entries) def _valid_approved_payload(payload: object) -> bool: if not isinstance(payload, dict) or payload.get("schema_version") != _SCHEMA_VERSION: return False generated_at = payload.get("generated_at_kst") clusters = payload.get("approved_clusters") replacements = payload.get("approved_safe_replacements") invariants = payload.get("invariants") if ( not isinstance(generated_at, str) or not generated_at.strip() or not isinstance(clusters, list) or not isinstance(replacements, list) or not isinstance(invariants, dict) or any(invariants.get(name) is not True for name in _APPROVED_INVARIANTS) ): return False for key, expected in ( ("approved_cluster_count", len(clusters)), ("approved_safe_replacement_count", len(replacements)), ): value = payload.get(key) if value is not None and (isinstance(value, bool) or not isinstance(value, int) or value != expected): return False accounting = payload.get("safe_replacement_accounting") if accounting is not None: if not isinstance(accounting, dict): return False if ( accounting.get("quarantined_original_count") is not None and accounting.get("quarantined_original_count") != len(replacements) ): return False if ( accounting.get("approved_safe_rewrite_count") is not None and accounting.get("approved_safe_rewrite_count") != len(replacements) ): return False if accounting.get("terminal_identity_counts_unchanged") is not True: return False return True def _parse_curated_cluster(raw: object, generated_at: str) -> _KnowledgeEntry | None: if not isinstance(raw, dict) or _has_forbidden_curated_fields(raw): return None cluster_id = raw.get("cluster_id") bounded_wording = _bounded_wording(raw.get("bounded_wording")) claim_type = raw.get("claim_type") topic = raw.get("topic") safety_class = raw.get("safety_class") member_source_ids = raw.get("member_source_ids") source_urls = raw.get("source_urls") provenance = raw.get("provenance") if ( not isinstance(cluster_id, str) or _APPROVED_CLUSTER_ID.fullmatch(cluster_id.strip()) is None or bounded_wording is None or not isinstance(claim_type, str) or claim_type.strip() not in _CURATED_CLAIM_TYPES or not isinstance(topic, str) or not isinstance(safety_class, str) or not isinstance(member_source_ids, list) or not isinstance(source_urls, list) or len(member_source_ids) == 0 or len(member_source_ids) != len(source_urls) or not _valid_provenance(provenance) ): return None topic = topic.strip() safety_class = safety_class.strip().casefold() if topic not in _CURATED_ALLOWED_TOPICS: return None if safety_class not in _RUNTIME_SAFETY_CLASSES: return None normalized_sources: list[str] = [] for source_id in member_source_ids: if not isinstance(source_id, str) or _SOURCE_ID.fullmatch(source_id.strip()) is None: return None normalized_sources.append(source_id.strip()) normalized_urls: list[str] = [] for source_url in source_urls: if not _valid_http_url(source_url): return None normalized_urls.append(source_url.strip()) if len(set(source_id.casefold() for source_id in normalized_sources)) != len(normalized_sources): return None if len(set(url.casefold() for url in normalized_urls)) != len(normalized_urls): return None source_platforms = frozenset(source_id.split(":", 1)[0].casefold() for source_id in normalized_sources) source = ",".join(sorted(source_platforms)) source_url = sorted(normalized_urls, key=str.casefold)[0] return _KnowledgeEntry( canonical_source_id=cluster_id.strip(), evidence_id=cluster_id.strip(), evidence_kind="curated_bounded_principle", source=source, source_platforms=source_platforms, topic=topic, source_url=source_url, text=bounded_wording, source_digest="", extracted_at_kst=generated_at.strip(), reviewed_at_kst=generated_at.strip(), evidence_grade="C", safety_class=safety_class, truncated=False, confidence="curated_approved", source_priority=_CURATED_KIND_PRIORITY["curated_bounded_principle"], tokens=_token_set(bounded_wording), ) def _parse_safe_replacement(raw: object, generated_at: str) -> _KnowledgeEntry | None: if not isinstance(raw, dict) or _has_forbidden_curated_fields(raw): return None replacement_id = raw.get("replacement_id") original_id = raw.get("quarantined_canonical_source_id") bounded_wording = _bounded_wording(raw.get("bounded_wording")) topic = raw.get("topic") source_url = raw.get("source_url") approval_status = raw.get("approval_status") original_claim_status = raw.get("original_claim_status") rationale = raw.get("rationale") citations = raw.get("citations") if ( not isinstance(replacement_id, str) or _SAFE_REWRITE_ID.fullmatch(replacement_id.strip()) is None or not isinstance(original_id, str) or _SOURCE_ID.fullmatch(original_id.strip()) is None or bounded_wording is None or not isinstance(topic, str) or not isinstance(approval_status, str) or approval_status.strip() != "approved_safe_rewrite" or not isinstance(original_claim_status, str) or original_claim_status.strip() != "quarantined" or not _valid_http_url(source_url) or not isinstance(rationale, str) or not rationale.strip() or not isinstance(citations, list) or not citations or not all(_valid_citation(citation) for citation in citations) ): return None topic = topic.strip() if topic not in _CURATED_ALLOWED_TOPICS: return None provided_safety = raw.get("safety_class") if provided_safety is not None and ( not isinstance(provided_safety, str) or provided_safety.strip().casefold() not in _RUNTIME_SAFETY_CLASSES ): return None safety_class = ( provided_safety.strip().casefold() if isinstance(provided_safety, str) else _CURATED_TOPIC_SAFETY.get(topic) ) if safety_class is None or _CURATED_TOPIC_SAFETY.get(topic) not in {safety_class, None}: return None source_platform = original_id.strip().split(":", 1)[0].casefold() return _KnowledgeEntry( canonical_source_id=replacement_id.strip(), evidence_id=replacement_id.strip(), evidence_kind="approved_safe_rewrite", source=source_platform, source_platforms=frozenset({source_platform}), topic=topic, source_url=source_url.strip(), text=bounded_wording, source_digest="", extracted_at_kst=generated_at.strip(), reviewed_at_kst=generated_at.strip(), evidence_grade="C", safety_class=safety_class, truncated=False, confidence="safe_rewrite_approved", source_priority=_CURATED_KIND_PRIORITY["approved_safe_rewrite"], tokens=_token_set(bounded_wording), ) def _bounded_wording(value: object) -> str | None: if not isinstance(value, str): return None compact = " ".join(value.split()) if not compact or len(compact) > _MAX_TEXT_LENGTH: return None return compact def _valid_provenance(value: object) -> bool: if not isinstance(value, list) or not value: return False for item in value: if not isinstance(item, dict): return False kind = item.get("kind") if not isinstance(kind, str) or kind.strip() not in _CURATED_PROVENANCE_KINDS: return False citations = item.get("citations") if citations is not None and ( not isinstance(citations, list) or not all(_valid_citation(citation) for citation in citations) ): return False if kind.strip() == "independent_review" and not isinstance(citations, list): return False if kind.strip() == "independent_review": review_source = item.get("review_source") if not isinstance(review_source, str) or not review_source.strip(): return False if kind.strip() == "semantic_extraction": if not any( isinstance(item.get(key), str) and item.get(key).strip() for key in ("batch", "claim_id") ): return False return True def _valid_citation(value: object) -> bool: if isinstance(value, str): return bool(value.strip()) if not isinstance(value, dict) or not value: return False if not any(isinstance(item, str) and item.strip() for item in value.values()): return False for key in ("url", "source_url"): if key in value and not _valid_http_url(value[key]): return False return True def _valid_http_url(value: object) -> bool: if not isinstance(value, str): return False candidate = value.strip() if not candidate or any(character.isspace() for character in candidate): return False parsed = urlparse(candidate) return parsed.scheme.lower() in {"http", "https"} and bool(parsed.netloc) def _has_forbidden_curated_fields(raw: dict[object, object]) -> bool: return any(key in raw for key in _FORBIDDEN_CURATED_KEYS) def _deduplicate_entries(entries: tuple[_KnowledgeEntry, ...] | list[_KnowledgeEntry]) -> tuple[_KnowledgeEntry, ...]: """Deduplicate malformed repeated evidence IDs while retaining curated cluster identity.""" by_id: dict[str, _KnowledgeEntry] = {} for entry in entries: previous = by_id.get(entry.evidence_id) if previous is None or _entry_quality_key(entry) > _entry_quality_key(previous): by_id[entry.evidence_id] = entry return tuple(sorted(by_id.values(), key=lambda entry: entry.evidence_id)) def _merge_entry_layers( runtime_entries: tuple[_KnowledgeEntry, ...], curated_entries: tuple[_KnowledgeEntry, ...], ) -> tuple[_KnowledgeEntry, ...]: """Keep exact runtime rows intact while collapsing redundant curated wording.""" runtime_ids = {entry.evidence_id for entry in runtime_entries} runtime_wordings = {_normalized_wording(entry.text) for entry in runtime_entries} return ( *runtime_entries, *( entry for entry in curated_entries if entry.evidence_id not in runtime_ids and _normalized_wording(entry.text) not in runtime_wordings ), ) def _parse_entry(raw: object) -> _KnowledgeEntry | None: """Accept one fully attributed, safety-reviewed runtime row at the JSON boundary.""" if not isinstance(raw, dict): return None canonical_source_id = raw.get("canonical_source_id") source = raw.get("source") topic = raw.get("topic") source_url = raw.get("source_url") text = raw.get("text") source_digest = raw.get("source_digest") extracted_at_kst = raw.get("extracted_at_kst") reviewed_at_kst = raw.get("reviewed_at_kst") evidence_grade = raw.get("evidence_grade") safety_class = raw.get("safety_class") if not all( isinstance(value, str) and value.strip() for value in ( canonical_source_id, source, topic, source_url, text, source_digest, extracted_at_kst, reviewed_at_kst, evidence_grade, safety_class, ) ): return None if raw.get("runtime_eligible") is not True or not isinstance(raw.get("truncated"), bool): return None if not _valid_http_url(source_url): return None if _has_rejected_row_status(raw): return None compact = " ".join(text.split()) title = raw.get("title") if isinstance(title, str) and compact == " ".join(title.split()): return None normalized_grade = _normalize_evidence_grade(evidence_grade) normalized_safety = _normalize_safety_class(safety_class, topic, compact) if normalized_grade is None or normalized_safety is None: return None tokens = _token_set(compact) return _KnowledgeEntry( canonical_source_id=canonical_source_id.strip(), evidence_id=canonical_source_id.strip(), evidence_kind="exact_source_excerpt", source=source.strip(), source_platforms=frozenset({source.strip().casefold()}), topic=topic.strip(), source_url=source_url.strip(), text=compact[:_MAX_TEXT_LENGTH], source_digest=source_digest.strip(), extracted_at_kst=extracted_at_kst.strip(), reviewed_at_kst=reviewed_at_kst.strip(), evidence_grade=normalized_grade, safety_class=normalized_safety, truncated=raw["truncated"], confidence="exact_digest_reviewed", source_priority=_CURATED_KIND_PRIORITY["exact_source_excerpt"], tokens=tokens, ) def _has_rejected_row_status(raw: dict[object, object]) -> bool: """Reject catalog/enrichment placeholders even when a malformed row claims eligibility.""" if raw.get("available") is False: return True for key in ("status", "availability", "content_status"): value = raw.get(key) if isinstance(value, str) and value.strip().casefold() in _REJECTED_ROW_STATUSES: return True return False def _normalize_evidence_grade(value: str) -> str | None: normalized = value.strip() if normalized.casefold() in {"owner-reviewed-extracted", "primary_source_extracted"}: return "primary_source_extracted" if normalized.upper() in {"A", "B", "C"}: return normalized.upper() return None def _normalize_safety_class(value: str, topic: str, text: str) -> str | None: normalized = value.strip().casefold() if normalized in _RUNTIME_SAFETY_CLASSES: return normalized if normalized != "general": return None context = f"{topic} {text}".casefold() if any(term in context for term in ("소화", "장", "복부", "변비", "설사", "식욕", "fodmap")): return "digestive_observation" if any(term in context for term in ("수면", "회복", "피로", "컨디션")): return "recovery" if any(term in context for term in ("칼로리", "식단", "영양", "체중", "감량", "증량", "벌크", "탄수화물", "단백질", "지방")): return "general_nutrition" return "general_training" def _entry_quality_key(entry: _KnowledgeEntry) -> tuple[int, int, float, float, str, str, str, str, str]: """Choose duplicate rows without depending on manifest or catalog order.""" return ( entry.source_priority, _EVIDENCE_GRADE_RANK[entry.evidence_grade], _recency_score(entry.reviewed_at_kst), _recency_score(entry.extracted_at_kst), entry.source.casefold(), entry.topic, entry.source_url, entry.text, entry.evidence_id, ) def _selection_dedupe_key(entry: _KnowledgeEntry) -> str: if entry.evidence_kind == "exact_source_excerpt": return entry.evidence_id return _normalized_wording(entry.text) def _normalized_wording(value: str) -> str: return " ".join(normalize("NFKC", value).split()).casefold() def _token_set(text: str) -> frozenset[str]: tokens: set[str] = set() for token in _TOKEN.findall(text): normalized = token.casefold() tokens.add(normalized) if _HANGUL.search(normalized) is not None: for suffix in _KOREAN_SUFFIXES: if normalized.endswith(suffix) and len(normalized) - len(suffix) >= 2: tokens.add(normalized[:-len(suffix)]) break return frozenset(tokens) def _recency_score(value: str) -> float: """Return a comparable timestamp while treating malformed nonempty metadata as oldest.""" try: parsed = datetime.fromisoformat(value) except (TypeError, ValueError, OverflowError): return float("-inf") if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) try: return parsed.timestamp() except (OverflowError, OSError, ValueError): return float("-inf")