"""Focused Telegram scheduling seam for Todo 5 reminders and cutoff.""" from __future__ import annotations from datetime import date, datetime from typing import TYPE_CHECKING, ClassVar, Protocol, runtime_checkable from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictInt, StrictStr from gateway.platforms.nutrition_weekly_operations_config import JsonValue from gateway.platforms.nutrition_weekly_reminder_authority import ( WeeklyReminderOwnerAuthority, ) if TYPE_CHECKING: from checkin_cli.weekly_operations_schedule_host_models_r4 import ( WeeklyOperationsSchedulePolicy, ) from gateway.platforms.nutrition_weekly_reminder import ( TelegramReminderDelivered, TelegramReminderRejected, ) from gateway.platforms.nutrition_weekly_reminder_authority import ( WeeklyOperationsTickSnapshot, ) class _PlanningWeekly(BaseModel): model_config: ClassVar[ConfigDict] = ConfigDict(strict=True, extra="forbid") enabled: StrictBool = False reminder_time: StrictStr = "20:00:00" missed_cutoff_time: StrictStr = "23:00:00" weekly_weekday: StrictInt = 0 feature_epoch: StrictStr = "weekly-operations-v1" registry_identity_binding_digest: StrictStr | None = None class _PlanningRoute(BaseModel): model_config: ClassVar[ConfigDict] = ConfigDict(strict=True, extra="forbid") user_id: StrictStr chat_id: StrictStr topic_id: StrictInt | StrictStr class _PlanningOwner(BaseModel): model_config: ClassVar[ConfigDict] = ConfigDict(strict=True, extra="forbid") user_id: StrictStr chat_id: StrictStr version: StrictInt class _PlanningReceipt(BaseModel): model_config: ClassVar[ConfigDict] = ConfigDict(strict=True, extra="forbid") schema_value: StrictStr = Field(alias="schema") candidate_digest: StrictStr config_digest: StrictStr enabled_customer_keys: list[StrictStr] owner: _PlanningOwner consent_digest: StrictStr registry_identity: dict[StrictStr, StrictStr | StrictInt] issued_at: StrictStr expires_at: StrictStr feature_epoch: StrictStr class _PlanningNutrition(BaseModel): model_config: ClassVar[ConfigDict] = ConfigDict(strict=True, extra="allow") weekly_operations: _PlanningWeekly | None = None operator_review: _PlanningRoute | None = None weekly_operations_authority: _PlanningReceipt | None = None class _PlanningEnvelope(BaseModel): model_config: ClassVar[ConfigDict] = ConfigDict(strict=True, extra="allow") nutrition_coaching: _PlanningNutrition | None = None class WeeklySchedulePlanningIncident(RuntimeError): """Enabled weekly planning is unavailable in the configured profile.""" @runtime_checkable class WeeklyReminderCoordinator(Protocol): def refresh_live_registry(self) -> bool: ... @property def weekly_reminder_authority_owner( self, ) -> WeeklyReminderOwnerAuthority: ... class WeeklyReminderTask(Protocol): @property def customer_key(self) -> str: ... @property def kind(self) -> str: ... @property def kst_day(self) -> date: ... class WeeklyReminderTelegramHost(Protocol): async def send_weekly_reminder( self, chat_id: str, topic_id: str | None, text: str ) -> TelegramReminderDelivered | TelegramReminderRejected: ... def weekly_operations_authority_current( self, snapshot: WeeklyOperationsTickSnapshot ) -> bool: ... def weekly_schedule_policy( enabled_customer_keys: tuple[str, ...], ) -> WeeklyOperationsSchedulePolicy: """Import the profile policy only after gateway config enabled weekly work.""" try: from checkin_cli.weekly_operations_schedule_host_models_r4 import ( WeeklyOperationsSchedulePolicy, ) except (ImportError, AttributeError) as error: raise WeeklySchedulePlanningIncident( "enabled weekly schedule policy is unavailable" ) from error return WeeklyOperationsSchedulePolicy(frozenset(enabled_customer_keys)) def configured_weekly_authority( encoded_extra: str, ): """Parse gateway JSON before conditionally importing the profile policy.""" from gateway.platforms.nutrition_weekly_operations_authority import ( parse_weekly_operations_authority, ) from gateway.platforms.nutrition_weekly_operations_config import ( parse_weekly_operations_config, ) envelope = _PlanningEnvelope.model_validate_json(encoded_extra) nutrition = envelope.nutrition_coaching if nutrition is None or nutrition.weekly_operations is None: return None weekly = nutrition.weekly_operations route = nutrition.operator_review extra: dict[str, JsonValue] = {"nutrition_coaching": { "weekly_operations": { "enabled": weekly.enabled, "reminder_time": weekly.reminder_time, "missed_cutoff_time": weekly.missed_cutoff_time, "weekly_weekday": weekly.weekly_weekday, "feature_epoch": weekly.feature_epoch, "registry_identity_binding_digest": weekly.registry_identity_binding_digest, }, **({} if route is None else {"operator_review": { "user_id": route.user_id, "chat_id": route.chat_id, "topic_id": route.topic_id, }}), }} config = parse_weekly_operations_config(extra) if not config.enabled: return None receipt = nutrition.weekly_operations_authority if receipt is None: return None raw: dict[str, JsonValue] = { "schema": receipt.schema_value, "candidate_digest": receipt.candidate_digest, "config_digest": receipt.config_digest, "enabled_customer_keys": receipt.enabled_customer_keys, "owner": { "user_id": receipt.owner.user_id, "chat_id": receipt.owner.chat_id, "version": receipt.owner.version, }, "consent_digest": receipt.consent_digest, "registry_identity": receipt.registry_identity, "issued_at": receipt.issued_at, "expires_at": receipt.expires_at, "feature_epoch": receipt.feature_epoch, } parsed = parse_weekly_operations_authority(raw) return config, parsed def configured_weekly_schedule_policy( encoded_extra: str, ) -> WeeklyOperationsSchedulePolicy | None: """Return the enabled profile schedule without importing it while OFF.""" authority = configured_weekly_authority(encoded_extra) if authority is None: return None _config, receipt = authority return weekly_schedule_policy(receipt.enabled_customer_keys) def _coordinator_owner( coordinator: WeeklyReminderCoordinator | None, ) -> WeeklyReminderOwnerAuthority: from gateway.platforms.nutrition_weekly_reminder_authority import ( WeeklyReminderAuthorityOwner, WeeklyReminderOwnerError, ) if not isinstance(coordinator, WeeklyReminderCoordinator): raise WeeklyReminderOwnerError( "weekly reminder coordinator protocol is unavailable" ) owner = coordinator.weekly_reminder_authority_owner if not isinstance(owner, WeeklyReminderAuthorityOwner): raise WeeklyReminderOwnerError("weekly reminder authority owner is invalid") return owner async def send_weekly_operations_task( host: WeeklyReminderTelegramHost, coordinator: WeeklyReminderCoordinator, task: WeeklyReminderTask, *, local_now: datetime, ) -> str | None: """Resolve authority from the real coordinator owner and run one due task.""" from checkin_cli.weekly_operations_cutoff import run_missed_cutoff from checkin_cli.weekly_operations_lifecycle import ( ReminderDependencies, run_due_reminder, ) from checkin_cli.weekly_reminder_authority import WeeklyReminderRequest from gateway.platforms.nutrition_weekly_reminder import ( TelegramWeeklyReminderProvider, ) from gateway.platforms.nutrition_weekly_reminder_fence import ( FencedReminderHost, ReminderProviderFence, ) owner = _coordinator_owner(coordinator) snapshot = owner.tick_snapshot(task.customer_key, local_now) bound = owner.bound_customer(task.customer_key, local_now) request = WeeklyReminderRequest(bound, task.kst_day, local_now) if task.kind == "cutoff": _ = run_missed_cutoff(request) return None result = await run_due_reminder( request, ReminderDependencies( TelegramWeeklyReminderProvider( FencedReminderHost( ReminderProviderFence( host, coordinator, snapshot, bound, local_now ) ) ), lambda: _coordinator_owner(coordinator).bound_customer( task.customer_key, local_now ), ), ) if result.receipt is None or result.receipt.state == "sent_audited": return None return f"{task.customer_key}:{task.kst_day}:{task.kind}"