""" Production Report — data aggregation + Excel generation. Queries all chassis with confirmed orders, joins across order details, special features, chassis inwarding data, and manual entries to produce the full production report — both as JSON (for the web viewer) and as a styled .xlsx download. The special-features section is **dynamic**: columns are determined at query time by the current set of active SpecialFeatureMaster rows, so additions/removals in Production Masters are reflected immediately. """ import io import uuid from datetime import date, datetime, timezone from typing import Optional import openpyxl from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased from app.models.yard_models import ( ChassisMaster, VehicleModelMaster, ) from app.models.sales_models import ( CustomerMaster, ClientMaster, TransportMaster, ) from app.models.dealer_models import ( DealerOrder, DealerOrderVehicle, DealerOrderVehicleChassis, DealerOrderVehicleFeature, BodyTypeMaster, CustomerSegmentMaster, SpecialFeatureMaster, SpecialFeatureOptionMaster, ) from app.models.rm_models import StationMaster from app.models.location_models import ChassisLocationMaster from app.models.production_report_models import ProductionReportManualEntry # ───────────────────────────────────────────────────────────────────── # 1. SPECIAL FEATURE COLUMN STRUCTURE # ───────────────────────────────────────────────────────────────────── def is_excluded_feature_value(val: Optional[str]) -> bool: """Return True if value represents 'None', 'Not Required', 'N/A', etc. (case-insensitive).""" if not val: return True cleaned = val.strip().lower().replace('_', ' ').replace('-', ' ') cleaned_no_dot = ' '.join(cleaned.replace('.', '').split()) cleaned_orig = ' '.join(val.strip().lower().split()) excluded_set = { 'none', 'not required', 'na', 'n/a', 'no', 'nil', 'not applicable', '-', '—', '--', } return cleaned_no_dot in excluded_set or cleaned_orig in excluded_set or cleaned in excluded_set async def get_special_feature_columns(db: AsyncSession) -> list[dict]: """ Return the dynamic column structure for special features. Each feature becomes a group; DROPDOWN features expand into sub-columns. Options representing 'None' or 'Not Required' are excluded from the columns. """ features = (await db.execute( select(SpecialFeatureMaster) .where(SpecialFeatureMaster.is_active == True) .order_by(SpecialFeatureMaster.sequence_no) )).scalars().all() options = (await db.execute( select(SpecialFeatureOptionMaster) .where(SpecialFeatureOptionMaster.is_active == True) .order_by(SpecialFeatureOptionMaster.sequence_no) )).scalars().all() opts_by_feature: dict[uuid.UUID, list] = {} for o in options: opt_val = (o.option_value or '').strip() if is_excluded_feature_value(opt_val): continue opts_by_feature.setdefault(o.feature_id, []).append({ 'option_id': str(o.option_id), 'option_value': opt_val, }) result = [] for f in features: feature_opts = opts_by_feature.get(f.feature_id, []) # If DROPDOWN feature has no options remaining (e.g. only had 'Not Required'), skip feature column if f.input_type == 'DROPDOWN' and not feature_opts: continue result.append({ 'feature_id': str(f.feature_id), 'feature_name': f.feature_name, 'section': f.section, 'input_type': f.input_type, 'options': feature_opts, }) return result # ───────────────────────────────────────────────────────────────────── # 2. REPORT DATA QUERY # ───────────────────────────────────────────────────────────────────── async def get_production_report_data(db: AsyncSession) -> dict: """ Build the full production report payload. Returns { columns: [...], feature_columns: [...], rows: [...] } """ today = date.today() # ── Feature columns (dynamic) ─────────────────────────────────── feature_columns = await get_special_feature_columns(db) # ── All chassis that have a confirmed order or are inwarded ───── ReceivingTransport = aliased(TransportMaster) stmt = ( select( ChassisMaster.chassis_id, ChassisMaster.chassis_no, ChassisMaster.light_type, ChassisMaster.da_lite_status, ChassisMaster.arrival_date, ChassisMaster.tracking_status, ChassisMaster.current_status, ChassisMaster.yard_location, ChassisMaster.handed_to_production_at, ChassisMaster.dispatched_at, ChassisMaster.transport_id, ChassisMaster.dealer_order_vehicle_id, ChassisMaster.current_station_id, ChassisMaster.current_location_id, VehicleModelMaster.model_number, StationMaster.station_name.label('current_station_name'), ChassisLocationMaster.location_name.label('location_name'), ReceivingTransport.name.label('receiving_transporter'), ChassisMaster.procurement_status, # Manual entry fields ProductionReportManualEntry.contractor_group, ProductionReportManualEntry.ac_make, ProductionReportManualEntry.interior_work, ProductionReportManualEntry.driver_request, ProductionReportManualEntry.document_pending, ProductionReportManualEntry.seat_fitting_pending, ProductionReportManualEntry.remark, ) .outerjoin(VehicleModelMaster, ChassisMaster.model_id == VehicleModelMaster.model_id) .outerjoin(StationMaster, ChassisMaster.current_station_id == StationMaster.station_id) .outerjoin(ChassisLocationMaster, ChassisMaster.current_location_id == ChassisLocationMaster.location_id) .outerjoin(ReceivingTransport, ChassisMaster.transport_id == ReceivingTransport.transport_id) .outerjoin(ProductionReportManualEntry, ProductionReportManualEntry.chassis_id == ChassisMaster.chassis_id) .where( ChassisMaster.current_status == 'HANDED_TO_PRODUCTION' ) .order_by(ChassisMaster.arrival_date.desc().nullslast(), ChassisMaster.created_at.desc()) ) chassis_rows = (await db.execute(stmt)).all() # ── Match Chassis to DealerOrderVehicle via Chassis No or Direct ID ─ chassis_nos = [r.chassis_no.strip() for r in chassis_rows if r.chassis_no] chassis_to_ov_map: dict[str, uuid.UUID] = {} if chassis_nos: dovc_stmt = ( select(DealerOrderVehicleChassis.chassis_no, DealerOrderVehicleChassis.order_vehicle_id) .where(DealerOrderVehicleChassis.chassis_no.in_(chassis_nos)) ) for c_no, ov_id in (await db.execute(dovc_stmt)).all(): if c_no and ov_id: chassis_to_ov_map[c_no.strip()] = ov_id # Collect all order vehicle IDs needed ov_ids_set = set() for r in chassis_rows: ov_id = r.dealer_order_vehicle_id or (chassis_to_ov_map.get(r.chassis_no.strip()) if r.chassis_no else None) if ov_id: ov_ids_set.add(ov_id) ov_ids = list(ov_ids_set) # Order vehicles + parent order info order_vehicle_map: dict[uuid.UUID, dict] = {} if ov_ids: ov_stmt = ( select( DealerOrderVehicle.order_vehicle_id, DealerOrderVehicle.model_id, DealerOrderVehicle.body_type_id, DealerOrderVehicle.segment_id, DealerOrderVehicle.order_id, DealerOrder.dealer_id, DealerOrder.client_id, BodyTypeMaster.name.label('body_type_name'), CustomerSegmentMaster.name.label('segment_name'), VehicleModelMaster.model_number.label('ov_model_number'), CustomerMaster.name.label('dealer_name'), ClientMaster.name.label('client_name'), ) .join(DealerOrder, DealerOrderVehicle.order_id == DealerOrder.order_id) .outerjoin(BodyTypeMaster, DealerOrderVehicle.body_type_id == BodyTypeMaster.body_type_id) .outerjoin(CustomerSegmentMaster, DealerOrderVehicle.segment_id == CustomerSegmentMaster.segment_id) .outerjoin(VehicleModelMaster, DealerOrderVehicle.model_id == VehicleModelMaster.model_id) .outerjoin(CustomerMaster, DealerOrder.dealer_id == CustomerMaster.customer_id) .outerjoin(ClientMaster, DealerOrder.client_id == ClientMaster.client_id) .where(DealerOrderVehicle.order_vehicle_id.in_(ov_ids)) ) for row in (await db.execute(ov_stmt)).all(): order_vehicle_map[row.order_vehicle_id] = { 'body_type_name': row.body_type_name, 'segment_name': row.segment_name, 'model_number': row.ov_model_number, 'dealer_name': row.dealer_name, 'client_name': row.client_name, 'dealer_id': row.dealer_id, 'client_id': row.client_id, } # ── Batch-load features for all linked order vehicles ─────────── features_map: dict[uuid.UUID, dict[uuid.UUID, dict]] = {} # ov_id -> { feature_id -> { ... } } if ov_ids: feat_stmt = ( select( DealerOrderVehicleFeature.order_vehicle_id, DealerOrderVehicleFeature.feature_id, DealerOrderVehicleFeature.feature_option_id, DealerOrderVehicleFeature.bool_value, DealerOrderVehicleFeature.text_value, SpecialFeatureMaster.feature_name, SpecialFeatureMaster.input_type, SpecialFeatureOptionMaster.option_value, ) .join(SpecialFeatureMaster, DealerOrderVehicleFeature.feature_id == SpecialFeatureMaster.feature_id) .outerjoin(SpecialFeatureOptionMaster, DealerOrderVehicleFeature.feature_option_id == SpecialFeatureOptionMaster.option_id) .where(DealerOrderVehicleFeature.order_vehicle_id.in_(ov_ids)) ) for row in (await db.execute(feat_stmt)).all(): ov_feats = features_map.setdefault(row.order_vehicle_id, {}) ov_feats[row.feature_id] = { 'feature_name': row.feature_name, 'input_type': row.input_type, 'option_id': str(row.feature_option_id) if row.feature_option_id else None, 'option_value': row.option_value, 'bool_value': row.bool_value, 'text_value': row.text_value, } # ── Build rows ────────────────────────────────────────────────── rows = [] for idx, r in enumerate(chassis_rows, start=1): c_no_clean = r.chassis_no.strip() if r.chassis_no else '' ov_id = r.dealer_order_vehicle_id or chassis_to_ov_map.get(c_no_clean) ov_data = order_vehicle_map.get(ov_id, {}) if ov_id else {} ov_features = features_map.get(ov_id, {}) if ov_id else {} # Description: "{model} {AC?} {door} {segment} BUS {layout} SEAT" model_num = ov_data.get('model_number') or r.model_number or '' segment_name = ov_data.get('segment_name', '') body_type_name = ov_data.get('body_type_name', '') # Derive AC, door, layout, supplier, faps from features ac_value = '' door_value = '' seating_layout = '' seat_supplier = '' faps_required = '' for fid, fdata in ov_features.items(): fname_lower = fdata['feature_name'].lower() opt_val = fdata.get('option_value') or '' txt_val = fdata.get('text_value') or '' # AC detection if 'ac' in fname_lower: if opt_val and opt_val.lower() not in ('not required', 'none', 'non-ac', 'no'): ac_value = 'AC' elif fdata.get('bool_value'): ac_value = 'AC' # Door detection if 'door' in fname_lower and ('operation' in fname_lower or 'type' in fname_lower or fname_lower == 'door'): if opt_val and opt_val.lower() not in ('not required', 'none'): door_value = opt_val elif txt_val: door_value = txt_val # Seating Layout / Type detection if 'seating' in fname_lower or 'seat type' in fname_lower: if opt_val and opt_val.lower() not in ('not required', 'none'): seating_layout = opt_val # Seat Supplier detection if 'seat' in fname_lower and 'supplier' in fname_lower: if opt_val: seat_supplier = opt_val elif txt_val: seat_supplier = txt_val # FAPS detection if 'faps' in fname_lower or 'fire' in fname_lower: if fdata.get('bool_value'): faps_required = 'Yes' elif opt_val and opt_val.lower() not in ('not required', 'none', 'no'): faps_required = opt_val elif txt_val and txt_val.lower() not in ('not required', 'none', 'no'): faps_required = txt_val # Build description string door_short = '' if door_value: if 'single' in door_value.lower(): door_short = 'SINGLE DOOR' elif 'double' in door_value.lower(): door_short = 'DOUBLE DOOR' else: door_short = door_value.upper() segment_short = segment_name.upper() if segment_name else '' layout_short = seating_layout.upper() if seating_layout else '' description_parts = [p for p in [ model_num, ac_value, door_short, f'{segment_short} BUS' if segment_short else '', f'{layout_short} SEAT' if layout_short else '', ] if p] description = ' '.join(description_parts) # Tail lamp tail_lamp = '' if r.light_type: if r.light_type == 'SQUARE_LIGHT': tail_lamp = 'SQUARE LIGHT' elif r.light_type == 'ROUND_LIGHT': tail_lamp = 'ROUND LIGHT' else: tail_lamp = r.light_type # Chassis age (days since arrival) chassis_age = None if r.arrival_date: chassis_age = (today - r.arrival_date).days # Induction date induction_date = None if r.handed_to_production_at: induction_date = r.handed_to_production_at.date() if isinstance(r.handed_to_production_at, datetime) else r.handed_to_production_at # Body age (days since induction) body_age = None if induction_date: body_age = (today - induction_date).days # Status status = '' if r.current_station_name: status = r.current_station_name elif r.tracking_status: status_map = { 'CHASSIS_IN': 'YARD', 'WIP': 'WIP', 'READY_FOR_DISPATCH': 'RFD', 'DISPATCHED': 'DISP', } status = status_map.get(r.tracking_status, r.tracking_status) # DA-lite status at FVI is the chassis' current DA-lite status fvi_da_lite_status = r.da_lite_status or '' # Build special features tick map: { feature_id: { option_id: True, ... } or bool } feature_ticks = {} for fc in feature_columns: fid = uuid.UUID(fc['feature_id']) feat_data = ov_features.get(fid) if fc['input_type'] == 'BOOLEAN': feature_ticks[fc['feature_id']] = { 'selected': bool(feat_data and feat_data.get('bool_value')), 'value': None, } elif fc['input_type'] == 'DROPDOWN': selected_option_id = feat_data.get('option_id') if feat_data else None selected_option_val = (feat_data.get('option_value') or '').strip() if feat_data else '' is_excluded = is_excluded_feature_value(selected_option_val) option_ticks = {} for opt in fc['options']: option_ticks[opt['option_id']] = ( opt['option_id'] == selected_option_id and not is_excluded ) feature_ticks[fc['feature_id']] = { 'selected': (selected_option_id is not None and not is_excluded), 'value': None if is_excluded else (feat_data.get('option_value') if feat_data else None), 'option_ticks': option_ticks, } elif fc['input_type'] == 'TEXT': raw_txt = (feat_data.get('text_value') or '').strip() if feat_data else '' is_excluded = is_excluded_feature_value(raw_txt) feature_ticks[fc['feature_id']] = { 'selected': bool(raw_txt and not is_excluded), 'value': None if is_excluded else (raw_txt or None), } row = { 'sr_no': idx, 'chassis_id': str(r.chassis_id), 'chassis_no': r.chassis_no or '', 'description': description, 'ac': ac_value, 'tail_lamp': tail_lamp, 'seat_supplier': seat_supplier, 'faps_required': faps_required, 'body_type': body_type_name, 'seating_layout': seating_layout, 'school_staff': segment_short, 'model': model_num, 'dealer_name': ov_data.get('dealer_name', ''), 'customer_name': ov_data.get('client_name', ''), 'contractor': r.contractor_group or '', 'chassis_da_lite_status': r.da_lite_status or '', 'chassis_receive_date': str(r.arrival_date) if r.arrival_date else '', 'chassis_age': chassis_age, 'status': status, 'procurement_status': r.procurement_status or '', 'ac_make': r.ac_make or '', 'interior_work': r.interior_work or '', 'driver_request': r.driver_request or '', 'document_pending': r.document_pending or '', 'seat_fitting_pending': r.seat_fitting_pending or '', 'da_lite_status_fvi': fvi_da_lite_status, 'induction_date': str(induction_date) if induction_date else '', 'body_age': body_age, 'dispatch_date': str(r.dispatched_at.date()) if r.dispatched_at else '', 'receiving_transporter': r.receiving_transporter or '', 'dispatch_transporter': '', # not recorded in the system; column kept for the report layout 'yard_chassis_location': r.yard_location or (r.location_name if r.location_name else ''), 'remark': r.remark or '', 'pdi_inspector_name': '', # not recorded in the system; column kept for the report layout 'feature_ticks': feature_ticks, } rows.append(row) return { 'feature_columns': feature_columns, 'rows': rows, 'generated_at': datetime.now(timezone.utc).isoformat(), 'total_count': len(rows), } # ───────────────────────────────────────────────────────────────────── # 3. MANUAL ENTRY CRUD # ───────────────────────────────────────────────────────────────────── async def upsert_manual_entry( db: AsyncSession, chassis_id: uuid.UUID, data: dict, ) -> dict: """Create or update manual-entry fields for a chassis.""" existing = (await db.execute( select(ProductionReportManualEntry) .where(ProductionReportManualEntry.chassis_id == chassis_id) )).scalar_one_or_none() allowed_fields = { 'contractor_group', 'ac_make', 'interior_work', 'driver_request', 'document_pending', 'seat_fitting_pending', 'remark', } if existing: for k, v in data.items(): if k in allowed_fields: setattr(existing, k, v) await db.commit() await db.refresh(existing) entry = existing else: filtered = {k: v for k, v in data.items() if k in allowed_fields} entry = ProductionReportManualEntry( id=uuid.uuid4(), chassis_id=chassis_id, **filtered, ) db.add(entry) await db.commit() await db.refresh(entry) return { 'chassis_id': str(entry.chassis_id), 'contractor_group': entry.contractor_group, 'ac_make': entry.ac_make, 'interior_work': entry.interior_work, 'driver_request': entry.driver_request, 'document_pending': entry.document_pending, 'seat_fitting_pending': entry.seat_fitting_pending, 'remark': entry.remark, } # ───────────────────────────────────────────────────────────────────── # 4. EXCEL GENERATION # ───────────────────────────────────────────────────────────────────── def _generate_production_report_excel(report_data: dict) -> io.BytesIO: """ Generate a styled Excel workbook with 2-tier grouped merged headers matching the web dashboard layout and displaying every sub-option clearly. """ wb = openpyxl.Workbook() ws = wb.active ws.title = "Production Report" ws.views.sheetView[0].showGridLines = True feature_columns = report_data['feature_columns'] rows = report_data['rows'] # ── Styles ────────────────────────────────────────────────────── super_font = Font(name='Segoe UI', size=10, bold=True, color='FFFFFF') sub_font = Font(name='Segoe UI', size=9, bold=True, color='FFFFFF') data_font = Font(name='Segoe UI', size=9) tick_font = Font(name='Segoe UI', size=11, bold=True, color='1E8449') thin_border = Border( left=Side(style='thin', color='CBD5E1'), right=Side(style='thin', color='CBD5E1'), top=Side(style='thin', color='CBD5E1'), bottom=Side(style='thin', color='CBD5E1'), ) fill_id = PatternFill(start_color='0F172A', end_color='0F172A', fill_type='solid') # Slate 900 fill_core = PatternFill(start_color='1E293B', end_color='1E293B', fill_type='solid') # Slate 800 fill_features_super = PatternFill(start_color='312E81', end_color='312E81', fill_type='solid') # Indigo 900 fill_features_sub = PatternFill(start_color='3730A3', end_color='3730A3', fill_type='solid') # Indigo 800 fill_seat = PatternFill(start_color='172554', end_color='172554', fill_type='solid') # Blue 950 fill_comm = PatternFill(start_color='1E293B', end_color='1E293B', fill_type='solid') # Slate 800 fill_line = PatternFill(start_color='134E4A', end_color='134E4A', fill_type='solid') # Teal 900 fill_track = PatternFill(start_color='0F172A', end_color='0F172A', fill_type='solid') # Slate 900 fill_notes = PatternFill(start_color='134E4A', end_color='134E4A', fill_type='solid') # Teal 900 even_fill = PatternFill(start_color='FFFFFF', end_color='FFFFFF', fill_type='solid') odd_fill = PatternFill(start_color='F8FAFC', end_color='F8FAFC', fill_type='solid') # ── Group Definitions ─────────────────────────────────────────── groups = [] # 1. Identification groups.append({ 'title': 'IDENTIFICATION', 'super_fill': fill_id, 'sub_fill': fill_id, 'cols': [ {'key': 'sr_no', 'sub_title': 'Sr No.', 'align': 'center', 'width': 7}, {'key': 'chassis_no', 'sub_title': 'Chassis No.', 'align': 'center', 'width': 15, 'bold': True}, ] }) # 2. Core Specs groups.append({ 'title': 'CORE SPECS', 'super_fill': fill_core, 'sub_fill': fill_core, 'cols': [ {'key': 'description', 'sub_title': 'Description', 'align': 'left', 'width': 35}, {'key': 'ac', 'sub_title': 'AC', 'align': 'center', 'width': 8}, {'key': 'tail_lamp', 'sub_title': 'Tail Lamp', 'align': 'center', 'width': 15}, ] }) # 3. Special Features (one 2-tier group per feature) for fc in feature_columns: sub_cols = [] if fc['input_type'] == 'DROPDOWN' and fc.get('options'): for opt in fc['options']: sub_cols.append({ 'feature_id': fc['feature_id'], 'option_id': opt['option_id'], 'sub_title': opt['option_value'], 'align': 'center', 'width': max(12, len(opt['option_value']) + 3), 'is_feature': True, }) elif fc['input_type'] == 'BOOLEAN': sub_cols.append({ 'feature_id': fc['feature_id'], 'option_id': None, 'sub_title': '✓ / ✗', 'align': 'center', 'width': 10, 'is_feature': True, }) else: sub_cols.append({ 'feature_id': fc['feature_id'], 'option_id': None, 'sub_title': 'Value', 'align': 'center', 'width': 14, 'is_feature': True, }) if sub_cols: groups.append({ 'title': fc['feature_name'].upper(), 'super_fill': fill_features_super, 'sub_fill': fill_features_sub, 'cols': sub_cols, }) # 4. Seat & Body Specs groups.append({ 'title': 'SEATING & BODY CONFIGURATION', 'super_fill': fill_seat, 'sub_fill': fill_seat, 'cols': [ {'key': 'seat_supplier', 'sub_title': 'Seat Supplier', 'align': 'center', 'width': 16}, {'key': 'faps_required', 'sub_title': 'FAPS Req.', 'align': 'center', 'width': 14}, {'key': 'body_type', 'sub_title': 'Body Type', 'align': 'center', 'width': 16}, {'key': 'seating_layout', 'sub_title': 'Layout', 'align': 'center', 'width': 14}, {'key': 'school_staff', 'sub_title': 'Segment', 'align': 'center', 'width': 12}, {'key': 'model', 'sub_title': 'Model', 'align': 'center', 'width': 12}, ] }) # 5. Commercial groups.append({ 'title': 'COMMERCIAL', 'super_fill': fill_comm, 'sub_fill': fill_comm, 'cols': [ {'key': 'dealer_name', 'sub_title': 'Dealer Name', 'align': 'left', 'width': 22}, {'key': 'customer_name', 'sub_title': 'Customer Name', 'align': 'left', 'width': 22}, ] }) # 6. Line & Shop Floor groups.append({ 'title': 'LINE & SHOP FLOOR (EDITABLE)', 'super_fill': fill_line, 'sub_fill': fill_line, 'cols': [ {'key': 'contractor', 'sub_title': 'Contractor', 'align': 'center', 'width': 12}, {'key': 'ac_make', 'sub_title': 'AC Make', 'align': 'center', 'width': 14}, {'key': 'interior_work', 'sub_title': 'Interior Work', 'align': 'center', 'width': 16}, {'key': 'driver_request', 'sub_title': 'Driver Request', 'align': 'left', 'width': 22}, {'key': 'document_pending', 'sub_title': 'Doc Pending', 'align': 'left', 'width': 18}, {'key': 'seat_fitting_pending', 'sub_title': 'Seat Fitting', 'align': 'left', 'width': 18}, ] }) # 7. Tracking & Ageing groups.append({ 'title': 'TRACKING, AGEING & LOGISTICS', 'super_fill': fill_track, 'sub_fill': fill_track, 'cols': [ {'key': 'chassis_da_lite_status', 'sub_title': 'DA-Lite (Yard)', 'align': 'center', 'width': 14}, {'key': 'chassis_receive_date', 'sub_title': 'Rec. Date', 'align': 'center', 'width': 14}, {'key': 'chassis_age', 'sub_title': 'Chassis Age', 'align': 'center', 'width': 12}, {'key': 'status', 'sub_title': 'Status', 'align': 'center', 'width': 14}, {'key': 'da_lite_status_fvi', 'sub_title': 'DA-Lite (FVI)', 'align': 'center', 'width': 14}, {'key': 'induction_date', 'sub_title': 'Induction Date', 'align': 'center', 'width': 14}, {'key': 'body_age', 'sub_title': 'Body Age', 'align': 'center', 'width': 12}, {'key': 'dispatch_date', 'sub_title': 'Dispatch Date', 'align': 'center', 'width': 14}, {'key': 'receiving_transporter', 'sub_title': 'Rec. TPT', 'align': 'center', 'width': 14}, {'key': 'yard_chassis_location', 'sub_title': 'Yard Location', 'align': 'left', 'width': 18}, ] }) # 8. Notes & PDI groups.append({ 'title': 'NOTES & PDI', 'super_fill': fill_notes, 'sub_fill': fill_notes, 'cols': [ {'key': 'remark', 'sub_title': 'Remark', 'align': 'left', 'width': 22}, {'key': 'pdi_inspector_name', 'sub_title': 'PDI Inspector', 'align': 'center', 'width': 16}, ] }) # ── Write Headers (Row 1 & Row 2) ─────────────────────────────── flat_cols = [] current_col = 1 for g in groups: start_col = current_col num_cols = len(g['cols']) end_col = start_col + num_cols - 1 # Row 1: Super Header ws.cell(row=1, column=start_col, value=g['title']) if num_cols > 1: ws.merge_cells(start_row=1, start_column=start_col, end_row=1, end_column=end_col) for c_idx in range(start_col, end_col + 1): c = ws.cell(row=1, column=c_idx) c.font = super_font c.fill = g['super_fill'] c.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True) c.border = thin_border # Row 2: Sub Headers for c_offset, col_spec in enumerate(g['cols']): col_num = start_col + c_offset cell_sub = ws.cell(row=2, column=col_num, value=col_spec['sub_title']) cell_sub.font = sub_font cell_sub.fill = g['sub_fill'] cell_sub.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True) cell_sub.border = thin_border # Set Column Width ws.column_dimensions[get_column_letter(col_num)].width = col_spec.get('width', 14) flat_cols.append(col_spec) current_col = end_col + 1 total_cols = current_col - 1 ws.row_dimensions[1].height = 26 ws.row_dimensions[2].height = 28 # ── Write Data Rows (Starting at Row 3) ─────────────────────────── for r_idx, r in enumerate(rows, start=3): ws.row_dimensions[r_idx].height = 22 row_fill = odd_fill if r_idx % 2 == 1 else even_fill for c_idx, col_spec in enumerate(flat_cols, start=1): cell = ws.cell(row=r_idx, column=c_idx) cell.fill = row_fill cell.border = thin_border if col_spec.get('is_feature'): fid = col_spec['feature_id'] oid = col_spec.get('option_id') ft = r.get('feature_ticks', {}).get(fid, {}) val = '' if oid is not None: opt_ticks = ft.get('option_ticks', {}) if opt_ticks.get(oid): val = '✓' else: if ft.get('selected'): val = '✓' if ft.get('value') is None else ft.get('value', '✓') cell.value = val cell.font = tick_font if val == '✓' else data_font cell.alignment = Alignment(horizontal='center', vertical='center') else: raw_val = r.get(col_spec['key'], '') cell.value = str(raw_val) if raw_val is not None else '' cell.font = data_font align_h = col_spec.get('align', 'left') cell.alignment = Alignment(horizontal=align_h, vertical='center') # ── Freeze panes (Row 1 & 2 headers + Col A & B identification) ── ws.freeze_panes = 'C3' # ── Auto-filter applied to sub-header row ─────────────────────── ws.auto_filter.ref = f"A2:{get_column_letter(total_cols)}{len(rows) + 2}" buf = io.BytesIO() wb.save(buf) buf.seek(0) return buf async def generate_production_report_excel(db: AsyncSession) -> io.BytesIO: """Fetch data and generate the Excel file.""" report_data = await get_production_report_data(db) return _generate_production_report_excel(report_data)