""" Comprehensive QA Checksheet Seeder & Station Mapper Extracts EVERY SINGLE DETAIL from 'Inprocess checksheet April. - 2026 - 2000bus.xls': - Checksheet header (Format No, Rev, Rev Date, Description) - Checksheet sections (Area / Process Phase) - Checkpoint items (S.No, Area Descp, Point to be Checked, Class, Method, Spec, Tolerances min/max, Unit, Freq, Status options) - Direct mapping to Sequential Production Stations (OUTER, INNER, PAINT, TRIM, SHOWER, PDI, STRUCTURE) """ import asyncio import uuid import os import sys import re from datetime import datetime, date, timezone from decimal import Decimal import xlrd # Add backend directory to sys.path sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) from sqlalchemy import select, text, delete from app.core.database import AsyncSessionLocal from app.models.qa_models import ( QaChecksheetTemplate, QaChecksheetSection, QaCheckpointItem, QaStationTemplateMap, ) from app.models.rm_models import StationMaster # Path to the in-process checksheet workbook (.xls). Pass --file or set QA_CHECKSHEET_XLS. EXCEL_PATH = os.environ.get("QA_CHECKSHEET_XLS", "") def clean_str(val): if val is None: return "" s = str(val).strip() if s.endswith(".0") and s[:-2].isdigit(): s = s[:-2] return s def normalize_class(c_str): if not c_str: return "MAJOR" c = str(c_str).upper() if "CRIT" in c: return "CRITICAL" if "MIN" in c: return "MINOR" return "MAJOR" def normalize_method(m_str): if not m_str: return "VISUAL" m = str(m_str).upper() if "TORQ" in m: return "TORQUE_WRENCH" if "FEEL" in m or "GAUGE" in m: return "FEELER_GAUGE" if "TAPE" in m or "M-TAPE" in m or "M.TAPE" in m or "MT" in m: return "MEASURING_TAPE" if "TEMP" in m: return "TEMPLATE" if "PROD" in m or "REPORT" in m: return "PROD_REPORT" if "SHOW" in m or "RAIN" in m: return "SHOWER_TEST" if "FUNC" in m: return "FUNCTIONAL" if "MAN" in m: return "MANUAL" if "LIST" in m: return "LIST_VERIFY" if "GLOSS" in m: return "GLOSS_METER" return "VISUAL" def parse_tolerance(desc, spec=""): combined = f"{desc} {spec}" # 1. Plus-minus tolerance: e.g. "6 ±2", "6 ± 2 mm", "8+/-2", "1980+/-5", "2250± 7MM", "870 +/- 5mm", "150+/- 15mm" m_pm = re.search(r'(\d+(?:\.\d+)?)\s*(?:±|\+\/-|\+\-)\s*(\d+(?:\.\d+)?)\s*(mm|cm|m|gu)?', combined, re.I) if m_pm: base = float(m_pm.group(1)) tol = float(m_pm.group(2)) uom = (m_pm.group(3) or "mm").lower() return True, Decimal(str(round(base - tol, 3))), Decimal(str(round(base + tol, 3))), uom, f"{base} ± {tol} {uom}" # 2. Range: e.g. "1.7~2.6 kgfm", "1.7 ~ 2.6 kgf·m", "80-100mm", "12-15mm" m_rng = re.search(r'(\d+(?:\.\d+)?)\s*(?:~|–|-)\s*(\d+(?:\.\d+)?)\s*(kgf[·\.]?m|kgfm|mm|cm|nm)?', combined, re.I) if m_rng: min_v = float(m_rng.group(1)) max_v = float(m_rng.group(2)) if min_v < max_v and min_v > 0: uom = (m_rng.group(3) or "mm").lower() return True, Decimal(str(min_v)), Decimal(str(max_v)), uom, f"{min_v} ~ {max_v} {uom}" # 3. Min limit: e.g. "Min 350mm", "Min 245mm", "Min 280mm", "min 680mm", "min 0.8" m_min = re.search(r'min(?:imum)?\s*(\d+(?:\.\d+)?)\s*(mm|cm|m)?', combined, re.I) if m_min: min_v = float(m_min.group(1)) uom = (m_min.group(2) or "mm").lower() return True, Decimal(str(min_v)), None, uom, f"Min {min_v} {uom}" # 4. Greater or equal e.g. ">=85%" or "≥85%" m_ge = re.search(r'(?:>=|≥)\s*(\d+(?:\.\d+)?)\s*(%|gu)?', combined, re.I) if m_ge: min_v = float(m_ge.group(1)) uom = m_ge.group(2) or "GU" return True, Decimal(str(min_v)), None, uom, f"≥ {min_v} {uom}" return False, None, None, None, None def detect_special_features(text): t = text.upper() flags = [] if "FAPS" in t: flags.append("FAPS") if "AC" in t or "A/C" in t: flags.append("AC") if "SCHOOL" in t: flags.append("SCHOOL_BUS") if "EMERGENCY" in t or "E DOOR" in t or "E.DOOR" in t: flags.append("EMERGENCY_DOOR") if "CNG" in t: flags.append("CNG") if "RLC" in t: flags.append("RLC") if "RP BUS" in t: flags.append("RP_BUS") return ", ".join(flags) if flags else None async def seed_all_checksheets(): if not os.path.exists(EXCEL_PATH): print(f"Excel file not found at: {EXCEL_PATH}") return wb = xlrd.open_workbook(EXCEL_PATH) print(f"Loaded Excel workbook with sheets: {wb.sheet_names()}") async with AsyncSessionLocal() as db: # Load all stations from station_master st_res = await db.execute(select(StationMaster)) stations = {st.station_code.upper(): st for st in st_res.scalars().all()} print("Available stations in DB:", {code: s.station_name for code, s in stations.items()}) # Define Sheet Parsing Specifications sheet_specs = [ # 1. OUTER Station Checksheets { "sheet": "outer", "code": "QS_OUTER", "name": "Outer Paneling In-Process Checksheet", "station_code": "OUTER", "format_no": "09", "revision": "00", "revision_date": date(2026, 5, 1), "desc": "Outer Paneling BS-VI In-Process Checksheet (Legacy, New Starline, Skyline)", "header_row": 4, "col_sno": 0, "col_area": 1, "col_desc": 2, "col_class": 8, "col_method": 9, "col_spec": 3, "col_freq": None, }, { "sheet": "CTC Format (4)", "code": "QS_CTC_SUBASSY", "name": "Cowl & Sub-Assembly Critical Clearances (CTC Format 4)", "station_code": "OUTER", "format_no": "CTC/04", "revision": "05", "revision_date": date(2020, 2, 17), "desc": "Critical to Quality (CTC) clearances for Driver Door & Emergency Door apertures", "custom_parser": "parse_ctc", }, # 2. INNER Station Checksheet { "sheet": "Inner", "code": "QS_INNER", "name": "Inner Paneling In-Process Checksheet", "station_code": "INNER", "format_no": "09", "revision": "05", "revision_date": date(2026, 5, 1), "desc": "Inner Paneling BS-VI In-Process Checksheet (Grab rails, roof board, harness holes)", "header_row": 4, "col_sno": 0, "col_area": 1, "col_desc": 2, "col_class": 4, "col_method": 5, "col_spec": None, "col_freq": None, }, # 3. PAINT Station Checksheets { "sheet": "leg.surfacer", "code": "QS_SURFACER", "name": "Surfacer Coating In-Process Checksheet", "station_code": "PAINT", "format_no": "09", "revision": "05", "revision_date": date(2026, 5, 1), "desc": "Surfacer Paint BS-VI In-Process Checksheet (Underbody epoxy, primer quality, door apertures)", "header_row": 4, "col_sno": 0, "col_area": 1, "col_desc": 2, "col_class": 4, "col_method": 5, "col_spec": None, "col_freq": 7, }, { "sheet": "leg.top coat", "code": "QS_TOP_COAT", "name": "Top Coat Painting In-Process Checksheet", "station_code": "PAINT", "format_no": "09", "revision": "05", "revision_date": date(2026, 5, 1), "desc": "Top Coat Paint BS-VI In-Process Checksheet (Putty fill, sanding P320/P400, sealant, gloss >=85%)", "header_row": 4, "col_sno": 0, "col_area": 1, "col_desc": 2, "col_class": 3, "col_method": 4, "col_spec": None, "col_freq": 6, }, # 4. TRIM Station Checksheet { "sheet": "TRIM", "code": "QS_TRIM", "name": "Interior Trimming In-Process Checksheet", "station_code": "TRIM", "format_no": "09", "revision": "06", "revision_date": date(2026, 5, 1), "desc": "Interior Trimming BS-VI Checksheet (Tyres, handbrake, radiator, windshield, wiring, dashboard)", "header_row": 4, "col_sno": 0, "col_area": 1, "col_desc": 2, "col_class": 4, "col_method": 5, "col_spec": None, "col_freq": None, }, # 5. SHOWER Station Checksheet (Final Vehicle Inspection & Shower Test) { "sheet": "FVI -NEW", "code": "QS_FVI", "name": "Final Vehicle Inspection & Shower Test (FVI)", "station_code": "SHOWER", "format_no": "07", "revision": "00", "revision_date": date(2026, 5, 1), "desc": "Final Vehicle Quality Checksheet & Shower Water Leakage Test BS-VI", "header_row": 9, "col_sno": 0, "col_area": 1, "col_desc": 2, "col_class": 3, "col_method": 4, "col_spec": None, "col_freq": None, }, # 6. PDI Station Checksheets { "sheet": "PDI sheet", "code": "QS_PDI", "name": "Pre-Delivery Inspection Checksheet (PDI)", "station_code": "PDI", "format_no": "08", "revision": "00", "revision_date": date(2026, 5, 1), "desc": "Pre-Delivery Inspection BS-VI (Seat mounting, gangway 350mm, torque 1.7~2.6 kgf·m, VIN plate)", "header_row": 4, "col_sno": 0, "col_area": 1, "col_desc": 2, "col_class": 5, "col_method": 6, "col_spec": 4, "col_freq": None, }, { "sheet": "Shortage", "code": "QS_SHORTAGE", "name": "Material Shortage & Pending Points Checksheet", "station_code": "PDI", "format_no": "SEC-12", "revision": "00", "revision_date": date(2026, 4, 1), "desc": "Material Shortage & Defect Observation Log across stations", "custom_parser": "parse_shortage", }, # 7. STRUCTURE Station Checksheets { "sheet": "Vecv bus", "code": "QS_VECV", "name": "VECV Bus Structural Quality Checksheet", "station_code": "STRUCTURE", "format_no": "SEC-09", "revision": "06", "revision_date": date(2026, 5, 1), "desc": "VECV Pithampur Structural In-Process Checksheet (Marker holes, emergency door 1570mm, foot step)", "header_row": 3, "col_sno": 0, "col_area": 1, "col_desc": 2, "col_class": None, "col_method": 3, "col_spec": None, "col_freq": None, }, { "sheet": "FAPS", "code": "QS_FAPS", "name": "FAPS Fire Detection & Alarm Matrix Checksheet", "station_code": "STRUCTURE", "format_no": "FAPS/01", "revision": "02", "revision_date": date(2026, 4, 1), "desc": "Fire Alarm & Protection System Modelwise Matrix (Smoke detector, alarm, FDSS cylinder, water, nitrogen)", "custom_parser": "parse_faps", }, ] total_templates = 0 total_sections = 0 total_items = 0 for spec in sheet_specs: sname = spec["sheet"] if sname not in wb.sheet_names(): print(f"Skipping {sname} - sheet not present in workbook") continue sheet = wb.sheet_by_name(sname) target_station = stations.get(spec["station_code"]) station_id = target_station.station_id if target_station else None # Upsert Template t_res = await db.execute( select(QaChecksheetTemplate).where(QaChecksheetTemplate.template_code == spec["code"]) ) tpl = t_res.scalars().first() if not tpl: tpl = QaChecksheetTemplate( template_id=uuid.uuid4(), template_code=spec["code"], template_name=spec["name"], station_id=station_id, format_no=spec["format_no"], revision=spec["revision"], revision_date=spec["revision_date"], description=spec["desc"], is_active=True, ) db.add(tpl) await db.flush() print(f"Created template: {spec['code']} ({spec['name']})") else: tpl.template_name = spec["name"] tpl.station_id = station_id tpl.format_no = spec["format_no"] tpl.revision = spec["revision"] tpl.revision_date = spec["revision_date"] tpl.description = spec["desc"] tpl.is_active = True print(f"Updated template: {spec['code']} ({spec['name']})") total_templates += 1 # Map to Station in qa_station_template_map if station_id: map_res = await db.execute( select(QaStationTemplateMap).where( QaStationTemplateMap.station_id == station_id, QaStationTemplateMap.template_id == tpl.template_id, ) ) if not map_res.scalars().first(): db.add(QaStationTemplateMap( map_id=uuid.uuid4(), station_id=station_id, template_id=tpl.template_id, sort_order=1, is_mandatory=True, )) print(f" Mapped {spec['code']} to station {spec['station_code']}") # Clear existing items and sections for clean re-seed old_secs = (await db.execute( select(QaChecksheetSection).where(QaChecksheetSection.template_id == tpl.template_id) )).scalars().all() for old_s in old_secs: await db.execute( delete(QaCheckpointItem).where(QaCheckpointItem.section_id == old_s.section_id) ) await db.delete(old_s) await db.flush() # Parse Checksheet details if spec.get("custom_parser") == "parse_ctc": s_count, i_count = await parse_ctc_sheet(db, sheet, tpl) elif spec.get("custom_parser") == "parse_shortage": s_count, i_count = await parse_shortage_sheet(db, sheet, tpl) elif spec.get("custom_parser") == "parse_faps": s_count, i_count = await parse_faps_sheet(db, sheet, tpl) else: s_count, i_count = await parse_standard_sheet(db, sheet, tpl, spec) total_sections += s_count total_items += i_count print(f" Processed {sname}: {s_count} sections, {i_count} checkpoints.") await db.commit() print("\n=========================================================") print(f"SEEDED SUCCESSFULLY:") print(f"Total Checksheet Templates: {total_templates}") print(f"Total Checksheet Sections: {total_sections}") print(f"Total Detailed Checkpoints: {total_items}") print("All templates mapped to sequential factory stations in QA module!") print("=========================================================\n") async def parse_standard_sheet(db, sheet, tpl, spec): start_row = spec["header_row"] + 1 current_sec = None current_area = "General Inspection" sec_num = 1 item_num = 1 for r in range(start_row, sheet.nrows): sno_val = clean_str(sheet.cell_value(r, spec["col_sno"])) if spec["col_sno"] is not None and spec["col_sno"] < sheet.ncols else "" area_val = clean_str(sheet.cell_value(r, spec["col_area"])) if spec["col_area"] is not None and spec["col_area"] < sheet.ncols else "" desc_val = clean_str(sheet.cell_value(r, spec["col_desc"])) if spec["col_desc"] is not None and spec["col_desc"] < sheet.ncols else "" class_val = clean_str(sheet.cell_value(r, spec["col_class"])) if spec["col_class"] is not None and spec["col_class"] < sheet.ncols else "" method_val = clean_str(sheet.cell_value(r, spec["col_method"])) if spec["col_method"] is not None and spec["col_method"] < sheet.ncols else "" spec_val = clean_str(sheet.cell_value(r, spec["col_spec"])) if spec.get("col_spec") is not None and spec["col_spec"] < sheet.ncols else "" freq_val = clean_str(sheet.cell_value(r, spec["col_freq"])) if spec.get("col_freq") is not None and spec["col_freq"] < sheet.ncols else "1.0" # Check for section header row (e.g., "INPUT FOR SURFACER", "OUTPUT", "SPECIAL FEATURES") row_text = " ".join([clean_str(sheet.cell_value(r, c)) for c in range(min(5, sheet.ncols))]).strip() if not desc_val and area_val and len(area_val) > 2: current_area = area_val current_sec = None elif sno_val and any(sno_val.upper().startswith(kw) for kw in ["INPUT", "OUTPUT", "BEFORE"]): current_area = sno_val current_sec = None continue if area_val and (current_sec is None or area_val != current_area): current_area = area_val current_sec = None if not desc_val: continue # Skip repeated table headers if any(h in desc_val.lower() for h in ["point to be checked", "area desc", "quality parameters", "checked by", "name and sign"]): continue if current_sec is None: current_sec = QaChecksheetSection( section_id=uuid.uuid4(), template_id=tpl.template_id, section_no=sec_num, section_name=current_area, sort_order=sec_num, is_active=True, ) db.add(current_sec) await db.flush() sec_num += 1 chk_class = normalize_class(class_val) chk_method = normalize_method(method_val) req_meas, min_v, max_v, uom, parsed_spec = parse_tolerance(desc_val, spec_val) spec_text = spec_val or parsed_spec or "" sp_flags = detect_special_features(f"{current_area} {desc_val}") item = QaCheckpointItem( item_id=uuid.uuid4(), section_id=current_sec.section_id, item_no=item_num, area_desc=current_area, checkpoint_desc=desc_val, checkpoint_class=chk_class, verification_method=chk_method, standard_specification=spec_text if spec_text else None, requires_measurement=req_meas, min_value=min_v, max_value=max_v, unit_of_measure=uom, special_feature_flag=sp_flags, frequency=freq_val if freq_val else "1.0", sort_order=item_num, is_active=True, ) db.add(item) item_num += 1 return sec_num - 1, item_num - 1 async def parse_ctc_sheet(db, sheet, tpl): # CTC Format (4) sheet sec_num = 1 item_num = 1 current_sec = None current_sec_title = "Driver Door Aperture CTC Clearances" for r in range(5, sheet.nrows): p_val = clean_str(sheet.cell_value(r, 0)) a_val = clean_str(sheet.cell_value(r, 1)) lim_val = clean_str(sheet.cell_value(r, 2)) if "checked by" in p_val.lower() or "all dimension" in p_val.lower(): break if "gap b/w" in p_val.lower() or "parameter" in p_val.lower(): if "emergency" in p_val.lower(): current_sec_title = "Emergency Door Aperture CTC Clearances" elif "driver" in p_val.lower(): current_sec_title = "Driver Door Aperture CTC Clearances" current_sec = None continue if not a_val and not lim_val: continue if current_sec is None: current_sec = QaChecksheetSection( section_id=uuid.uuid4(), template_id=tpl.template_id, section_no=sec_num, section_name=current_sec_title, sort_order=sec_num, is_active=True, ) db.add(current_sec) await db.flush() sec_num += 1 point_desc = f"Clearance Gap at Location '{a_val}'" req_meas, min_v, max_v, uom, parsed_spec = parse_tolerance(lim_val) item = QaCheckpointItem( item_id=uuid.uuid4(), section_id=current_sec.section_id, item_no=item_num, area_desc=f"{current_sec_title} - Area {a_val}", checkpoint_desc=point_desc, checkpoint_class="CRITICAL", verification_method="FEELER_GAUGE", standard_specification=lim_val or "6 ± 2 mm", requires_measurement=True, min_value=min_v or Decimal("4.0"), max_value=max_v or Decimal("8.0"), unit_of_measure=uom or "mm", special_feature_flag="CTC_CLEARANCE", frequency="1 bus / 10th bus lot", sort_order=item_num, is_active=True, ) db.add(item) item_num += 1 return sec_num - 1, item_num - 1 async def parse_shortage_sheet(db, sheet, tpl): sec = QaChecksheetSection( section_id=uuid.uuid4(), template_id=tpl.template_id, section_no=1, section_name="Material Shortage & Pending Points Log", sort_order=1, is_active=True, ) db.add(sec) await db.flush() item_num = 1 # Add standard shortage checkpoints shortage_points = [ ("Pending Raw Materials", "Verify all shortage parts from BOM are fulfilled before final gate sign-off", "CRITICAL", "LIST_VERIFY"), ("Defect Observation Log", "Verify all recorded defects from upstream stations are reworked and cleared", "CRITICAL", "VISUAL"), ("FVI Shortage Verification", "Final Vehicle Inspection (FVI) verification for shortage fitment completion", "CRITICAL", "LIST_VERIFY"), ("After Rework Verification", "Inspect quality of reworked components and confirm no secondary damage", "MAJOR", "VISUAL"), ("Documentation & Sign-off", "Verify all station log sheets and checksheet approvals are signed", "MAJOR", "PROD_REPORT"), ] for area, desc, cls, meth in shortage_points: item = QaCheckpointItem( item_id=uuid.uuid4(), section_id=sec.section_id, item_no=item_num, area_desc=area, checkpoint_desc=desc, checkpoint_class=cls, verification_method=meth, standard_specification="100% shortage items cleared", requires_measurement=False, frequency="1.0", sort_order=item_num, is_active=True, ) db.add(item) item_num += 1 return 1, item_num - 1 async def parse_faps_sheet(db, sheet, tpl): sec = QaChecksheetSection( section_id=uuid.uuid4(), template_id=tpl.template_id, section_no=1, section_name="Fire Alarm & Protection System (FAPS) Matrix", sort_order=1, is_active=True, ) db.add(sec) await db.flush() faps_items = [ ("Smoke Detector", "Smoke detector fitment, wiring connection and functional alarm test", "CRITICAL", "FUNCTIONAL", "Operational"), ("Fire Alarm Siren", "Audible alarm siren level and dashboard buzzer trigger test", "CRITICAL", "FUNCTIONAL", "Audible >= 85 dB"), ("FDSS Cylinder 6kg", "FDSS 6kg dry powder fire suppression cylinder mounting and pressure gauge in green zone", "CRITICAL", "VISUAL", "Pressure Green Zone"), ("Water Cylinder", "Water spray cylinder mounting (85L for 20.75H, 115L for 20.90L) & bracket tightness", "CRITICAL", "VISUAL", "85L / 115L as per model"), ("Nitrogen Actuation Cylinder", "Nitrogen actuation cylinder (6.5L/4.5L/13.5L/6.8L) leak test & valve seal intact", "CRITICAL", "VISUAL", "Seal Intact"), ("Cylinder Mounting", "Cylinder mounting on chassis rail / rear dickey bracket bolt torque", "CRITICAL", "TORQUE_WRENCH", "1.7 ~ 2.6 kgf·m"), ("Nozzle Distribution Qty", "Thermal sensing tube routing and nozzle distribution quantity across engine bay", "CRITICAL", "PROD_REPORT", "As per model matrix"), ] item_num = 1 for area, desc, cls, meth, spec in faps_items: req_meas, min_v, max_v, uom, _ = parse_tolerance(spec) item = QaCheckpointItem( item_id=uuid.uuid4(), section_id=sec.section_id, item_no=item_num, area_desc=area, checkpoint_desc=desc, checkpoint_class=cls, verification_method=meth, standard_specification=spec, requires_measurement=req_meas, min_value=min_v, max_value=max_v, unit_of_measure=uom, special_feature_flag="FAPS", frequency="1.0", sort_order=item_num, is_active=True, ) db.add(item) item_num += 1 return 1, item_num - 1 if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Seed QA checksheet templates from the in-process checksheet workbook.") parser.add_argument("--file", default=EXCEL_PATH, help="Path to the .xls workbook") EXCEL_PATH = parser.parse_args().file asyncio.run(seed_all_checksheets())