from __future__ import annotations import json import shlex from datetime import datetime, time from typing import Any from .exceptions import CommandError GOAL_FLAG_MAP = { "name": "name", "n": "name", "description": "description", "desc": "description", "status": "status", "s": "status", } DAYLOG_FLAG_MAP = { "date": "log_date", "d": "log_date", "bed": "bed_time", "b": "bed_time", "wake": "wake_time", "w": "wake_time", "sleep": "sleep_time", "s": "sleep_time", "sleephr": "sleep_duration_hours", "sh": "sleep_duration_hours", "totalcal": "total_calories", "tc": "total_calories", "totalprod": "total_productivity_min", "tp": "total_productivity_min", "note": "notes", "n": "notes", } ACTIVITY_FLAG_MAP = { "goal": "goal_name", "g": "goal_name", "date": "entry_date", "d": "entry_date", "name": "activity_type_name", "n": "activity_type_name", "type": "category", "t": "category", "status": "status", "s": "status", "time": "occurred_at", "ti": "occurred_at", "min": "duration_min", "m": "duration_min", "place": "location", "p": "location", "quan": "quantity_text", "q": "quantity_text", "what": "what_i_did", "w": "what_i_did", "result": "result_text", "r": "result_text", "note": "note", "no": "note", "detail": "details", "de": "details", } def _parse_bool(value: Any) -> bool: if isinstance(value, bool): return value str_value = str(value).strip().lower() if str_value in ("true", "1", "yes"): return True elif str_value in ("false", "0", "no"): return False else: raise CommandError(400, "invalid_field_type", f"Invalid boolean value '{value}'. Expected true/false, 1/0, or yes/no.") def _parse_date(value: Any): try: return datetime.strptime(str(value), "%Y-%m-%d").date() except ValueError as exc: raise CommandError(400, "invalid_field_type", f"Invalid date value '{value}'. Expected YYYY-MM-DD.") from exc def _parse_time(value: Any) -> time: for fmt in ("%H:%M:%S", "%H:%M"): try: return datetime.strptime(str(value), fmt).time() except ValueError: continue raise CommandError(400, "invalid_field_type", f"Invalid time value '{value}'. Expected HH:MM or HH:MM:SS.") def _parse_int(value: Any) -> int: try: return int(str(value)) except ValueError as exc: raise CommandError(400, "invalid_field_type", f"Invalid integer value '{value}'.") from exc def _parse_float(value: Any) -> float: try: return float(str(value)) except ValueError as exc: raise CommandError(400, "invalid_field_type", f"Invalid number value '{value}'.") from exc def _parse_detail(value: Any) -> dict[str, Any]: if isinstance(value, dict): return value try: parsed = json.loads(str(value)) except json.JSONDecodeError as exc: raise CommandError(400, "invalid_field_type", "The detail field must be valid JSON.") from exc if not isinstance(parsed, dict): raise CommandError(400, "invalid_field_type", "The detail field must be a JSON object.") return parsed def _flag_map_for_intent(intent: str) -> dict[str, str]: if intent == "goal": return GOAL_FLAG_MAP if intent == "daylog": return DAYLOG_FLAG_MAP if intent == "activity": return ACTIVITY_FLAG_MAP raise CommandError(400, "unknown_intent", f"Unknown intent '{intent}'.") def parse_command_text(command_text: str) -> tuple[str, Optional[str], dict[str, Any]]: try: tokens = shlex.split(command_text.strip()) except ValueError as exc: raise CommandError(400, "invalid_command", "Command text contains unmatched quotes.") from exc if not tokens: raise CommandError(400, "invalid_command", "Command text cannot be empty.") first_token = tokens[0] if not first_token.startswith("/"): raise CommandError(400, "invalid_command", "Commands must start with '/'.") intent = first_token[1:].strip().lower() if not intent: raise CommandError(400, "invalid_command", "Command intent is missing.") if intent == "help": help_topic = tokens[1].lower() if len(tokens) > 1 else None if len(tokens) > 2: raise CommandError(400, "invalid_command", "/help accepts at most one optional topic.") return intent, help_topic, {} flag_map = _flag_map_for_intent(intent) values: dict[str, Any] = {} index = 1 while index < len(tokens): token = tokens[index] if not token.startswith("-"): raise CommandError(400, "invalid_command", f"Unexpected value '{token}'. Every value must follow a flag.") flag = token.lstrip("-").lower() if flag not in flag_map: raise CommandError(400, "unknown_flag", f"Unknown flag '{token}' for intent '{intent}'.") index += 1 if index >= len(tokens): raise CommandError(400, "missing_required_field", f"Missing value for flag '{token}'.") value = tokens[index] if value.startswith("-"): raise CommandError(400, "missing_required_field", f"Missing value for flag '{token}'.") canonical_name = flag_map[flag] if canonical_name in values: raise CommandError(400, "invalid_command", f"Flag '{token}' was provided more than once.") values[canonical_name] = value index += 1 return intent, None, values def build_typed_payload(intent: str, raw_values: dict[str, Any]) -> dict[str, Any]: if intent == "goal": payload = dict(raw_values) return payload if intent == "daylog": payload = dict(raw_values) if "log_date" in payload: payload["log_date"] = _parse_date(payload["log_date"]) for field_name in ("bed_time", "wake_time", "sleep_time"): if field_name in payload: payload[field_name] = _parse_time(payload[field_name]) if "sleep_duration_hours" in payload: payload["sleep_duration_hours"] = _parse_float(payload["sleep_duration_hours"]) for field_name in ("total_productivity_min", "total_calories"): if field_name in payload: payload[field_name] = _parse_int(payload[field_name]) return payload if intent == "activity": payload = dict(raw_values) if "entry_date" in payload: payload["entry_date"] = _parse_date(payload["entry_date"]) if "occurred_at" in payload: payload["occurred_at"] = _parse_time(payload["occurred_at"]) if "duration_min" in payload: payload["duration_min"] = _parse_int(payload["duration_min"]) if "details" in payload: payload["details"] = _parse_detail(payload["details"]) if "category" in payload: payload["category"] = str(payload["category"]).lower() return payload raise CommandError(400, "unknown_intent", f"Unknown intent '{intent}'.") def validate_required_fields(intent: str, payload: dict[str, Any]) -> None: required_fields = { "goal": ("name",), "daylog": ("log_date",), "activity": ("entry_date", "activity_type_name", "category"), } missing = [field for field in required_fields[intent] if field not in payload or payload[field] in (None, "")] if missing: raise CommandError(400, "missing_required_field", f"Missing required field(s): {', '.join(missing)}.", {"fields": missing})