Jesse_Chen
171 lines
7.9 KiB
Python
171 lines
7.9 KiB
Python
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"]
|
|
|
|
SCOREABLE_EVENT_KINDS: dict[str, frozenset[str]] = {
|
|
"education": frozenset({"education_milestone"}),
|
|
"relocation": frozenset({"relocation"}),
|
|
"relationship": frozenset({"relationship_start", "relationship_end", "relationship_change"}),
|
|
"career": frozenset({"career_change"}),
|
|
"finance": frozenset({"finance_change"}),
|
|
"health_pressure": frozenset({"self_health_event"}),
|
|
}
|
|
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"})
|
|
_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]
|
|
|
|
|
|
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 _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 SCOREABLE_EVENT_KINDS:
|
|
raise ValueError(f"events[{index}].domain is not scoreable")
|
|
if event_kind not in SCOREABLE_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")
|
|
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(),
|
|
}
|
|
_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)
|
|
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)
|