from __future__ import annotations import math import re import uuid from datetime import date from typing import Any, Literal, NotRequired, TypedDict, cast DatePrecision = Literal["day", "month", "quarter", "year", "range"] EVENT_CONTRACT_VERSION = "rectification-event-contract-v2" EVENT_KINDS: dict[str, frozenset[str]] = { "education": frozenset({ "education_start", "education_completion", "education_interruption", "education_change", "education_milestone", # v1 compatibility }), "career": frozenset({ "career_entry", "career_change", "promotion", "career_pressure", "career_exit", "business_start", }), "relationship": frozenset({ "relationship_start", "relationship_commitment", "relationship_separation", "relationship_end", "relationship_change", # v1 compatibility }), "relocation": frozenset({"relocation", "foreign_move", "return", "home_change"}), "finance": frozenset({"finance_gain", "finance_loss", "income_change", "asset_change", "finance_change"}), "health": frozenset({"self_health_event", "pressure_period"}), "health_pressure": frozenset({"self_health_event", "pressure_period"}), # v1 domain compatibility "family": frozenset({"family_event"}), "other": frozenset({"other"}), } BACKGROUND_EVENT_KINDS = frozenset({"family_event", "other"}) SCOREABLE_EVENT_KINDS: dict[str, frozenset[str]] = { domain: frozenset(kind for kind in kinds if kind not in BACKGROUND_EVENT_KINDS) for domain, kinds in EVENT_KINDS.items() if any(kind not in BACKGROUND_EVENT_KINDS for kind in kinds) } DATE_PRECISIONS = frozenset({"day", "month", "quarter", "year", "range"}) _BIRTH_TIME_SOURCES = frozenset({"hospital_record", "family_exact", "approximate", "period_only", "unknown", "legacy_import"}) _LOCAL_TIME_STATUSES = frozenset({"resolved", "not_provided", "ambiguous", "nonexistent"}) _REQUEST_PROVENANCE_FIELDS = frozenset({"birth_time_source", "timezone_id", "timezone_source", "local_time_status"}) _EVENT_PROVENANCE_FIELDS = frozenset({ "date_source", "date_reliability", "date_corroboration", "date_conflict_status", "source_turn_id", "subject", }) _REQUEST_FIELDS = frozenset({"birth_date", "start_time", "end_time", "lat", "lon", "tz", "events"}) | _REQUEST_PROVENANCE_FIELDS _EVENT_FIELDS = frozenset({"id", "domain", "event_kind", "date_start", "date_end", "precision", "summary"}) | _EVENT_PROVENANCE_FIELDS _CLOCK = re.compile(r"(?:[01]\d|2[0-3]):[0-5]\d\Z") class LifeEvent(TypedDict): id: str domain: str event_kind: str date_start: str date_end: str precision: DatePrecision summary: NotRequired[str] date_source: NotRequired[str | None] date_reliability: NotRequired[str | None] date_corroboration: NotRequired[str | None] date_conflict_status: NotRequired[str | None] source_turn_id: NotRequired[str | None] subject: NotRequired[Literal["self", "family", "other"]] class RectificationRequest(TypedDict): birth_date: str start_time: str end_time: str lat: float lon: float tz: float events: list[LifeEvent] birth_time_source: NotRequired[str | None] timezone_id: NotRequired[str | None] timezone_source: NotRequired[str | None] local_time_status: NotRequired[str | None] JsonObject = dict[str, Any] def is_scoreable_event(event: LifeEvent) -> bool: return event["event_kind"] not in BACKGROUND_EVENT_KINDS def _bounded_number(body: dict[str, Any], name: str, minimum: float, maximum: float) -> float: value = body.get(name) if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(float(value)): raise ValueError(f"{name} must be a finite number") result = float(value) if not minimum <= result <= maximum: raise ValueError(f"{name} must be between {minimum:g} and {maximum:g}") return result def _calendar_date(value: Any, label: str) -> date: if not isinstance(value, str): raise ValueError(f"{label} must be a valid YYYY-MM-DD value") try: return date.fromisoformat(value) except ValueError as exc: raise ValueError(f"{label} must be a valid YYYY-MM-DD value") from exc def _copy_nullable_text( source: dict[str, Any], target: dict[str, Any], name: str, label: str, maximum: int, allowed: frozenset[str] | None = None, ) -> None: if name not in source: return value = source[name] if value is None: target[name] = None return if not isinstance(value, str) or not value.strip() or len(value.strip()) > maximum: raise ValueError(f"{label} must be null or a non-empty string up to {maximum} characters") cleaned = value.strip() if allowed is not None and cleaned not in allowed: raise ValueError(f"{label} is invalid") target[name] = cleaned def normalize_rectification_request(body: Any, *, today: date | None = None) -> RectificationRequest: if not isinstance(body, dict): raise ValueError("request body must be an object") unsupported = sorted(set(body) - _REQUEST_FIELDS) if unsupported: raise ValueError(f"unsupported rectification field: {unsupported[0]}") birth_day = _calendar_date(body.get("birth_date"), "birth_date") start_time, end_time = body.get("start_time"), body.get("end_time") if not isinstance(start_time, str) or not _CLOCK.fullmatch(start_time): raise ValueError("start_time must be HH:MM") if not isinstance(end_time, str) or not _CLOCK.fullmatch(end_time): raise ValueError("end_time must be HH:MM") events = body.get("events") if not isinstance(events, list) or not 1 <= len(events) <= 100: raise ValueError("events must contain between 1 and 100 items") upper_date = today or date.today() cleaned_events: list[LifeEvent] = [] for index, raw_event in enumerate(events): if not isinstance(raw_event, dict): raise ValueError(f"events[{index}] must be an object") unsupported_event_fields = sorted(set(raw_event) - _EVENT_FIELDS) if unsupported_event_fields: raise ValueError(f"events[{index}] contains unsupported field: {unsupported_event_fields[0]}") try: event_id = str(uuid.UUID(str(raw_event.get("id") or ""))) except (ValueError, AttributeError) as exc: raise ValueError(f"events[{index}].id must be a UUID") from exc domain, event_kind = raw_event.get("domain"), raw_event.get("event_kind") if domain not in EVENT_KINDS: raise ValueError(f"events[{index}].domain is invalid") if event_kind not in EVENT_KINDS[cast(str, domain)]: raise ValueError(f"events[{index}].event_kind does not match domain") precision = raw_event.get("precision") if precision not in DATE_PRECISIONS: raise ValueError(f"events[{index}].precision is invalid") start_day = _calendar_date(raw_event.get("date_start"), f"events[{index}].date_start") end_day = _calendar_date(raw_event.get("date_end"), f"events[{index}].date_end") if start_day > end_day: raise ValueError(f"events[{index}].date_start must not exceed date_end") if start_day < birth_day or end_day > upper_date: raise ValueError(f"events[{index}] dates must be between birth_date and today") summary = raw_event.get("summary", "") if not isinstance(summary, str) or len(summary) > 1_000: raise ValueError(f"events[{index}].summary must be a string up to 1000 characters") subject = raw_event.get("subject") if subject is None: subject = "family" if domain == "family" else "other" if domain == "other" else "self" if subject not in {"self", "family", "other"}: raise ValueError(f"events[{index}].subject is invalid") if event_kind not in BACKGROUND_EVENT_KINDS and subject != "self": raise ValueError(f"events[{index}].subject must be self for scoreable events") cleaned_event: dict[str, Any] = { "id": event_id, "domain": cast(str, domain), "event_kind": cast(str, event_kind), "date_start": start_day.isoformat(), "date_end": end_day.isoformat(), "precision": cast(DatePrecision, precision), "summary": summary.strip(), "subject": cast(Literal["self", "family", "other"], subject), } _copy_nullable_text(raw_event, cleaned_event, "date_source", f"events[{index}].date_source", 120) _copy_nullable_text(raw_event, cleaned_event, "date_reliability", f"events[{index}].date_reliability", 120) _copy_nullable_text(raw_event, cleaned_event, "date_corroboration", f"events[{index}].date_corroboration", 1_000) _copy_nullable_text(raw_event, cleaned_event, "date_conflict_status", f"events[{index}].date_conflict_status", 120) if "source_turn_id" in raw_event: source_turn_id = raw_event.get("source_turn_id") if source_turn_id is None: cleaned_event["source_turn_id"] = None else: try: cleaned_event["source_turn_id"] = str(uuid.UUID(str(source_turn_id))) except (ValueError, AttributeError) as exc: raise ValueError(f"events[{index}].source_turn_id must be null or a UUID") from exc cleaned_events.append(cast(LifeEvent, cleaned_event)) cleaned_request: dict[str, Any] = { "birth_date": birth_day.isoformat(), "start_time": start_time, "end_time": end_time, "lat": _bounded_number(body, "lat", -90, 90), "lon": _bounded_number(body, "lon", -180, 180), "tz": _bounded_number(body, "tz", -14, 14), "events": cleaned_events, } _copy_nullable_text(body, cleaned_request, "birth_time_source", "birth_time_source", 120, _BIRTH_TIME_SOURCES) _copy_nullable_text(body, cleaned_request, "timezone_id", "timezone_id", 120) _copy_nullable_text(body, cleaned_request, "timezone_source", "timezone_source", 80) _copy_nullable_text(body, cleaned_request, "local_time_status", "local_time_status", 120, _LOCAL_TIME_STATUSES) return cast(RectificationRequest, cleaned_request)