"""Operator CLI for nutrition onboarding state.""" from __future__ import annotations import argparse import json import sys from datetime import date from pathlib import Path from typing import Sequence from checkin_cli.customer_coaching import load_customer_registry from checkin_cli.nutrition_onboarding import NutritionOnboardingService from checkin_cli.nutrition_onboarding_contract import ( MessageEvidence, OnboardingAuthority, ) from checkin_cli.nutrition_onboarding_migration import ( commit_legacy_migration, legacy_migration_preflight, ) from checkin_cli.nutrition_restriction_kb import seed_restriction_kb def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="nutrition-onboarding") subcommands = parser.add_subparsers(dest="command", required=True) status = subcommands.add_parser("status") status.add_argument("--profile-root", type=Path, required=True) status.add_argument("--customer-key", required=True) purge = subcommands.add_parser("purge-incomplete") purge.add_argument("--profile-root", type=Path, required=True) purge.add_argument("--customer-key", required=True) review_actions = { "operator-review": ("approve", "revise", "reject", "safety-hold"), } for name, actions in review_actions.items(): review = subcommands.add_parser(name) review.add_argument("--profile-root", type=Path, required=True) review.add_argument("--customer-key", required=True) review.add_argument("--action", choices=actions, required=True) review.add_argument("--field") review.add_argument("--message-id", type=int, required=True) review.add_argument("--update-id", type=int, required=True) seed = subcommands.add_parser("seed-kb") seed.add_argument("--profile-root", type=Path, required=True) seed.add_argument("--source", type=Path, required=True) seed.add_argument("--owner-digest", required=True) seed.add_argument("--as-of", type=date.fromisoformat, required=True) seed.add_argument("--commit", action="store_true") for name in ("migration-preflight", "migration-commit"): migration = subcommands.add_parser(name) migration.add_argument("--profile-root", type=Path, required=True) migration.add_argument("--expected-enabled-customer", required=True) migration.add_argument("--owner-digest", required=True) if name == "migration-commit": migration.add_argument( "--confirm-activation-receipt-digest", required=True, ) return parser def load_current_authority( *, profile_root: Path, customer_key: str, ) -> OnboardingAuthority: registry = load_customer_registry( profile_root / "customers" / "registry.json", profile_root, ) matches = [ runtime for runtime in registry.customers if runtime.spec.customer_key == customer_key ] if len(matches) != 1: raise ValueError("current onboarding review authority is unavailable") customer = matches[0].spec consent = customer.ai_processing_consent return OnboardingAuthority( customer_key=customer.customer_key, customer_user_id=int(customer.telegram.user_id), customer_chat_id=int(customer.telegram.chat_id), customer_topic_id=int(customer.telegram.topic_id), owner_user_id=int(registry.owner.user_id), owner_chat_id=int(registry.owner.chat_id), owner_topic_id=int(registry.owner.topic_id), consent_notice_version=consent.notice_version or "", consent_granted=consent.granted, customer_enabled=customer.enabled, ) def _review_evidence( authority: OnboardingAuthority, *, role: str, message_id: int, update_id: int, ) -> MessageEvidence: return MessageEvidence( actor_user_id=getattr(authority, f"{role}_user_id"), chat_id=getattr(authority, f"{role}_chat_id"), topic_id=getattr(authority, f"{role}_topic_id"), message_id=message_id, update_id=update_id, ) def main(argv: Sequence[str] | None = None) -> int: parser = _parser() arguments = parser.parse_args(argv) if ( arguments.command == "operator-review" and arguments.action == "revise" and not arguments.field ): parser.error("--field is required for revise") if arguments.command == "seed-kb": result = seed_restriction_kb( profile_root=arguments.profile_root, source=arguments.source, owner_digest=arguments.owner_digest, commit=arguments.commit, as_of=arguments.as_of, ) print(json.dumps(result.model_dump(mode="json"), sort_keys=True)) return 0 if arguments.command == "migration-preflight": result = legacy_migration_preflight( profile_root=arguments.profile_root, expected_enabled_customer=arguments.expected_enabled_customer, owner_digest=arguments.owner_digest, ) print(json.dumps(result, sort_keys=True)) return 0 if arguments.command == "migration-commit": result = commit_legacy_migration( profile_root=arguments.profile_root, expected_enabled_customer=arguments.expected_enabled_customer, owner_digest=arguments.owner_digest, confirm_activation_receipt_digest=( arguments.confirm_activation_receipt_digest ), ) print(json.dumps(result, sort_keys=True)) return 0 service = NutritionOnboardingService( profile_root=arguments.profile_root, customer_key=arguments.customer_key, ) try: if arguments.command == "operator-review": authority = load_current_authority( profile_root=arguments.profile_root, customer_key=arguments.customer_key, ) evidence = _review_evidence( authority, role="owner", message_id=arguments.message_id, update_id=arguments.update_id, ) result = service.review_as_owner( decision={ "approve": "approved", "revise": "revise", "reject": "rejected", "safety-hold": "safety_hold", }[arguments.action], revision_field=arguments.field, authority=authority, evidence=evidence, ) print( json.dumps( result.model_dump(mode="json"), ensure_ascii=False, sort_keys=True, ) ) return 0 if arguments.command == "status": print( json.dumps( service.status().model_dump(mode="json"), ensure_ascii=False, sort_keys=True, ) ) return 0 status = service.status() if status.state.value in {"ready", "finalizing"}: print("committed onboarding state cannot be purged", file=sys.stderr) return 2 removed = service.store.purge_transient() print(json.dumps({"removed": removed}, sort_keys=True)) return 0 except FileNotFoundError: print("nutrition onboarding state not found", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())