"""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)