import asyncio import uuid import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from typing import List, Dict, Any from sqlalchemy import select, text from app.core.database import AsyncSessionLocal, engine, Base from app.models.location_models import ChassisLocationMaster, StationQaChecklistItemMaster from app.models.rm_models import StationMaster # Import all models so Base.metadata can resolve all foreign key references import app.models.core_models import app.models.access_models import app.models.yard_models import app.models.sales_models import app.models.bom_models DEFAULT_LOCATIONS: List[Dict[str, Any]] = [ { "code": "CHASSIS_INWARD", "name": "Chassis Inward", "seq": 1, "desc": "Initial chassis receipt and inward gate inspection", "stations": [ {"code": "INWARD_GATE", "name": "Inward Gate Scan", "seq": 1, "qa": True}, {"code": "INWARD_VISUAL", "name": "Visual Inspection Bay", "seq": 2, "qa": True}, ] }, { "code": "UNIT_2", "name": "Unit-2", "seq": 2, "desc": "Unit-2 sub-assembly and pre-line preparation", "stations": [ {"code": "U2_SUB_ASSY_1", "name": "Sub-Assembly Bay 1", "seq": 1, "qa": False}, {"code": "U2_SUB_ASSY_2", "name": "Sub-Assembly Bay 2", "seq": 2, "qa": False}, ] }, { "code": "PRODUCTION", "name": "Production", "seq": 3, "desc": "Main vehicle assembly line", "stations": [ {"code": "PROD_INNER", "name": "Inner Body Structure", "seq": 1, "qa": True}, {"code": "PROD_OUTER", "name": "Outer Panelling", "seq": 2, "qa": True}, {"code": "PROD_PAINTING", "name": "Paint Shop & Coating", "seq": 3, "qa": True}, {"code": "PROD_TRIMMING", "name": "Interior Trimming & Seats", "seq": 4, "qa": True}, {"code": "PROD_ELECTRICAL", "name": "Electrical Harnessing", "seq": 5, "qa": True}, ] }, { "code": "FVI", "name": "FVI", "seq": 4, "desc": "Final Vehicle Inspection line", "stations": [ {"code": "FVI_LINE_1", "name": "FVI Main Inspection Line", "seq": 1, "qa": True}, {"code": "FVI_REWORK", "name": "FVI Rework Bay", "seq": 2, "qa": True}, ] }, { "code": "PDI", "name": "PDI", "seq": 5, "desc": "Pre-Delivery Inspection", "stations": [ {"code": "PDI_BAY_1", "name": "PDI Quality Bay", "seq": 1, "qa": True}, {"code": "PDI_CLEAN", "name": "Final Wash & Detailing", "seq": 2, "qa": False}, ] }, { "code": "DISPATCH", "name": "Dispatch", "seq": 6, "desc": "Final dispatch and handover yard", "stations": [ {"code": "DISPATCH_BAY", "name": "Dispatch Bay", "seq": 1, "qa": True}, {"code": "DISPATCH_GATE", "name": "Outward Main Gate", "seq": 2, "qa": False}, ] } ] async def seed_locations_and_stations(): async with engine.begin() as conn: await conn.run_sync(lambda c: Base.metadata.create_all(c, checkfirst=True)) # Ensure column additions on existing tables await conn.execute(text("ALTER TABLE station_master ADD COLUMN IF NOT EXISTS location_id UUID REFERENCES chassis_location_master(location_id);")) await conn.execute(text("ALTER TABLE chassis_master ADD COLUMN IF NOT EXISTS current_location_id UUID REFERENCES chassis_location_master(location_id);")) await conn.execute(text("ALTER TABLE chassis_master ADD COLUMN IF NOT EXISTS current_station_id UUID REFERENCES station_master(station_id);")) async with AsyncSessionLocal() as session: print("Seeding Factory Locations & Stations...") for loc_data in DEFAULT_LOCATIONS: loc_code: str = loc_data["code"] loc_name: str = loc_data["name"] loc_seq: int = loc_data["seq"] loc_desc: str = loc_data["desc"] stations_list: List[Dict[str, Any]] = loc_data["stations"] res = await session.execute( select(ChassisLocationMaster).where(ChassisLocationMaster.location_code == loc_code) ) location = res.scalar_one_or_none() if not location: location = ChassisLocationMaster( location_id=uuid.uuid4(), location_code=loc_code, location_name=loc_name, sequence_no=loc_seq, description=loc_desc, is_active=True ) session.add(location) await session.flush() print(f" + Added Location: [{location.sequence_no}] {location.location_name}") else: location.sequence_no = loc_seq location.location_name = loc_name print(f" ~ Updated Location: [{location.sequence_no}] {location.location_name}") for stn_data in stations_list: stn_code: str = stn_data["code"] stn_name: str = stn_data["name"] stn_seq: int = stn_data["seq"] stn_qa: bool = stn_data["qa"] res_stn = await session.execute( select(StationMaster).where(StationMaster.station_code == stn_code) ) station = res_stn.scalar_one_or_none() if not station: station = StationMaster( station_id=uuid.uuid4(), location_id=location.location_id, station_code=stn_code, station_name=stn_name, sequence_no=stn_seq, requires_qa=stn_qa, is_active=True ) session.add(station) await session.flush() print(f" + Added Station: [{station.sequence_no}] {station.station_name}") else: station.location_id = location.location_id station.sequence_no = stn_seq station.requires_qa = stn_qa print(f" ~ Updated Station: [{station.sequence_no}] {station.station_name}") if stn_qa: res_qa = await session.execute( select(StationQaChecklistItemMaster).where(StationQaChecklistItemMaster.station_id == station.station_id) ) qa_items = res_qa.scalars().all() if not qa_items: sample_item = StationQaChecklistItemMaster( item_id=uuid.uuid4(), station_id=station.station_id, zone_name="General QA", checkpoint_text=f"Verify {station.station_name} quality & specifications", specification="Visual & Functional Check OK", sequence_no=1, is_active=True ) session.add(sample_item) print(f" + Added sample QA item for station: {station.station_name}") await session.commit() print("Seeding completed successfully!") if __name__ == "__main__": asyncio.run(seed_locations_and_stations())