"""Typed Topic-59 card projection with no raw sidecar identifiers retained.""" from __future__ import annotations import hashlib import json from collections.abc import Mapping from dataclasses import dataclass from datetime import date from enum import StrEnum from typing import Final, Protocol from .nutrition_weekly_operations_identifiers import ( MAX_COMPLETED_FIELD_COUNT, OpaqueProjectionCustomerId, OpaqueProjectionDigest, OpaqueProjectionEventId, OpaqueProjectionSlot, OpaqueProjectionStatusId, TOPIC59_PROJECTION_SCHEMA, Topic59PrivacyError, opaque_customer_id, opaque_event_id, opaque_status_id, require_opaque_digest, topic59_payload_is_safe, ) class Topic59DayCardError(Topic59PrivacyError): """A card field is outside the finite Topic-59 projection contract.""" class Topic59DayState(StrEnum): SUBMITTED = "submitted" MISSED = "missed" LATE_SUBMITTED = "late_submitted" class Topic59DraftState(StrEnum): NOT_CREATED = "not_created" DRAFTED = "drafted" READY = "ready" class Topic59ApprovalState(StrEnum): NOT_REQUESTED = "not_requested" PENDING = "pending" APPROVED = "approved" class Topic59DeliveryState(StrEnum): NOT_SENT = "not_sent" HELD = "held" SENT_AUDITED = "sent_audited" class Topic59SidecarRow(Protocol): """Only the source fields allowed to cross this projection boundary.""" @property def customer_identity_digest(self) -> str: ... @property def kst_day(self) -> date: ... @property def state(self) -> str: ... @property def source_event_id(self) -> str | None: ... @property def logical_key(self) -> str: ... @property def row_digest(self) -> str: ... _DRAFT_LABELS: Final = { Topic59DraftState.NOT_CREATED: "미작성", Topic59DraftState.DRAFTED: "초안", Topic59DraftState.READY: "준비됨", } _APPROVAL_LABELS: Final = { Topic59ApprovalState.NOT_REQUESTED: "요청 전", Topic59ApprovalState.PENDING: "검토 대기", Topic59ApprovalState.APPROVED: "승인됨", } _DELIVERY_LABELS: Final = { Topic59DeliveryState.NOT_SENT: "미발송", Topic59DeliveryState.HELD: "보류", Topic59DeliveryState.SENT_AUDITED: "발송 확인", } _STATE_LABELS: Final = { Topic59DayState.SUBMITTED: "제출", Topic59DayState.MISSED: "미제출", Topic59DayState.LATE_SUBMITTED: "지연 제출", } _STATE_BY_VALUE: Final = {state.value: state for state in Topic59DayState} @dataclass(frozen=True, slots=True) class Topic59DayCard: customer_identity_digest: OpaqueProjectionCustomerId kst_day: date state: Topic59DayState completed_field_count: int draft_state: Topic59DraftState approval_state: Topic59ApprovalState delivery_state: Topic59DeliveryState event_id: OpaqueProjectionEventId status_row_id: OpaqueProjectionStatusId status_digest: OpaqueProjectionDigest candidate_digest: OpaqueProjectionDigest def __post_init__(self) -> None: if type(self.kst_day) is not date: raise Topic59DayCardError("KST day") if type(self.completed_field_count) is not int or not 0 <= self.completed_field_count <= MAX_COMPLETED_FIELD_COUNT: raise Topic59DayCardError("completed field count") if type(self.state) is not Topic59DayState or type(self.draft_state) is not Topic59DraftState: raise Topic59DayCardError("card enum") if type(self.approval_state) is not Topic59ApprovalState or type(self.delivery_state) is not Topic59DeliveryState: raise Topic59DayCardError("card enum") for value, reason in ( (self.customer_identity_digest, "customer identity"), (self.event_id, "event id"), (self.status_row_id, "status row id"), (self.status_digest, "status digest"), (self.candidate_digest, "candidate digest"), ): _ = require_opaque_digest(value, reason) @classmethod def from_sidecar( cls, row: Topic59SidecarRow, *, completed_field_count: int, draft_state: Topic59DraftState, approval_state: Topic59ApprovalState, delivery_state: Topic59DeliveryState, candidate_digest: str, ) -> Topic59DayCard: state = _STATE_BY_VALUE.get(str(row.state)) if state is None: raise Topic59DayCardError("day state") try: return cls( customer_identity_digest=opaque_customer_id(row.customer_identity_digest), kst_day=row.kst_day, state=state, completed_field_count=completed_field_count, draft_state=draft_state, approval_state=approval_state, delivery_state=delivery_state, event_id=opaque_event_id(row.source_event_id), status_row_id=opaque_status_id(row.logical_key), status_digest=require_opaque_digest(row.row_digest, "status digest"), candidate_digest=require_opaque_digest(candidate_digest, "candidate digest"), ) except Topic59PrivacyError as error: raise Topic59DayCardError("sidecar projection") from error def payload(self) -> dict[str, str | int]: return { "customer_identity_digest": self.customer_identity_digest, "kst_day": self.kst_day.isoformat(), "state": self.state.value, "completed_field_count": self.completed_field_count, "draft_state": _DRAFT_LABELS[self.draft_state], "approval_state": _APPROVAL_LABELS[self.approval_state], "delivery_state": _DELIVERY_LABELS[self.delivery_state], "event_id": self.event_id, "status_row_id": self.status_row_id, } def render(self) -> str: values = self.payload() return "\n".join( ( "[운영 현황]", f"고객 식별자: {values['customer_identity_digest']}", f"KST 날짜: {values['kst_day']}", f"제출 상태: {_STATE_LABELS[self.state]}", f"완료 필드 수: {values['completed_field_count']}", f"초안 상태: {values['draft_state']}", f"승인 상태: {values['approval_state']}", f"발송 상태: {values['delivery_state']}", f"이벤트 ID: {values['event_id']}", f"상태 행 ID: {values['status_row_id']}", ) ) @dataclass(frozen=True, slots=True) class Topic59Projection: card_slot: OpaqueProjectionSlot logical_key: OpaqueProjectionDigest customer_identity_digest: OpaqueProjectionCustomerId kst_day: date config_digest: OpaqueProjectionDigest candidate_digest: OpaqueProjectionDigest route_digest: OpaqueProjectionDigest status_digest: OpaqueProjectionDigest payload_digest: OpaqueProjectionDigest payload: dict[str, str | int] text: str def project_topic59_day_card( card: Topic59DayCard, *, config_digest: str, route_key: tuple[str, str, str], ) -> Topic59Projection: user_id, chat_id, topic_id = route_key if type(user_id) is not str or type(chat_id) is not str or not user_id or not chat_id or topic_id != "59": raise Topic59DayCardError("Topic-59 route") config = require_opaque_digest(config_digest, "config digest") payload = card.payload() if not topic59_payload_is_safe(payload): raise Topic59DayCardError("payload") route = OpaqueProjectionDigest(_hash("route", user_id, chat_id, topic_id)) payload_digest = OpaqueProjectionDigest(_hash("payload", _canonical(payload))) slot = OpaqueProjectionSlot(_hash("card-slot", card.customer_identity_digest, card.kst_day.isoformat())) logical = OpaqueProjectionDigest( _hash("logical", slot, config, card.candidate_digest, route, card.status_row_id, card.status_digest, payload_digest) ) return Topic59Projection( slot, logical, card.customer_identity_digest, card.kst_day, config, card.candidate_digest, route, card.status_digest, payload_digest, payload, card.render(), ) def _canonical(value: Mapping[str, str | int]) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def _hash(domain: str, *parts: str) -> str: material = "\0".join((TOPIC59_PROJECTION_SCHEMA, domain, *parts)).encode("utf-8") return hashlib.sha256(material).hexdigest()