"""Immutable versioned profile contracts.""" from __future__ import annotations import hashlib import json import re from dataclasses import dataclass, field from enum import Enum from types import MappingProxyType from typing import Iterable, Mapping from ..domain.enums import ProfileEligibilityState from ..domain.models import ProfileVersion, RegionScopedIdentity from ..errors import ContractViolation _MAX_REGEX_LENGTH = 256 _MAX_RULES_PER_PROFILE = 4096 _MAX_OVERRIDES_PER_PROFILE = 512 _MAX_RULE_PRIORITY = 1000 _UNSAFE_REGEX_TOKENS = ( "(?", "\\1", "\\2", "\\3", "\\4", "\\5", "\\6", "\\7", "\\8", "\\9", ) _NESTED_QUANTIFIER = re.compile(r"\([^)]*[+*{][^)]*\)[+*{]") _REPEATED_GROUP = re.compile(r"\)[+*{]") def _nonblank(value: object, field_name: str) -> str: if not isinstance(value, str) or not value.strip(): raise ContractViolation(f"{field_name} must be a nonblank string") return value.strip() def _optional_text(value: object, field_name: str) -> str | None: if value is None: return None return _nonblank(value, field_name) def _text_tuple(values: Iterable[str], field_name: str) -> tuple[str, ...]: if isinstance(values, str): raise ContractViolation(f"{field_name} must be an iterable, not a string") return tuple(sorted({_nonblank(value, field_name) for value in values})) class ProfileLifecycleState(str, Enum): """Lifecycle of an immutable profile definition.""" DRAFT = "draft" ACTIVE = "active" RETIRED = "retired" class RuleEffect(str, Enum): """Whether a matching rule includes or excludes an object.""" INCLUDE = "include" EXCLUDE = "exclude" class RuleMatchMode(str, Enum): """Supported deterministic tag match modes.""" EXACT = "exact" REGEX = "regex" PRESENT = "present" def _enum_value(value: object, enum_type: type[Enum], field_name: str) -> Enum: if isinstance(value, enum_type): return value try: return enum_type(value) except (TypeError, ValueError) as exc: raise ContractViolation(f"{field_name} is not supported") from exc def _validate_regex(pattern: str) -> re.Pattern[str]: if len(pattern) > _MAX_REGEX_LENGTH: raise ContractViolation("regex pattern exceeds the bounded length") if any(token in pattern for token in _UNSAFE_REGEX_TOKENS): raise ContractViolation("regex pattern uses an unsupported construct") if _NESTED_QUANTIFIER.search(pattern) or _REPEATED_GROUP.search(pattern) or ".*.*" in pattern: raise ContractViolation("regex pattern contains an unsafe nested quantifier") try: return re.compile(pattern, re.IGNORECASE | re.ASCII) except re.error as exc: raise ContractViolation("regex pattern is invalid") from exc @dataclass(frozen=True, slots=True) class ProfileRule: """One immutable include or exclude rule over an OSM tag.""" rule_id: str key: str effect: RuleEffect match_mode: RuleMatchMode pattern: str | None = None category: str | None = None subcategory: str | None = None evidence: tuple[str, ...] = () priority: int = 0 _compiled_regex: re.Pattern[str] | None = field( init=False, default=None, repr=False, compare=False, ) def __post_init__(self) -> None: object.__setattr__(self, "rule_id", _nonblank(self.rule_id, "rule_id")) object.__setattr__(self, "key", _nonblank(self.key, "key").casefold()) effect = _enum_value(self.effect, RuleEffect, "effect") match_mode = _enum_value(self.match_mode, RuleMatchMode, "match_mode") object.__setattr__(self, "effect", effect) object.__setattr__(self, "match_mode", match_mode) priority = self.priority if ( isinstance(priority, bool) or not isinstance(priority, int) or priority < 0 or priority > _MAX_RULE_PRIORITY ): raise ContractViolation("priority must be an integer from 0 to 1000") if effect is RuleEffect.EXCLUDE and priority != 0: raise ContractViolation("exclude rules cannot define priority") object.__setattr__(self, "priority", priority) pattern = _optional_text(self.pattern, "pattern") if match_mode is RuleMatchMode.PRESENT: if pattern is not None: raise ContractViolation("present rules cannot define a pattern") elif pattern is None: raise ContractViolation("exact and regex rules require a pattern") elif match_mode is RuleMatchMode.EXACT: object.__setattr__(self, "pattern", pattern.casefold()) else: object.__setattr__(self, "_compiled_regex", _validate_regex(pattern)) object.__setattr__( self, "pattern", pattern if match_mode is not RuleMatchMode.EXACT else self.pattern ) category = _optional_text(self.category, "category") subcategory = _optional_text(self.subcategory, "subcategory") if effect is RuleEffect.INCLUDE and category is None: raise ContractViolation("include rules require a category") if effect is RuleEffect.EXCLUDE and (category is not None or subcategory is not None): raise ContractViolation("exclude rules cannot define category mappings") if subcategory is not None and category is None: raise ContractViolation("subcategory requires category") object.__setattr__(self, "category", category) object.__setattr__(self, "subcategory", subcategory) object.__setattr__(self, "evidence", _text_tuple(self.evidence, "evidence")) def matches(self, tags: Mapping[str, str]) -> bool: """Return whether the rule matches a normalized tag mapping.""" value = tags.get(self.key) if value is None: return False if self.match_mode is RuleMatchMode.PRESENT: return bool(value.strip()) if self.match_mode is RuleMatchMode.EXACT: return value.strip().casefold() == self.pattern compiled = self._compiled_regex if compiled is None: raise ContractViolation("compiled regex is unavailable") return compiled.search(value) is not None def canonical_data(self) -> dict[str, object]: return { "category": self.category, "effect": self.effect.value, "evidence": self.evidence, "key": self.key, "match_mode": self.match_mode.value, "pattern": self.pattern, "priority": self.priority, "rule_id": self.rule_id, "subcategory": self.subcategory, } @dataclass(frozen=True, slots=True) class RegionRuleOverride: """Region-specific disabled and additional profile rules.""" region_id: str disabled_rule_ids: tuple[str, ...] = () additional_rules: tuple[ProfileRule, ...] = () def __post_init__(self) -> None: object.__setattr__(self, "region_id", _nonblank(self.region_id, "region_id")) object.__setattr__( self, "disabled_rule_ids", _text_tuple(self.disabled_rule_ids, "disabled_rule_ids"), ) raw_rules = tuple(self.additional_rules) if any(not isinstance(rule, ProfileRule) for rule in raw_rules): raise ContractViolation("additional_rules must contain ProfileRule values") rules = tuple(sorted(raw_rules, key=lambda rule: rule.rule_id)) if len({rule.rule_id for rule in rules}) != len(rules): raise ContractViolation("additional rule IDs must be unique") object.__setattr__(self, "additional_rules", rules) def canonical_data(self) -> dict[str, object]: return { "additional_rules": tuple(rule.canonical_data() for rule in self.additional_rules), "disabled_rule_ids": self.disabled_rule_ids, "region_id": self.region_id, } @dataclass(frozen=True, slots=True) class ProfileDefinition: """One immutable versioned profile definition. Phase 2A.4 structurally rejects production insertion enablement. Profile evaluation is shadow/report-only even for an active definition. """ profile: ProfileVersion lifecycle_state: ProfileLifecycleState rules: tuple[ProfileRule, ...] provenance: tuple[str, ...] overrides: tuple[RegionRuleOverride, ...] = () approval_references: tuple[str, ...] = () shadow_only: bool = False production_insert_enabled: bool = False definition_digest: str = field(init=False) def __post_init__(self) -> None: if not isinstance(self.profile, ProfileVersion): raise ContractViolation("profile must be a ProfileVersion") lifecycle = _enum_value( self.lifecycle_state, ProfileLifecycleState, "lifecycle_state", ) object.__setattr__(self, "lifecycle_state", lifecycle) if not isinstance(self.shadow_only, bool): raise ContractViolation("shadow_only must be a boolean") if not isinstance(self.production_insert_enabled, bool): raise ContractViolation("production_insert_enabled must be a boolean") if self.production_insert_enabled: raise ContractViolation("production inserts are unavailable in Phase 2A.4") raw_rules = tuple(self.rules) if not raw_rules: raise ContractViolation("profile must contain at least one rule") if len(raw_rules) > _MAX_RULES_PER_PROFILE: raise ContractViolation("profile exceeds the bounded rule count") if any(not isinstance(rule, ProfileRule) for rule in raw_rules): raise ContractViolation("rules must contain ProfileRule values") rules = tuple(sorted(raw_rules, key=lambda rule: rule.rule_id)) rule_ids = {rule.rule_id for rule in rules} if len(rule_ids) != len(rules): raise ContractViolation("profile rule IDs must be unique") object.__setattr__(self, "rules", rules) raw_overrides = tuple(self.overrides) if len(raw_overrides) > _MAX_OVERRIDES_PER_PROFILE: raise ContractViolation("profile exceeds the bounded override count") if any(not isinstance(override, RegionRuleOverride) for override in raw_overrides): raise ContractViolation("overrides must contain RegionRuleOverride values") overrides = tuple(sorted(raw_overrides, key=lambda override: override.region_id)) if len({override.region_id for override in overrides}) != len(overrides): raise ContractViolation("profile region overrides must be unique") for override in overrides: unknown = set(override.disabled_rule_ids) - rule_ids if unknown: raise ContractViolation("region override disables an unknown rule") added_ids = {rule.rule_id for rule in override.additional_rules} if rule_ids & added_ids: raise ContractViolation("additional rule IDs cannot shadow base rules") object.__setattr__(self, "overrides", overrides) provenance = _text_tuple(self.provenance, "provenance") if not provenance: raise ContractViolation("profile provenance is required") approvals = _text_tuple(self.approval_references, "approval_references") if lifecycle is ProfileLifecycleState.ACTIVE and not self.shadow_only and not approvals: raise ContractViolation("active production profiles require approval references") object.__setattr__(self, "provenance", provenance) object.__setattr__(self, "approval_references", approvals) canonical = json.dumps( { "approval_references": approvals, "lifecycle_state": lifecycle.value, "overrides": tuple(override.canonical_data() for override in overrides), "profile_id": self.profile.profile_id, "profile_version": self.profile.version, "provenance": provenance, "rules": tuple(rule.canonical_data() for rule in rules), "shadow_only": self.shadow_only, }, sort_keys=True, separators=(",", ":"), ).encode("utf-8") object.__setattr__(self, "definition_digest", hashlib.sha256(canonical).hexdigest()) def effective_rules(self, region_id: str) -> tuple[ProfileRule, ...]: """Return deterministic base plus region-specific rules.""" region = _nonblank(region_id, "region_id") override = next((item for item in self.overrides if item.region_id == region), None) if override is None: return self.rules disabled = set(override.disabled_rule_ids) return ( tuple( sorted( (rule for rule in self.rules if rule.rule_id not in disabled), key=lambda rule: rule.rule_id, ) ) + override.additional_rules ) @property def production_insert_allowed(self) -> bool: return False @dataclass(frozen=True, slots=True) class ProfileDecision: """Deterministic report-only result of evaluating one source object.""" identity: RegionScopedIdentity profile: ProfileVersion state: ProfileEligibilityState definition_digest: str matched_rule_ids: tuple[str, ...] = () excluded_rule_ids: tuple[str, ...] = () category: str | None = None subcategory: str | None = None evidence: tuple[str, ...] = () production_insert_allowed: bool = False def __post_init__(self) -> None: if not isinstance(self.identity, RegionScopedIdentity): raise ContractViolation("identity must be RegionScopedIdentity") if not isinstance(self.profile, ProfileVersion): raise ContractViolation("profile must be ProfileVersion") try: state = ( self.state if isinstance(self.state, ProfileEligibilityState) else ProfileEligibilityState(self.state) ) except (TypeError, ValueError) as exc: raise ContractViolation("state is not supported") from exc object.__setattr__(self, "state", state) digest = _nonblank(self.definition_digest, "definition_digest") if not re.fullmatch(r"[0-9a-f]{64}", digest): raise ContractViolation("definition_digest must be lowercase SHA-256") object.__setattr__( self, "matched_rule_ids", _text_tuple(self.matched_rule_ids, "matched_rule_ids") ) object.__setattr__( self, "excluded_rule_ids", _text_tuple(self.excluded_rule_ids, "excluded_rule_ids") ) object.__setattr__(self, "category", _optional_text(self.category, "category")) object.__setattr__(self, "subcategory", _optional_text(self.subcategory, "subcategory")) object.__setattr__(self, "evidence", _text_tuple(self.evidence, "evidence")) if self.production_insert_allowed: raise ContractViolation("profile decisions cannot enable production inserts") if state is ProfileEligibilityState.ACTIVE and self.category is None: raise ContractViolation("active decisions require a category") if self.subcategory is not None and self.category is None: raise ContractViolation("subcategory requires category") @dataclass(frozen=True, slots=True) class ProfileImpactReport: """Bounded deterministic state/category counts for a profile dry run.""" profile: ProfileVersion definition_digest: str object_count: int state_counts: tuple[tuple[str, int], ...] category_counts: tuple[tuple[str, int], ...] def __post_init__(self) -> None: if not isinstance(self.profile, ProfileVersion): raise ContractViolation("profile must be ProfileVersion") if isinstance(self.object_count, bool) or not isinstance(self.object_count, int): raise ContractViolation("object_count must be a nonnegative integer") if self.object_count < 0: raise ContractViolation("object_count must be a nonnegative integer") digest = _nonblank(self.definition_digest, "definition_digest") if not re.fullmatch(r"[0-9a-f]{64}", digest): raise ContractViolation("definition_digest must be lowercase SHA-256") object.__setattr__(self, "definition_digest", digest) normalized_counts: dict[str, tuple[tuple[str, int], ...]] = {} for field_name in ("state_counts", "category_counts"): values = tuple(getattr(self, field_name)) for item in values: if not isinstance(item, tuple) or len(item) != 2: raise ContractViolation(f"{field_name} contains invalid count entries") key, count = item if ( not isinstance(key, str) or not key or isinstance(count, bool) or not isinstance(count, int) or count < 0 ): raise ContractViolation(f"{field_name} contains invalid counts") if len({key for key, _ in values}) != len(values): raise ContractViolation(f"{field_name} keys must be unique") normalized_counts[field_name] = tuple(sorted(values)) object.__setattr__(self, field_name, normalized_counts[field_name]) state_counts = dict(normalized_counts["state_counts"]) supported_states = {state.value for state in ProfileEligibilityState} if set(state_counts) - supported_states: raise ContractViolation("state_counts contains an unsupported eligibility state") if sum(state_counts.values()) != self.object_count: raise ContractViolation("state_counts must total object_count") active_count = state_counts.get(ProfileEligibilityState.ACTIVE.value, 0) if sum(dict(normalized_counts["category_counts"]).values()) != active_count: raise ContractViolation("category_counts must total the active state count") class ProfileRegistry: """Immutable exact-version profile registry with no implicit latest lookup.""" def __init__(self, definitions: Iterable[ProfileDefinition]) -> None: values = tuple(definitions) if any(not isinstance(item, ProfileDefinition) for item in values): raise ContractViolation("registry values must be ProfileDefinition instances") indexed: dict[tuple[str, str], ProfileDefinition] = {} for definition in values: key = (definition.profile.profile_id, definition.profile.version) if key in indexed: raise ContractViolation("registry contains a duplicate profile version") indexed[key] = definition self._definitions: Mapping[tuple[str, str], ProfileDefinition] = MappingProxyType(indexed) def get(self, profile: ProfileVersion) -> ProfileDefinition: if not isinstance(profile, ProfileVersion): raise ContractViolation("profile must be ProfileVersion") try: return self._definitions[(profile.profile_id, profile.version)] except KeyError: raise ContractViolation("profile version is not registered") from None def list_versions(self, profile_id: str | None = None) -> tuple[ProfileVersion, ...]: normalized = None if profile_id is None else _nonblank(profile_id, "profile_id") return tuple( definition.profile for _, definition in sorted(self._definitions.items()) if normalized is None or definition.profile.profile_id == normalized )