"""Data models for work order ingestion.""" from dataclasses import dataclass, field, asdict from datetime import datetime from enum import Enum from typing import Optional class EmailType(str, Enum): NEW_WORK_ORDER = "new_work_order" UPDATE = "update" COMMENT = "comment" UNKNOWN = "unknown" class WorkOrderStatus(str, Enum): NEW = "new" ASSIGNED = "assigned" IN_PROGRESS = "in_progress" ON_HOLD = "on_hold" COMPLETED = "completed" CANCELLED = "cancelled" UNKNOWN = "unknown" @dataclass class Comment: work_order_id: str comment_id: str # generated: {work_order_id}#{timestamp} commenter: str text: str created_at: str # ISO 8601 source_email_s3_key: str ingested_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) def to_dynamo_item(self) -> dict: return {k: v for k, v in asdict(self).items() if v is not None} @dataclass class WorkOrder: work_order_id: str description: Optional[str] = None status: str = WorkOrderStatus.UNKNOWN.value customer: str = "AMAZON" site_code: Optional[str] = None building: Optional[str] = None address: Optional[str] = None severity: Optional[str] = None priority: Optional[str] = None date_reported: Optional[str] = None scheduled_start: Optional[str] = None due_date: Optional[str] = None assigned_to: Optional[str] = None source_email_s3_key: Optional[str] = None created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) updated_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) def to_dynamo_item(self) -> dict: return {k: v for k, v in asdict(self).items() if v is not None} @dataclass class ParsedEmail: """Result of AI parsing an inbound email.""" email_type: str # EmailType value work_order_id: Optional[str] = None description: Optional[str] = None status: Optional[str] = None site_code: Optional[str] = None building: Optional[str] = None address: Optional[str] = None severity: Optional[str] = None priority: Optional[str] = None date_reported: Optional[str] = None scheduled_start: Optional[str] = None due_date: Optional[str] = None assigned_to: Optional[str] = None commenter: Optional[str] = None comment_text: Optional[str] = None comment_time: Optional[str] = None raw_subject: Optional[str] = None