mirror of
https://github.com/Sea-Haven-Industries/workorder-ingest.git
synced 2026-05-18 20:20:12 +00:00
83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
|
"""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
|