MapMil/contracts/heuristic_profile.py

"""Declarative heuristic profile: schema + static rule interpreter (CA preview + CP runtime)."""

from __future__ import annotations

import re
from typing import Any, Literal

from pydantic import BaseModel, Field, field_validator, model_validator

TARGET_FIELDS: tuple[str, ...] = (
    "title",
    "description",
    "locality",
    "event_date",
    "coords",
    "topic",
    "region",
)

COORDS_RE = re.compile(r"(-?\d{1,3}\.\d+)\s*,\s*(-?\d{1,3}\.\d+)")

Strategy = Literal["regex", "line", "after_marker", "between", "full_text", "literal"]


class FieldRule(BaseModel):
    strategy: Strategy
    # regex: named or group(1); line: 0-based index; after_marker/between: markers
    pattern: str | None = None
    group: int = 1
    line_index: int | None = None
    marker: str | None = None
    end_marker: str | None = None
    value: str | None = None  # literal
    flags: str = ""  # e.g. "im" → re.I|re.M
    strip: bool = True

    @field_validator("pattern", "marker", "end_marker", "value", mode="before")
    @classmethod
    def empty_to_none(cls, value: Any) -> Any:
        if value is None:
            return None
        if isinstance(value, str) and not value.strip():
            return None
        return value


class HeuristicProfile(BaseModel):
    version: Literal[1] = 1
    fields: dict[str, FieldRule] = Field(default_factory=dict)
    required_fields: list[str] = Field(default_factory=list)
    notes: str = ""

    @field_validator("required_fields")
    @classmethod
    def known_required(cls, value: list[str]) -> list[str]:
        seen: list[str] = []
        unknown: list[str] = []
        for name in value:
            if name not in TARGET_FIELDS:
                unknown.append(name)
                continue
            if name not in seen:
                seen.append(name)
        if unknown:
            raise ValueError(f"Unknown required_fields: {sorted(set(unknown))}")
        return seen

    @model_validator(mode="after")
    def known_fields_only(self) -> "HeuristicProfile":
        unknown = set(self.fields) - set(TARGET_FIELDS)
        if unknown:
            raise ValueError(f"Unknown profile fields: {sorted(unknown)}")
        return self


def target_field_specs() -> list[dict[str, str]]:
    """Fixed target table for profile UI (roadmap: custom tables later)."""
    return [
        {"name": "title", "type": "string", "description": "Short event title"},
        {"name": "description", "type": "string", "description": "Event summary / body"},
        {"name": "locality", "type": "string", "description": "Place / settlement name"},
        {
            "name": "event_date",
            "type": "string",
            "description": "Date as DD.MM.YYYY or YYYY-MM-DD",
        },
        {
            "name": "coords",
            "type": "string",
            "description": "Latitude, longitude if present",
        },
        {"name": "topic", "type": "string", "description": "Short topic tag"},
        {"name": "region", "type": "string", "description": "Region (optional)"},
    ]


def _compile_flags(flags: str) -> int:
    mapping = {
        "i": re.IGNORECASE,
        "m": re.MULTILINE,
        "s": re.DOTALL,
    }
    result = 0
    for ch in (flags or "").lower():
        result |= mapping.get(ch, 0)
    return result


def _apply_rule(text: str, rule: FieldRule) -> str:
    raw = text or ""
    value = ""

    if rule.strategy == "literal":
        value = rule.value or ""
    elif rule.strategy == "full_text":
        value = raw
    elif rule.strategy == "line":
        lines = raw.splitlines()
        idx = 0 if rule.line_index is None else rule.line_index
        if 0 <= idx < len(lines):
            value = lines[idx]
    elif rule.strategy == "regex":
        if not rule.pattern:
            return ""
        match = re.search(rule.pattern, raw, _compile_flags(rule.flags))
        if match:
            try:
                value = match.group(rule.group)
            except IndexError:
                value = match.group(0)
    elif rule.strategy == "after_marker":
        marker = rule.marker or ""
        if not marker:
            return ""
        pos = raw.find(marker)
        if pos < 0:
            return ""
        start = pos + len(marker)
        rest = raw[start:]
        if rule.end_marker:
            end = rest.find(rule.end_marker)
            value = rest[:end] if end >= 0 else rest
        elif rule.pattern:
            match = re.search(rule.pattern, rest, _compile_flags(rule.flags))
            if match:
                try:
                    value = match.group(rule.group)
                except IndexError:
                    value = match.group(0)
        else:
            # first non-empty line after marker
            for line in rest.splitlines():
                if line.strip():
                    value = line
                    break
    elif rule.strategy == "between":
        start_m = rule.marker or ""
        end_m = rule.end_marker or ""
        if not start_m or not end_m:
            return ""
        start = raw.find(start_m)
        if start < 0:
            return ""
        start += len(start_m)
        end = raw.find(end_m, start)
        if end < 0:
            return ""
        value = raw[start:end]

    if rule.strip:
        value = value.strip()
    return value


def apply_profile(text: str, profile: HeuristicProfile | dict[str, Any]) -> dict[str, str]:
    """Apply static rules to post text → string field map (no LLM)."""
    if isinstance(profile, dict):
        profile = HeuristicProfile.model_validate(profile)

    result: dict[str, str] = {name: "" for name in TARGET_FIELDS}
    for name, rule in profile.fields.items():
        result[name] = _apply_rule(text, rule)
    return result


def parse_coords(raw: str) -> tuple[float | None, float | None]:
    match = COORDS_RE.search(raw or "")
    if not match:
        return None, None
    return float(match.group(1)), float(match.group(2))


def _field_is_filled(name: str, value: str) -> bool:
    if not (value or "").strip():
        return False
    if name == "coords":
        lat, lng = parse_coords(value)
        return lat is not None and lng is not None
    return True


def match_profile(
    text: str,
    profile: HeuristicProfile | dict[str, Any],
) -> tuple[bool, dict[str, str], list[str]]:
    """Apply profile and report whether required_fields are filled.

    Empty required_fields means no gate (legacy profiles ingest every post).
    """
    if isinstance(profile, dict):
        profile = HeuristicProfile.model_validate(profile)
    fields = apply_profile(text, profile)
    missing = [
        name
        for name in profile.required_fields
        if not _field_is_filled(name, fields.get(name, ""))
    ]
    return (not missing, fields, missing)