# app/modules/chassis_tracker/service.py from datetime import datetime, timezone, date from uuid import UUID from typing import List, Dict, Any, Optional from sqlalchemy import select, or_, and_, func from sqlalchemy.ext.asyncio import AsyncSession from fastapi import HTTPException from app.models.yard_models import ( ChassisMaster, ChassisMovementLog, ChassisStoreHandover, VehicleModelMaster ) from app.models.sales_models import CustomerMaster, ClientMaster from app.models.dealer_models import DealerOrderVehicle, DealerOrder from app.models.rm_models import StationMaster from app.models.production_models import ChassisStationLog, ProductionHold, DelayReasonMaster from app.models.qa_models import QaChecksheetInstance, QaCheckpointResponse from app.models.dispatch_models import DispatchThermalChecksheet, DispatchGatePass, DispatchGatePassItem from app.models.core_models import LoginUsers from .schemas import ( ChassisSearchResult, ChassisLifecycleResponse, StationVisitLog, TimelineMilestone, ClientUpdateMessages ) from .order_spec_service import resolve_chassis_order_spec class ChassisTrackerService: def __init__(self, db: AsyncSession): self.db = db async def search_chassis(self, query: str, limit: int = 25) -> List[ChassisSearchResult]: q = (query or '').strip() if not q: stmt = select(ChassisMaster).order_by(ChassisMaster.created_at.desc()).limit(limit) else: search_pattern = f"%{q}%" stmt = ( select(ChassisMaster) .outerjoin(VehicleModelMaster, ChassisMaster.model_id == VehicleModelMaster.model_id) .outerjoin(CustomerMaster, ChassisMaster.dealer_id == CustomerMaster.customer_id) .where( or_( ChassisMaster.chassis_no.ilike(search_pattern), VehicleModelMaster.model_number.ilike(search_pattern), CustomerMaster.name.ilike(search_pattern), ) ) .order_by(ChassisMaster.created_at.desc()) .limit(limit) ) result = await self.db.execute(stmt) chassis_list = result.scalars().all() if not chassis_list: return [] # Batch resolve foreign tables for fast response model_ids = {c.model_id for c in chassis_list if c.model_id} dealer_ids = {c.dealer_id for c in chassis_list if c.dealer_id} client_ids = {c.client_id for c in chassis_list if c.client_id} station_ids = {c.current_station_id for c in chassis_list if c.current_station_id} models_map = {} if model_ids: m_res = await self.db.execute(select(VehicleModelMaster).where(VehicleModelMaster.model_id.in_(model_ids))) models_map = {m.model_id: m for m in m_res.scalars().all()} dealers_map = {} if dealer_ids: d_res = await self.db.execute(select(CustomerMaster).where(CustomerMaster.customer_id.in_(dealer_ids))) dealers_map = {d.customer_id: d.name for d in d_res.scalars().all()} clients_map = {} if client_ids: c_res = await self.db.execute(select(ClientMaster).where(ClientMaster.client_id.in_(client_ids))) clients_map = {c.client_id: c.name for c in c_res.scalars().all()} stations_map = {} if station_ids: s_res = await self.db.execute(select(StationMaster).where(StationMaster.station_id.in_(station_ids))) stations_map = {s.station_id: s.station_name for s in s_res.scalars().all()} out = [] for c in chassis_list: model = models_map.get(c.model_id) dealer_name = dealers_map.get(c.dealer_id) client_name = clients_map.get(c.client_id) station_name = stations_map.get(c.current_station_id) progress = self._calculate_progress_percent(c) out.append( ChassisSearchResult( chassis_id=c.chassis_id, chassis_no=c.chassis_no, model_name=model.model_number if model else None, model_code=model.model_number if model else None, dealer_name=dealer_name, client_name=client_name, current_status=c.current_status, tracking_status=c.tracking_status, current_station_name=station_name, progress_percent=progress, arrival_date=c.arrival_date, ) ) return out async def get_chassis_lifecycle(self, chassis_identifier: str) -> ChassisLifecycleResponse: ident = (chassis_identifier or '').strip() if not ident: raise HTTPException(status_code=400, detail="Chassis number or ID required") # Try match by UUID or chassis_no c = None try: val_uuid = UUID(ident) c = (await self.db.execute(select(ChassisMaster).where(ChassisMaster.chassis_id == val_uuid))).scalar_one_or_none() except ValueError: pass if not c: c = (await self.db.execute(select(ChassisMaster).where(ChassisMaster.chassis_no.ilike(ident)))).scalar_one_or_none() if not c: raise HTTPException(status_code=404, detail=f"Chassis '{ident}' not found in the system.") # 1. Resolve Model model_name = None model_code = None if c.model_id: m = (await self.db.execute(select(VehicleModelMaster).where(VehicleModelMaster.model_id == c.model_id))).scalar_one_or_none() if m: model_name = getattr(m, 'model_number', None) model_code = getattr(m, 'model_number', None) # 2. Resolve Dealer & Client & Order dealer_name = None dealer_contacts = [] if c.dealer_id: dealer = (await self.db.execute(select(CustomerMaster).where(CustomerMaster.customer_id == c.dealer_id))).scalar_one_or_none() if dealer: dealer_name = dealer.name client_name = None client_phone = None if c.client_id: client = (await self.db.execute(select(ClientMaster).where(ClientMaster.client_id == c.client_id))).scalar_one_or_none() if client: client_name = client.name client_phone = client.phone order_no = None rep_name = None rep_contact = None if c.dealer_order_vehicle_id: dov = (await self.db.execute(select(DealerOrderVehicle).where(DealerOrderVehicle.order_vehicle_id == c.dealer_order_vehicle_id))).scalar_one_or_none() if dov: do = (await self.db.execute(select(DealerOrder).where(DealerOrder.order_id == dov.order_id))).scalar_one_or_none() if do: order_no = do.order_no rep_name = do.rep_name rep_contact = do.rep_contact # 3. Resolve Current Station name current_station_name = None if c.current_station_id: curr_s = (await self.db.execute(select(StationMaster).where(StationMaster.station_id == c.current_station_id))).scalar_one_or_none() if curr_s: current_station_name = curr_s.station_name # 4. Resolve Production Station Roadmap & Logs stations_query = await self.db.execute( select(StationMaster).where(StationMaster.is_active == True).order_by(StationMaster.sequence_no.nulls_last()) ) all_stations = stations_query.scalars().all() s_logs_res = await self.db.execute( select(ChassisStationLog).where(ChassisStationLog.chassis_id == c.chassis_id).order_by(ChassisStationLog.scanned_at.asc()) ) station_logs = s_logs_res.scalars().all() # Collect user ids for resolving names user_ids = {log.scanned_by for log in station_logs if log.scanned_by} delay_reason_ids = {log.delay_reason_id for log in station_logs if log.delay_reason_id} # Also check QA checksheets qa_res = await self.db.execute( select(QaChecksheetInstance).where(QaChecksheetInstance.chassis_id == c.chassis_id).order_by(QaChecksheetInstance.inspection_date.asc()) ) qa_instances = qa_res.scalars().all() for q in qa_instances: if q.inspector_id: user_ids.add(q.inspector_id) # Store Handover store_ho = (await self.db.execute(select(ChassisStoreHandover).where(ChassisStoreHandover.chassis_id == c.chassis_id))).scalar_one_or_none() if store_ho and store_ho.store_keeper_id: user_ids.add(store_ho.store_keeper_id) # Production Holds holds_res = await self.db.execute( select(ProductionHold).where(ProductionHold.chassis_id == c.chassis_id).order_by(ProductionHold.raised_at.desc()) ) holds = holds_res.scalars().all() for h in holds: if h.raised_by: user_ids.add(h.raised_by) if h.resolved_by: user_ids.add(h.resolved_by) # Dispatch Thermal Checksheet thermal_res = await self.db.execute( select(DispatchThermalChecksheet).where(DispatchThermalChecksheet.chassis_id == c.chassis_id).order_by(DispatchThermalChecksheet.created_at.desc()) ) thermal_check = thermal_res.scalars().first() # Dispatch Gate Pass gp_res = await self.db.execute( select(DispatchGatePass).where(DispatchGatePass.chassis_id == c.chassis_id).order_by(DispatchGatePass.generated_at.desc()) ) gate_pass = gp_res.scalars().first() gp_items = [] if gate_pass: gpi_res = await self.db.execute( select(DispatchGatePassItem).where(DispatchGatePassItem.gate_pass_id == gate_pass.gate_pass_id) ) gp_items = [{"label": i.item_label, "checked": i.is_checked, "qty": i.qty} for i in gpi_res.scalars().all()] # Resolve Users Dictionary users_map = {} if user_ids: u_query = await self.db.execute(select(LoginUsers).where(LoginUsers.user_id.in_(user_ids))) users_map = {u.user_id: u.name for u in u_query.scalars().all()} # Resolve Delay Reasons Dictionary delay_map = {} if delay_reason_ids: dr_query = await self.db.execute(select(DelayReasonMaster).where(DelayReasonMaster.delay_reason_id.in_(delay_reason_ids))) delay_map = {dr.delay_reason_id: dr.reason_label for dr in dr_query.scalars().all()} # Group station logs by station_id logs_by_station: Dict[UUID, Dict[str, Any]] = {} for l in station_logs: if l.station_id not in logs_by_station: logs_by_station[l.station_id] = { "arrivals": [], "completions": [], "latest": l } if l.event_type == 'ARRIVAL': logs_by_station[l.station_id]["arrivals"].append(l) elif l.event_type == 'COMPLETION': logs_by_station[l.station_id]["completions"].append(l) # Build QA stats per station qa_by_station: Dict[UUID, Dict[str, Any]] = {} qa_inspections_list = [] for q in qa_instances: # Check responses count chk_res = await self.db.execute( select( func.count().label("total"), func.count().filter(QaCheckpointResponse.status == 'OK').label("passed") ).where(QaCheckpointResponse.instance_id == q.instance_id) ) row = chk_res.one() total_items = row.total passed_items = row.passed station_name_q = next((s.station_name for s in all_stations if s.station_id == q.station_id), "Station") qa_data = { "instance_id": q.instance_id, "station_id": q.station_id, "station_name": station_name_q, "inspection_date": q.inspection_date, "overall_result": q.overall_result, "station_qa_status": q.station_qa_status, "inspector_name": users_map.get(q.inspector_id, "QA Inspector"), "passed_count": passed_items, "total_count": total_items, "remarks": q.remarks, "rework_notes": q.rework_notes, "completed_at": q.completed_at } qa_inspections_list.append(qa_data) if q.station_id not in qa_by_station or q.overall_result == 'PASSED': qa_by_station[q.station_id] = qa_data # Build production_stations visit logs production_stations: List[StationVisitLog] = [] for s in all_stations: st_logs = logs_by_station.get(s.station_id) qa_info = qa_by_station.get(s.station_id) status = "PENDING" arr_time = None comp_time = None scanned_by_name = None elapsed_m = None is_overdue = False delay_reason = None delay_remarks = None if st_logs: first_arr = st_logs["arrivals"][0] if st_logs["arrivals"] else None last_comp = st_logs["completions"][-1] if st_logs["completions"] else None if first_arr: arr_time = first_arr.scanned_at scanned_by_name = users_map.get(first_arr.scanned_by) if last_comp: comp_time = last_comp.scanned_at elapsed_m = last_comp.elapsed_minutes is_overdue = last_comp.is_overdue delay_reason = delay_map.get(last_comp.delay_reason_id) delay_remarks = last_comp.delay_remarks status = "COMPLETED" elif first_arr: status = "AT_STATION" elapsed_m = int((datetime.now(timezone.utc) - first_arr.scanned_at).total_seconds() / 60) elif c.current_station_id == s.station_id: status = "AT_STATION" arr_time = c.station_arrived_at production_stations.append( StationVisitLog( station_id=s.station_id, station_code=s.station_code, station_name=s.station_name, sequence_no=s.sequence_no, status=status, arrival_time=arr_time, completion_time=comp_time, scanned_by_name=scanned_by_name, elapsed_minutes=elapsed_m, standard_cycle_time_min=s.standard_cycle_time_min, is_overdue=is_overdue, delay_reason=delay_reason, delay_remarks=delay_remarks, qa_status=qa_info.get("overall_result") if qa_info else None, qa_passed_count=qa_info.get("passed_count") if qa_info else None, qa_total_count=qa_info.get("total_count") if qa_info else None, ) ) # Calculate progress % and stage label progress_percent = self._calculate_progress_percent(c, production_stations, gate_pass, thermal_check) stage_label = self._determine_stage_label(c, current_station_name, gate_pass, thermal_check) # 5. Build Timeline Milestones timeline = self._build_timeline(c, store_ho, production_stations, qa_inspections_list, thermal_check, gate_pass) # 6. Generate Formatted Client Status Update Messages client_msgs = self._generate_client_messages( c=c, model_name=model_name or "Standard Bus", dealer_name=dealer_name or client_name or "Valued Client", order_no=order_no or f"ORD-{c.chassis_no}", stage_label=stage_label, progress_percent=progress_percent, timeline=timeline, current_station_name=current_station_name, gate_pass=gate_pass ) return ChassisLifecycleResponse( chassis_id=c.chassis_id, chassis_no=c.chassis_no, model={ "model_id": c.model_id, "model_name": model_name or "Bus Model", "model_code": model_code or "N/A", "fuel_type": c.fuel_type or "DIESEL", "light_type": c.light_type, "voltage": c.voltage or "12V", "oem": c.oem or "OEM", "year_of_manufacture": c.year_of_manufacture }, customer={ "dealer_id": c.dealer_id, "dealer_name": dealer_name, "client_id": c.client_id, "client_name": client_name, "client_phone": client_phone, "order_no": order_no, "rep_name": rep_name, "rep_contact": rep_contact }, yard={ "arrival_date": c.arrival_date, "yard_location": c.yard_location or "Plant Yard", "fuel_receiving_qty": float(c.fuel_receiving_qty or 0), "chassis_condition": c.chassis_condition or "OK", "chassis_odometer": c.chassis_odometer, "inbound_vehicle_no": c.inbound_vehicle_no, "driver_name": c.driver_name, "driver_phone": c.driver_phone, "transport_company": c.transport_company or c.transport_id, "da_lite_status": c.da_lite_status or "PENDING", "da_lite_checked_at": c.da_lite_checked_at, "da_lite_remarks": c.da_lite_remarks, "da_lite_photo_url": c.da_lite_photo_url }, store={ "handover_date": store_ho.handover_date if store_ho else None, "is_locked": store_ho.is_locked if store_ho else False, "signed_store_person": store_ho.signed_store_person if store_ho else False, "signed_yard_supervisor": store_ho.signed_yard_supervisor if store_ho else False, "store_keeper_name": users_map.get(store_ho.store_keeper_id) if store_ho and store_ho.store_keeper_id else None, "procurement_status": c.procurement_status or "PENDING" }, current_state={ "current_status": c.current_status, "tracking_status": c.tracking_status, "production_hold_status": c.production_hold_status or "NONE", "current_station_id": c.current_station_id, "current_station_name": current_station_name or ("Dispatched" if c.tracking_status == 'DISPATCHED' else "Yard"), "stage_label": stage_label, "progress_percent": progress_percent, "station_arrived_at": c.station_arrived_at, "scheduled_induction_date": c.scheduled_induction_date, "dispatched_at": c.dispatched_at }, production_stations=production_stations, production_holds=[ { "hold_id": h.hold_id, "hold_type": h.hold_type, "hold_reason": h.hold_reason, "status": h.status, "raised_by": users_map.get(h.raised_by, "Staff"), "raised_at": h.raised_at, "resolved_by": users_map.get(h.resolved_by), "resolved_at": h.resolved_at, "resolution_notes": h.resolution_notes } for h in holds ], qa_inspections=qa_inspections_list, fvi_testing={ "checksheet_id": thermal_check.checksheet_id, "check_date": thermal_check.check_date, "all_ok": thermal_check.all_ok, "signature": thermal_check.signature, "photo_front": thermal_check.photo_front, "photo_rear": thermal_check.photo_rear, "photo_rh": thermal_check.photo_rh, "photo_lh": thermal_check.photo_lh, "submitted_at": thermal_check.submitted_at } if thermal_check else None, dispatch={ "gate_pass_id": gate_pass.gate_pass_id, "generated_at": gate_pass.generated_at, "driver_name": gate_pass.driver_name, "dispatch_authority": gate_pass.dispatch_authority, "fuel_before_ltr": float(gate_pass.fuel_before_ltr or 0), "filling_fuel_qty_ltr": float(gate_pass.filling_fuel_qty_ltr or 0), "items": gp_items, "dispatched_at": c.dispatched_at } if gate_pass else None, timeline=timeline, client_messages=client_msgs, order_spec=await resolve_chassis_order_spec(self.db, c) ) def _calculate_progress_percent( self, c: ChassisMaster, stations: Optional[List[StationVisitLog]] = None, gate_pass: Optional[DispatchGatePass] = None, thermal_check: Optional[DispatchThermalChecksheet] = None ) -> int: if c.tracking_status == 'DISPATCHED' or c.current_status == 'DISPATCHED' or c.dispatched_at: return 100 if gate_pass: return 95 if thermal_check and thermal_check.all_ok: return 90 if stations and len(stations) > 0: completed_count = sum(1 for s in stations if s.status == 'COMPLETED') in_progress_count = sum(1 for s in stations if s.status == 'AT_STATION') total = len(stations) # Production spans from 25% to 85% prod_fraction = (completed_count + (0.5 if in_progress_count else 0)) / total return int(25 + (prod_fraction * 60)) if c.handed_to_production_at or c.scheduled_induction_date: return 20 if c.da_lite_status == 'OK': return 15 if c.arrival_date: return 10 return 5 def _determine_stage_label( self, c: ChassisMaster, current_station: Optional[str], gate_pass: Optional[DispatchGatePass], thermal_check: Optional[DispatchThermalChecksheet] ) -> str: if c.tracking_status == 'DISPATCHED' or c.current_status == 'DISPATCHED': return "Dispatched & Delivered" if gate_pass: return "Ready for Gate Exit / Dispatched" if thermal_check: return "Final Inspection & Testing Completed" if c.production_hold_status == 'ON_HOLD': return f"Production On Hold ({current_station or 'Shopfloor'})" if current_station: return f"In Assembly - {current_station}" if c.handed_to_production_at: return "Handed over to Production Line" if c.da_lite_status == 'OK': return "Yard Inspected / Awaiting Induction" if c.arrival_date: return "Inwarded at Yard" return "Registered" def _build_timeline( self, c: ChassisMaster, store_ho: Optional[ChassisStoreHandover], stations: List[StationVisitLog], qa_inspections: List[Dict[str, Any]], thermal_check: Optional[DispatchThermalChecksheet], gate_pass: Optional[DispatchGatePass] ) -> List[TimelineMilestone]: timeline: List[TimelineMilestone] = [] # Milestone 1: Yard Inwarding & Reception yard_status = "COMPLETED" if c.arrival_date else "PENDING" yard_summary = f"Arrived on {c.arrival_date.strftime('%d-%b-%Y') if c.arrival_date else 'Pending'} at {c.yard_location or 'Yard'}" if c.da_lite_status == 'OK': yard_summary += " • DA Lite Inspection Passed" timeline.append( TimelineMilestone( stage_id="yard_inward", stage_name="1. Yard Reception & Inwarding", status=yard_status, timestamp=datetime.combine(c.arrival_date, datetime.min.time()).replace(tzinfo=timezone.utc) if c.arrival_date else None, summary=yard_summary, details={ "fuel_received": f"{c.fuel_receiving_qty} {('Litres' if c.fuel_type == 'DIESEL' else 'Kg') if c.fuel_receiving_qty else ''}", "condition": c.chassis_condition or "OK", "driver_name": c.driver_name, "inbound_vehicle": c.inbound_vehicle_no, "da_lite_status": c.da_lite_status } ) ) # Milestone 2: Store Handover & Allocation store_status = "PENDING" store_ts = None store_summary = "Awaiting store parts handover & allocation" if store_ho and (store_ho.signed_store_person or store_ho.is_locked): store_status = "COMPLETED" store_ts = datetime.combine(store_ho.handover_date, datetime.min.time()).replace(tzinfo=timezone.utc) if store_ho.handover_date else None store_summary = f"Parts handover completed on {store_ho.handover_date.strftime('%d-%b-%Y') if store_ho.handover_date else 'Verified'}" elif c.handed_to_production_at: store_status = "COMPLETED" store_ts = c.handed_to_production_at store_summary = f"Handed to production on {c.handed_to_production_at.strftime('%d-%b-%Y')}" timeline.append( TimelineMilestone( stage_id="store_handover", stage_name="2. Store Handover & Material Allocation", status=store_status, timestamp=store_ts, summary=store_summary, details={ "procurement_status": c.procurement_status, "scheduled_induction": c.scheduled_induction_date.strftime('%d-%b-%Y') if c.scheduled_induction_date else None } ) ) # Milestone 3: Assembly Line / Shopfloor Movement comp_stations = [s for s in stations if s.status == 'COMPLETED'] curr_stations = [s for s in stations if s.status == 'AT_STATION'] prod_status = "PENDING" prod_ts = None if len(comp_stations) == len(stations) and len(stations) > 0: prod_status = "COMPLETED" prod_ts = comp_stations[-1].completion_time prod_summary = f"All {len(stations)} stations successfully completed" elif curr_stations: prod_status = "IN_PROGRESS" prod_ts = curr_stations[0].arrival_time prod_summary = f"Currently at {curr_stations[0].station_name} ({len(comp_stations)}/{len(stations)} stations done)" elif comp_stations: prod_status = "IN_PROGRESS" prod_ts = comp_stations[-1].completion_time prod_summary = f"Completed {len(comp_stations)} of {len(stations)} stations" else: prod_summary = "Scheduled for induction on shopfloor" timeline.append( TimelineMilestone( stage_id="production", stage_name="3. Assembly Line & Stations", status=prod_status, timestamp=prod_ts, summary=prod_summary, details={ "completed_stations": len(comp_stations), "total_stations": len(stations), "current_station": curr_stations[0].station_name if curr_stations else None, "on_hold": c.production_hold_status == 'ON_HOLD' } ) ) # Milestone 4: QA Inspection & Quality Checkpoints qa_passed = [q for q in qa_inspections if q.get("overall_result") == 'PASSED'] qa_pending = [q for q in qa_inspections if q.get("overall_result") in ('PENDING', 'IN_PROGRESS')] qa_failed = [q for q in qa_inspections if q.get("overall_result") == 'FAILED'] qa_status = "PENDING" qa_ts = None if qa_passed and not qa_pending and not qa_failed: qa_status = "COMPLETED" qa_ts = qa_passed[-1].get("inspection_date") qa_summary = f"{len(qa_passed)} QA inspections cleared with 100% OK" elif qa_inspections: qa_status = "IN_PROGRESS" if not qa_failed else "ON_HOLD" qa_ts = qa_inspections[-1].get("inspection_date") qa_summary = f"{len(qa_passed)} cleared, {len(qa_pending)} in-progress" + (f", {len(qa_failed)} rework required" if qa_failed else "") else: qa_summary = "Station checkpoint QA inspections in queue" timeline.append( TimelineMilestone( stage_id="qa_inspection", stage_name="4. Quality Assurance & Audits", status=qa_status, timestamp=qa_ts, summary=qa_summary, details={ "passed_inspections": len(qa_passed), "total_inspections": len(qa_inspections) } ) ) # Milestone 5: Final Vehicle Inspection (FVI) & Thermal Testing fvi_status = "PENDING" fvi_ts = None fvi_summary = "Awaiting final road testing & thermal inspection" if thermal_check: fvi_status = "COMPLETED" if thermal_check.all_ok else "ON_HOLD" fvi_ts = thermal_check.submitted_at fvi_summary = "Final 26-point thermal inspection & 4-view bus photos approved" if thermal_check.all_ok else "Inspection recorded with open issues" timeline.append( TimelineMilestone( stage_id="fvi_testing", stage_name="5. Final Inspection & Thermal Testing", status=fvi_status, timestamp=fvi_ts, summary=fvi_summary, details={ "all_ok": thermal_check.all_ok if thermal_check else False, "signature": thermal_check.signature if thermal_check else None } ) ) # Milestone 6: Dispatch & Gate Pass disp_status = "PENDING" disp_ts = None disp_summary = "Pending final gate pass & vehicle dispatch" if c.dispatched_at or c.tracking_status == 'DISPATCHED': disp_status = "COMPLETED" disp_ts = c.dispatched_at or (gate_pass.generated_at if gate_pass else None) disp_summary = f"Dispatched on {disp_ts.strftime('%d-%b-%Y %I:%M %p') if disp_ts else 'Delivered'}" elif gate_pass: disp_status = "IN_PROGRESS" disp_ts = gate_pass.generated_at disp_summary = f"Gate Pass #{str(gate_pass.gate_pass_id)[:8].upper()} issued to driver {gate_pass.driver_name or ''}" timeline.append( TimelineMilestone( stage_id="dispatch", stage_name="6. Vehicle Dispatch & Delivery", status=disp_status, timestamp=disp_ts, summary=disp_summary, details={ "driver_name": gate_pass.driver_name if gate_pass else None, "dispatch_authority": gate_pass.dispatch_authority if gate_pass else None, "gate_pass_id": str(gate_pass.gate_pass_id) if gate_pass else None } ) ) return timeline def _generate_client_messages( self, c: ChassisMaster, model_name: str, dealer_name: str, order_no: str, stage_label: str, progress_percent: int, timeline: List[TimelineMilestone], current_station_name: Optional[str], gate_pass: Optional[DispatchGatePass] ) -> ClientUpdateMessages: today_str = datetime.now().strftime("%d-%b-%Y") arrival_str = c.arrival_date.strftime("%d-%b-%Y") if c.arrival_date else "N/A" # Generate progress bar representation (e.g. [██████░░░░] 60%) filled_bars = int(progress_percent / 10) unfilled_bars = 10 - filled_bars progress_bar = "█" * filled_bars + "░" * unfilled_bars # Highlight recent milestone completed_milestones = [m for m in timeline if m.status == 'COMPLETED'] last_milestone = completed_milestones[-1].stage_name if completed_milestones else "Yard Inwarding" # WhatsApp Template wa = ( f"🚍 *VEHICLE PRODUCTION STATUS UPDATE*\n" f"━━━━━━━━━━━━━━━━━━━━━━━━━━\n" f"*Chassis No:* {c.chassis_no}\n" f"*Model / Variant:* {model_name}\n" f"*Client / Dealer:* {dealer_name}\n" f"*Order Reference:* #{order_no}\n" f"━━━━━━━━━━━━━━━━━━━━━━━━━━\n" f"📍 *Current Status:* {stage_label}\n" f"📊 *Progress:* [{progress_bar}] *{progress_percent}%*\n" f"📅 *As of:* {today_str}\n\n" f"*Milestone Highlights:*\n" f"• Inward Date: {arrival_str}\n" f"• Latest Completed Stage: {last_milestone}\n" ) if current_station_name: wa += f"• Active Operation: Fitting & assembly at *{current_station_name}*\n" if gate_pass: wa += f"• Gate Pass: #{str(gate_pass.gate_pass_id)[:8].upper()} (Driver: {gate_pass.driver_name or 'Assigned'})\n" elif progress_percent >= 90: wa += f"• Final inspection cleared. Preparing dispatch documentation.\n" else: wa += f"• Assembly proceeding smoothly adhering to strict QA standards.\n" wa += ( f"\nShould you have any inquiries, please contact your account representative.\n" f"━━━━━━━━━━━━━━━━━━━━━━━━━━\n" f"_Astute Bridge · Automobile Manufacturing ERP_" ) # Email Subject & Body email_subj = f"Vehicle Production Status Update - Chassis #{c.chassis_no} ({model_name})" email_body = ( f"Dear {dealer_name},\n\n" f"We are pleased to provide you with the latest progress update for your vehicle currently in production.\n\n" f"VEHICLE & ORDER DETAILS:\n" f"--------------------------------------------------\n" f"• Chassis Number : {c.chassis_no}\n" f"• Vehicle Model : {model_name} ({c.fuel_type or 'Diesel'})\n" f"• Order Reference: #{order_no}\n" f"• Inward Date : {arrival_str}\n\n" f"CURRENT PRODUCTION STATUS:\n" f"--------------------------------------------------\n" f"• Stage Status : {stage_label}\n" f"• Progress Level : {progress_percent}% Completed\n" f"• Latest Milestone: {last_milestone}\n" ) if current_station_name: email_body += f"• Current Station: {current_station_name}\n" if gate_pass: email_body += f"• Dispatch Pass : Issued (Driver: {gate_pass.driver_name or 'N/A'})\n" email_body += ( f"\nOur production and quality assurance teams ensure every step meets rigorous specifications.\n\n" f"Please feel free to reach out to us if you need further details.\n\n" f"Warm regards,\n" f"Customer Success & Operations Team\n" f"Bhagirath ERP / Astute Bridge\n" ) # Brief SMS sms = ( f"Status Update: Chassis {c.chassis_no} ({model_name}) is currently at '{stage_label}' " f"with {progress_percent}% overall completion as of {today_str}. - Bhagirath ERP" ) return ClientUpdateMessages( whatsapp_text=wa, email_subject=email_subj, email_body=email_body, sms_text=sms )