src/range_curate/_core/config.py

"""Strict configuration loader for the public news-relevance API."""

from __future__ import annotations

from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from types import MappingProxyType
from typing import Any

import yaml

from .models import ExposureFamily

_PACKAGE_ROOT = Path(__file__).resolve().parent
DEFAULT_CONFIG_PATH = _PACKAGE_ROOT / "configs" / "default.yml"
_TOP_LEVEL_KEYS = {
    "version",
    "mode",
    "catalogs",
    "normalization",
    "alias_policy_overrides",
    "context_rules",
    "market_wide_conjunctions",
    "exposure_families",
    "relationship_paths",
    "constituent_evidence_enabled",
    "licensed_adapters",
    "precedence",
    "relevant_k",
    "ranked_k",
}
_TIE_BREAKERS = (
    "accepted_tag_details_desc",
    "independent_detection_methods_desc",
    "article_timestamp_desc",
    "article_id_asc",
)


def _mapping(value: object, name: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping):
        raise TypeError(f"{name} must be a mapping")
    return value


def _exact_keys(value: Mapping[str, Any], allowed: set[str], name: str) -> None:
    unknown = set(value) - allowed
    if unknown:
        raise ValueError(f"unknown {name} keys: {sorted(unknown)}")


def _catalog_path(value: object, name: str, *, base_dir: Path) -> Path:
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"{name} must be a non-empty path")
    resolved = (base_dir / value).resolve()
    if not resolved.is_file():
        raise ValueError(f"{name} does not exist: {resolved}")
    return resolved


@dataclass(frozen=True, slots=True)
class ContextRule:
    window_chars: int
    accept_any_clauses: tuple[tuple[str, ...], ...]
    reject_any_terms: tuple[str, ...]


@dataclass(frozen=True, slots=True)
class MarketWideConjunctionRule:
    actor_phrases: tuple[str, ...]
    actor_identifiers: tuple[str, ...]
    action_phrases: tuple[str, ...]
    object_phrases: tuple[str, ...]
    scope_phrases: tuple[str, ...]
    entity_concept: str
    tag_type: str
    target_id: str
    event_family: str


@dataclass(frozen=True, slots=True)
class NewsRelevanceConfig:
    version: str
    mode: str
    entity_catalog_path: Path
    lexical_catalog_path: Path
    alias_policy_overrides: Mapping[str, Mapping[str, Any]]
    context_rules: Mapping[str, ContextRule]
    market_wide_conjunctions: Mapping[str, MarketWideConjunctionRule]
    exposure_members: Mapping[ExposureFamily, frozenset[str]]
    relationship_paths: Mapping[str, tuple[str, ...]]
    precedence: tuple[str, ...]
    accepted_link_statuses: frozenset[str]
    ranking_tiers: Mapping[str, int]
    ranking_budget: int
    constituent_evidence_enabled: bool


def load_config(path: str | Path | None = None) -> NewsRelevanceConfig:
    """Load a fail-closed configuration; unknown fields are rejected."""

    config_path = Path(path).resolve() if path is not None else DEFAULT_CONFIG_PATH
    catalog_root = config_path.parent if path is not None else _PACKAGE_ROOT
    with config_path.open(encoding="utf-8") as handle:
        root = yaml.safe_load(handle)
    root = _mapping(root, "config")
    _exact_keys(root, _TOP_LEVEL_KEYS, "top-level")
    if root.get("mode") != "catalog_lite":
        raise ValueError("only mode=catalog_lite is currently implemented")
    if root.get("constituent_evidence_enabled") is not False:
        raise ValueError("catalog_lite must fail closed on constituent evidence")
    adapters = _mapping(root.get("licensed_adapters"), "licensed_adapters")
    _exact_keys(adapters, {"crsp", "compustat"}, "licensed_adapters")
    if any(value is not False for value in adapters.values()):
        raise ValueError("licensed adapters are unavailable in the public core")

    catalogs = _mapping(root.get("catalogs"), "catalogs")
    _exact_keys(catalogs, {"entities", "lexical"}, "catalogs")
    normalization = _mapping(root.get("normalization"), "normalization")
    _exact_keys(normalization, {"version", "fuzzy_edits"}, "normalization")
    if normalization != {
        "version": "financial-phrase-nfkc-v2",
        "fuzzy_edits": False,
    }:
        raise ValueError("the conservative v2 normalization contract is required")

    overrides = _mapping(root.get("alias_policy_overrides"), "alias_policy_overrides")
    override_values: dict[str, Mapping[str, Any]] = {}
    for alias_id, raw in overrides.items():
        value = _mapping(raw, f"alias_policy_overrides.{alias_id}")
        _exact_keys(
            value, {"context_required", "context_rule_id"}, f"alias override {alias_id}"
        )
        if not isinstance(value.get("context_required"), bool):
            raise TypeError(f"alias override {alias_id} needs boolean context_required")
        override_values[str(alias_id)] = MappingProxyType(dict(value))

    rules = _mapping(root.get("context_rules"), "context_rules")
    context_rules: dict[str, ContextRule] = {}
    for rule_id, raw in rules.items():
        value = _mapping(raw, f"context_rules.{rule_id}")
        _exact_keys(
            value,
            {"window_chars", "accept_any_clauses", "reject_any_terms"},
            f"context rule {rule_id}",
        )
        width = value.get("window_chars")
        if isinstance(width, bool) or not isinstance(width, int) or width <= 0:
            raise ValueError(f"context rule {rule_id} needs positive window_chars")
        context_rules[str(rule_id)] = ContextRule(
            window_chars=width,
            accept_any_clauses=tuple(
                tuple(str(term) for term in clause)
                for clause in value.get("accept_any_clauses", ())
            ),
            reject_any_terms=tuple(
                str(term) for term in value.get("reject_any_terms", ())
            ),
        )

    raw_conjunctions = _mapping(
        root.get("market_wide_conjunctions"), "market_wide_conjunctions"
    )
    market_wide_conjunctions: dict[str, MarketWideConjunctionRule] = {}
    conjunction_keys = {
        "actor_phrases",
        "actor_identifiers",
        "action_phrases",
        "object_phrases",
        "scope_phrases",
        "entity_concept",
        "tag_type",
        "target_id",
        "event_family",
    }
    for rule_id, raw in raw_conjunctions.items():
        value = _mapping(raw, f"market_wide_conjunctions.{rule_id}")
        _exact_keys(value, conjunction_keys, f"market-wide conjunction {rule_id}")
        sequence_fields = (
            "actor_phrases",
            "actor_identifiers",
            "action_phrases",
            "object_phrases",
            "scope_phrases",
        )
        sequences: dict[str, tuple[str, ...]] = {}
        for field in sequence_fields:
            raw_values = value.get(field)
            if not isinstance(raw_values, list) or not raw_values:
                raise TypeError(
                    f"market-wide conjunction {rule_id}.{field} must be a non-empty list"
                )
            sequences[field] = tuple(str(item) for item in raw_values)
        scalar_values = {}
        for field in ("entity_concept", "tag_type", "target_id", "event_family"):
            raw_value = value.get(field)
            if not isinstance(raw_value, str) or not raw_value.strip():
                raise TypeError(
                    f"market-wide conjunction {rule_id}.{field} must be non-empty"
                )
            scalar_values[field] = raw_value
        if scalar_values["target_id"] != ExposureFamily.MARKET_WIDE.value:
            raise ValueError("market-wide conjunction target must be MARKET_WIDE")
        market_wide_conjunctions[str(rule_id)] = MarketWideConjunctionRule(
            **sequences,
            **scalar_values,
        )

    raw_families = _mapping(root.get("exposure_families"), "exposure_families")
    exposure_members: dict[ExposureFamily, frozenset[str]] = {}
    seen_members: set[str] = set()
    for family_name, raw in raw_families.items():
        family = ExposureFamily(str(family_name))
        value = _mapping(raw, f"exposure_families.{family_name}")
        _exact_keys(value, {"canonical_members"}, f"exposure family {family_name}")
        members = frozenset(
            str(member) for member in value.get("canonical_members", ())
        )
        if not members or seen_members & members:
            raise ValueError("exposure family members must be non-empty and disjoint")
        seen_members.update(members)
        exposure_members[family] = members

    raw_paths = _mapping(root.get("relationship_paths"), "relationship_paths")
    relationship_paths = {
        str(canonical_id): tuple(str(item) for item in path_value)
        for canonical_id, path_value in raw_paths.items()
    }
    if any(not path_value for path_value in relationship_paths.values()):
        raise ValueError("relationship paths cannot be empty")

    precedence = _mapping(root.get("precedence"), "precedence")
    _exact_keys(precedence, {"detection_methods"}, "precedence")
    methods = tuple(str(item) for item in precedence.get("detection_methods", ()))
    if not methods or len(methods) != len(set(methods)):
        raise ValueError("precedence detection methods must be unique and non-empty")

    relevant = _mapping(root.get("relevant_k"), "relevant_k")
    _exact_keys(relevant, {"accepted_link_statuses"}, "relevant_k")
    link_statuses = frozenset(
        str(item) for item in relevant.get("accepted_link_statuses", ())
    )
    if not link_statuses:
        raise ValueError("Relevant-K requires accepted link statuses")

    ranked = _mapping(root.get("ranked_k"), "ranked_k")
    _exact_keys(ranked, {"strategy", "budget", "tiers", "tie_breakers"}, "ranked_k")
    if ranked.get("strategy") != "per_bucket_exposure_lexicographic_v2":
        raise ValueError("unsupported Ranked-K strategy")
    if tuple(ranked.get("tie_breakers", ())) != _TIE_BREAKERS:
        raise ValueError("Ranked-K tie-breakers differ from the frozen contract")
    budget = ranked.get("budget")
    if isinstance(budget, bool) or not isinstance(budget, int) or budget <= 0:
        raise ValueError("ranked_k.budget must be a positive integer")
    tiers_raw = _mapping(ranked.get("tiers"), "ranked_k.tiers")
    tiers = {str(name): int(value) for name, value in tiers_raw.items()}
    if any(value < 0 for value in tiers.values()):
        raise ValueError("ranking tiers must be non-negative")
    missing_conjunction_tiers = {
        rule.tag_type
        for rule in market_wide_conjunctions.values()
        if rule.tag_type not in tiers
    }
    if missing_conjunction_tiers:
        raise ValueError(
            "market-wide conjunction tag types need ranking tiers: "
            f"{sorted(missing_conjunction_tiers)}"
        )

    version = root.get("version")
    if not isinstance(version, str) or not version.strip():
        raise ValueError("version must be non-empty")
    return NewsRelevanceConfig(
        version=version,
        mode="catalog_lite",
        entity_catalog_path=_catalog_path(
            catalogs.get("entities"), "catalogs.entities", base_dir=catalog_root
        ),
        lexical_catalog_path=_catalog_path(
            catalogs.get("lexical"), "catalogs.lexical", base_dir=catalog_root
        ),
        alias_policy_overrides=MappingProxyType(override_values),
        context_rules=MappingProxyType(context_rules),
        market_wide_conjunctions=MappingProxyType(market_wide_conjunctions),
        exposure_members=MappingProxyType(exposure_members),
        relationship_paths=MappingProxyType(relationship_paths),
        precedence=methods,
        accepted_link_statuses=link_statuses,
        ranking_tiers=MappingProxyType(tiers),
        ranking_budget=budget,
        constituent_evidence_enabled=False,
    )