""" Ticket Assignment Model Tracks the complete history of ticket assignments including: - Who is assigned to what ticket - Timeline of work (assigned → accepted → journey → arrived → completed) - GPS tracking (journey breadcrumbs + arrival verification) - Execution planning (agent's queue management) - Performance metrics (travel time, work time) Key Principles: - Each row = one work attempt by one agent on one day - Reassignments create NEW rows (preserves daily history) - Multiple active assignments = team work - State derived from timeline fields (no separate status column) """ from datetime import datetime from typing import Optional from sqlalchemy import Column, Integer, Text, Boolean, DateTime, DECIMAL, ForeignKey from sqlalchemy.dialects.postgresql import UUID, JSONB from sqlalchemy.orm import relationship from app.models.base import BaseModel import uuid class TicketAssignment(BaseModel): """ Ticket Assignment - Tracks WHO is doing WHAT work Timeline States (derived from fields): - PENDING: assigned_at set, responded_at NULL - ACCEPTED: responded_at set, journey_started_at NULL - IN_TRANSIT: journey_started_at set, arrived_at NULL - ON_SITE: arrived_at set, ended_at NULL - CLOSED: ended_at set """ __tablename__ = "ticket_assignments" # Primary Key id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) # Foreign Keys ticket_id = Column(UUID(as_uuid=True), ForeignKey("tickets.id", ondelete="CASCADE"), nullable=False, index=True) user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="RESTRICT"), nullable=False, index=True) assigned_by_user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True) # Assignment Details action = Column(Text, nullable=False, index=True) # 'assigned', 'self_assigned', 'accepted', 'rejected', 'dropped', 'reassigned', 'unassigned' is_self_assigned = Column(Boolean, default=False, nullable=False) # True when agent picks from pool # Execution Planning (Agent Queue Management) execution_order = Column(Integer, nullable=True) # Agent's planned sequence (1, 2, 3...) planned_start_time = Column(DateTime(timezone=True), nullable=True) # Timeline (defines assignment state) assigned_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.utcnow()) responded_at = Column(DateTime(timezone=True), nullable=True) # When agent accepted/rejected journey_started_at = Column(DateTime(timezone=True), nullable=True) # When agent started traveling arrived_at = Column(DateTime(timezone=True), nullable=True) # When agent arrived at site ended_at = Column(DateTime(timezone=True), nullable=True) # When assignment closed (dropped/completed) # Journey Start Location (where agent was when they began journey) journey_start_latitude = Column(DECIMAL(precision=10, scale=7), nullable=True) journey_start_longitude = Column(DECIMAL(precision=10, scale=7), nullable=True) # Arrival Location (GPS verification of arrival at customer site) arrival_latitude = Column(DECIMAL(precision=10, scale=7), nullable=True) arrival_longitude = Column(DECIMAL(precision=10, scale=7), nullable=True) arrival_verified = Column(Boolean, default=False, nullable=False) # Human verified (no auto distance check) # Location Tracking During Journey (GPS breadcrumb trail) # Format: [{"lat": -1.2921, "lng": 36.8219, "accuracy": 10, "timestamp": "2024-01-15T09:30:00Z", "speed": 45}] journey_location_history = Column(JSONB, default=list, nullable=False) # Reason/Notes reason = Column(Text, nullable=True) # Why assigned/dropped/rejected notes = Column(Text, nullable=True) # Timesheet Sync Tracking timesheet_synced = Column(Boolean, default=False, nullable=False) # Has this been synced to timesheets? timesheet_synced_at = Column(DateTime(timezone=True), nullable=True) # When was it synced? # Relationships ticket = relationship("Ticket", back_populates="assignments", lazy="select") user = relationship("User", foreign_keys=[user_id], lazy="select") assigned_by = relationship("User", foreign_keys=[assigned_by_user_id], lazy="select") expenses = relationship("TicketExpense", back_populates="assignment", cascade="all, delete-orphan") # status_changes = relationship("TicketStatusHistory", back_populates="assignment", cascade="all, delete-orphan") # Model doesn't exist yet def __repr__(self): return f"" # ============================================ # Computed Properties # ============================================ @property def status(self) -> str: """ Derive assignment status from timeline fields. No separate status column needed. """ if self.ended_at: return "CLOSED" if self.arrived_at: return "ON_SITE" if self.journey_started_at: return "IN_TRANSIT" if self.responded_at: return "ACCEPTED" return "PENDING" @property def is_active(self) -> bool: """Assignment is active if not ended""" return self.ended_at is None and self.deleted_at is None @property def travel_time_minutes(self) -> Optional[int]: """Calculate travel time (responded_at → arrived_at)""" if self.responded_at and self.arrived_at: delta = self.arrived_at - self.responded_at return int(delta.total_seconds() / 60) return None @property def work_time_minutes(self) -> Optional[int]: """Calculate work time (arrived_at → ended_at)""" if self.arrived_at and self.ended_at: delta = self.ended_at - self.arrived_at return int(delta.total_seconds() / 60) return None @property def total_time_minutes(self) -> Optional[int]: """Calculate total assignment time (assigned_at → ended_at)""" if self.assigned_at and self.ended_at: delta = self.ended_at - self.assigned_at return int(delta.total_seconds() / 60) return None @property def journey_distance_km(self) -> Optional[float]: """ Calculate approximate journey distance from GPS breadcrumbs. Uses centralized Haversine formula from geo utilities. """ if not self.journey_location_history or len(self.journey_location_history) < 2: return None from app.utils.geo import haversine_distance total_distance = 0.0 for i in range(len(self.journey_location_history) - 1): point1 = self.journey_location_history[i] point2 = self.journey_location_history[i + 1] if all(k in point1 for k in ["lat", "lng"]) and all(k in point2 for k in ["lat", "lng"]): try: distance = haversine_distance( float(point1["lat"]), float(point1["lng"]), float(point2["lat"]), float(point2["lng"]) ) total_distance += distance except (ValueError, TypeError): # Skip invalid coordinates continue return round(total_distance, 2) # ============================================ # Business Logic Methods # ============================================ def mark_accepted(self, notes: Optional[str] = None): """Agent accepts assignment""" self.responded_at = datetime.utcnow() if notes: self.notes = notes def mark_rejected(self, reason: str): """Agent rejects assignment - closes assignment immediately""" self.responded_at = datetime.utcnow() self.ended_at = datetime.utcnow() self.reason = reason def start_journey(self, latitude: float, longitude: float): """Agent starts traveling to site""" self.journey_started_at = datetime.utcnow() self.journey_start_latitude = latitude self.journey_start_longitude = longitude def add_location_breadcrumb(self, latitude: float, longitude: float, accuracy: Optional[float] = None, speed: Optional[float] = None, **kwargs): """Add GPS location to journey trail""" if not isinstance(self.journey_location_history, list): self.journey_location_history = [] breadcrumb = { "lat": latitude, "lng": longitude, "timestamp": datetime.utcnow().isoformat(), } if accuracy is not None: breadcrumb["accuracy"] = accuracy if speed is not None: breadcrumb["speed"] = speed # Add any additional metadata breadcrumb.update(kwargs) self.journey_location_history.append(breadcrumb) def mark_arrived(self, latitude: float, longitude: float): """Agent arrives at customer site""" self.arrived_at = datetime.utcnow() self.arrival_latitude = latitude self.arrival_longitude = longitude # arrival_verified is set manually by human (no auto distance check) def mark_dropped(self, reason: str): """Agent drops ticket (can't complete)""" self.ended_at = datetime.utcnow() self.reason = reason def mark_completed(self): """Work completed - assignment ends""" self.ended_at = datetime.utcnow() def update_execution_order(self, order: int, planned_start: Optional[datetime] = None): """Agent reorders their ticket queue""" self.execution_order = order if planned_start: self.planned_start_time = planned_start def can_claim_expenses(self) -> bool: """ Check if agent can claim expenses for this assignment. Must have arrived at site. """ return self.arrived_at is not None