import asyncio import os import sys import pandas as pd from datetime import datetime import math sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from sqlalchemy.future import select from dotenv import load_dotenv load_dotenv() DATABASE_URL = os.environ.get("DATABASE_URL") engine = create_async_engine(DATABASE_URL) SessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) from app.models.maintenance_models import MaintenanceAsset, MaintenanceRecord DIR_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "Maintainance")) def is_valid_date(val): if pd.isna(val): return False if isinstance(val, datetime): return True if isinstance(val, str) and len(val) >= 8 and '202' in val: return True return False async def get_asset_by_type_and_num(session, m_type, num): stmt = select(MaintenanceAsset).where( MaintenanceAsset.machine_type == m_type, MaintenanceAsset.machine_number == num ) result = await session.execute(stmt) return result.scalars().first() async def process_file(filename, machine_type_enum): path = os.path.join(DIR_PATH, filename) print(f"\n--- Processing {filename} ---") try: xl = pd.ExcelFile(path) except Exception as e: print(f"Error opening file: {e}") return async with SessionLocal() as session: for sheet_name in xl.sheet_names: print(f" Reading sheet {sheet_name}...") # Try to extract machine number from sheet name if possible num_str = ''.join(filter(str.isdigit, sheet_name)) if not num_str: continue machine_num = int(num_str) asset = await get_asset_by_type_and_num(session, machine_type_enum, machine_num) if not asset: print(f" Warning: Asset {machine_type_enum} #{machine_num} not found in DB.") continue try: df = pd.read_excel(path, sheet_name=sheet_name, header=None) except Exception as e: print(f" Error reading sheet: {e}") continue records_to_add = [] current_record = None for index, row in df.iterrows(): vals = [v for v in row.values] # Look for a date in column 1 (index 1) or column 2 (index 2) date_val = None date_col_idx = -1 for i, v in enumerate(vals[:4]): if is_valid_date(v): date_val = v date_col_idx = i break if date_val is not None: # Save previous record if current_record: records_to_add.append(current_record) # Parse new record if isinstance(date_val, str): try: date_val = pd.to_datetime(date_val, dayfirst=True).to_pydatetime() except: pass elif isinstance(date_val, datetime) or isinstance(date_val, pd.Timestamp): try: date_val = date_val.replace(month=date_val.day, day=date_val.month) except ValueError: pass if not isinstance(date_val, datetime): continue # Unparseable date record_type_str = str(vals[date_col_idx+1]) if len(vals) > date_col_idx+1 else "" is_breakdown = "break" in record_type_str.lower() or "bd" in record_type_str.lower() record_type = "BREAKDOWN" if is_breakdown else "PREVENTIVE" problem = str(vals[date_col_idx+1]).strip() if len(vals) > date_col_idx+1 and not pd.isna(vals[date_col_idx+1]) else "" if "preventive" in problem.lower(): problem = "" action = str(vals[date_col_idx+2]).strip() if len(vals) > date_col_idx+2 and not pd.isna(vals[date_col_idx+2]) else "" # Robustly find time_taken and supervisor from the remaining cells remaining = [str(v).strip() for v in vals[date_col_idx+3:] if not pd.isna(v) and str(v).strip() and str(v).lower() != 'nan'] time_taken = "" attended_by = "" for v in reversed(remaining): v_lower = v.lower() if 'min' in v_lower or 'hr' in v_lower or any(c.isdigit() for c in v): if not time_taken: time_taken = v else: if not attended_by and not time_taken: attended_by = v current_record = { "asset_id": asset.id, "maintenance_type": "BD" if record_type == "BREAKDOWN" else "PM", "record_date": date_val.date(), "problem_faced": problem[:1000] if problem else "", "time_taken": time_taken[:50] if time_taken else "", "supervisor_name": attended_by[:150] if attended_by else "", "remarks": "", "actions": [] } if action and action.lower() != 'nan': current_record["actions"].append(action) else: # Continuation row if current_record and len(vals) > 0: text = "" for v in vals: s = str(v).strip() if s and s.lower() != 'nan': text = s break if text and len(text) > 3: current_record["actions"].append(text) # Save the last record if current_record: records_to_add.append(current_record) from app.models.maintenance_models import MaintenanceRecordAction # Insert into DB for r in records_to_add: actions_texts = r.pop("actions") db_rec = MaintenanceRecord(**r) for idx, text in enumerate(actions_texts): db_rec.actions.append(MaintenanceRecordAction(action_taken=text, sr_no=idx+1)) session.add(db_rec) if records_to_add: await session.commit() print(f" Inserted {len(records_to_add)} records for Machine #{machine_num}") async def main(): print("Starting history card ingestion...") files_to_process = [ ("CO2 MIG MACHINE HISTORY CARD 2026.xlsx", "CO2_MIG"), ("BLOWER HISTORY CARD 2026.xlsx", "BLOWER"), ("HOIST HISTORY CARD 2026.xlsx", "HOIST"), ("COMPRESORE HISTORY CARD 2026.xlsx", "COMPRESSOR"), ("OVAN HISTORY CARD 2026.xlsx", "OVAN"), ("SHOWER TESTING 2026.xlsx", "SHOWER"), ] for filename, m_type in files_to_process: await process_file(filename, m_type) print("Ingestion complete.") if __name__ == "__main__": asyncio.run(main())