""" Migration utility to backfill FileProcessingStep table from existing ProcessingLog entries. This script analyzes historical logs and populates the FileProcessingStep table for files that were processed before the status tracking table was created. """ import logging from typing import Dict, List from sqlalchemy.orm import Session from app.models import FileProcessingStep, ProcessingLog logger = logging.getLogger(__name__) def migrate_logs_to_steps(db: Session, file_id: int, dry_run: bool = False) -> Dict: """ Migrate ProcessingLog entries to FileProcessingStep for a single file. Analyzes all logs for the file and creates/updates FileProcessingStep entries based on the latest status per step found in the logs. Args: db: Database session file_id: ID of the file to migrate dry_run: If True, don't commit changes, just report what would be done Returns: Dictionary with migration results: { "file_id": 123, "steps_created": 5, "steps_updated": 3, "steps_skipped": 2, "errors": [] } """ results = {"file_id": file_id, "steps_created": 0, "steps_updated": 0, "steps_skipped": 0, "errors": []} try: # Get all logs for this file, ordered by timestamp logs = ( db.query(ProcessingLog) .filter(ProcessingLog.file_id == file_id) .order_by(ProcessingLog.timestamp.asc()) .all() ) if not logs: logger.info(f"No logs found for file {file_id}") return results # Parse logs to extract latest status per step step_states = _parse_logs_to_step_states(logs) # Create or update FileProcessingStep entries for step_name, state in step_states.items(): existing_step = ( db.query(FileProcessingStep) .filter(FileProcessingStep.file_id == file_id, FileProcessingStep.step_name == step_name) .first() ) if existing_step: # Update existing step only if it differs if ( existing_step.status != state["status"] or existing_step.started_at != state["started_at"] or existing_step.completed_at != state["completed_at"] ): logger.info( f"Updating step {step_name} for file {file_id}: {existing_step.status} -> {state['status']}" ) existing_step.status = state["status"] existing_step.started_at = state["started_at"] existing_step.completed_at = state["completed_at"] existing_step.error_message = state["error_message"] results["steps_updated"] += 1 else: logger.debug(f"Step {step_name} for file {file_id} already up to date") results["steps_skipped"] += 1 else: # Create new step logger.info(f"Creating step {step_name} for file {file_id} with status {state['status']}") new_step = FileProcessingStep( file_id=file_id, step_name=step_name, status=state["status"], started_at=state["started_at"], completed_at=state["completed_at"], error_message=state["error_message"], ) db.add(new_step) results["steps_created"] += 1 if not dry_run: db.commit() logger.info( f"Migration complete for file {file_id}: " f"{results['steps_created']} created, " f"{results['steps_updated']} updated, " f"{results['steps_skipped']} skipped" ) else: db.rollback() logger.info(f"Dry run complete for file {file_id} (no changes committed)") except Exception as e: logger.error(f"Error migrating file {file_id}: {str(e)}") results["errors"].append(str(e)) db.rollback() return results def _parse_logs_to_step_states(logs: List[ProcessingLog]) -> Dict[str, Dict]: """ Parse logs to determine the final state of each step. Processes logs in chronological order to build a state machine tracking the progression of each step. Args: logs: List of ProcessingLog entries ordered by timestamp Returns: Dictionary mapping step_name to state dict: { "step_name": { "status": "success", "started_at": datetime, "completed_at": datetime, "error_message": None } } """ step_states = {} for log in logs: step_name = log.step_name status = log.status.lower() # Initialize step state if first time seeing this step if step_name not in step_states: step_states[step_name] = { "status": "pending", "started_at": None, "completed_at": None, "error_message": None, } # Update state based on log entry if status == "in_progress": # Step started if step_states[step_name]["started_at"] is None: step_states[step_name]["started_at"] = log.timestamp step_states[step_name]["status"] = "in_progress" elif status == "success": # Step completed successfully if step_states[step_name]["started_at"] is None: # If we never saw in_progress, assume it started around this time step_states[step_name]["started_at"] = log.timestamp step_states[step_name]["status"] = "success" step_states[step_name]["completed_at"] = log.timestamp step_states[step_name]["error_message"] = None elif status == "failure": # Step failed if step_states[step_name]["started_at"] is None: step_states[step_name]["started_at"] = log.timestamp step_states[step_name]["status"] = "failure" step_states[step_name]["completed_at"] = log.timestamp step_states[step_name]["error_message"] = log.message elif status in ["pending", "queued"]: # Only update if step hasn't started yet if step_states[step_name]["status"] == "pending": step_states[step_name]["status"] = "pending" # Note: If we see success/failure after a previous success/failure, # the later one wins (represents retry/reprocessing) return step_states def migrate_all_files(db: Session, batch_size: int = 100, dry_run: bool = False) -> Dict: """ Migrate all files that have logs but no FileProcessingStep entries. Args: db: Database session batch_size: Number of files to process in each batch dry_run: If True, don't commit changes Returns: Dictionary with overall migration statistics: { "total_files": 150, "files_migrated": 145, "files_failed": 5, "total_steps_created": 1200, "total_steps_updated": 350, "errors": [...] } """ summary = { "total_files": 0, "files_migrated": 0, "files_failed": 0, "total_steps_created": 0, "total_steps_updated": 0, "total_steps_skipped": 0, "errors": [], } # Find all files that have logs but no steps files_with_logs = db.query(ProcessingLog.file_id).distinct().all() file_ids_with_logs = {file_id for (file_id,) in files_with_logs if file_id is not None} files_with_steps = db.query(FileProcessingStep.file_id).distinct().all() file_ids_with_steps = {file_id for (file_id,) in files_with_steps} files_to_migrate = list(file_ids_with_logs - file_ids_with_steps) summary["total_files"] = len(files_to_migrate) logger.info(f"Found {len(files_to_migrate)} files to migrate") # Process in batches for i in range(0, len(files_to_migrate), batch_size): batch = files_to_migrate[i : i + batch_size] logger.info( f"Processing batch {i // batch_size + 1}: files {i + 1} to {min(i + batch_size, len(files_to_migrate))}" ) for file_id in batch: result = migrate_logs_to_steps(db, file_id, dry_run=dry_run) if result["errors"]: summary["files_failed"] += 1 summary["errors"].extend(result["errors"]) else: summary["files_migrated"] += 1 summary["total_steps_created"] += result["steps_created"] summary["total_steps_updated"] += result["steps_updated"] summary["total_steps_skipped"] += result["steps_skipped"] logger.info( f"Migration summary: {summary['files_migrated']} files migrated, " f"{summary['files_failed']} failed, " f"{summary['total_steps_created']} steps created, " f"{summary['total_steps_updated']} steps updated" ) return summary def verify_migration(db: Session, file_id: int) -> Dict: """ Verify that migration for a file is correct by comparing logs to steps. Args: db: Database session file_id: ID of the file to verify Returns: Dictionary with verification results: { "file_id": 123, "is_valid": True, "discrepancies": [], "log_steps": ["step1", "step2"], "table_steps": ["step1", "step2"] } """ result = {"file_id": file_id, "is_valid": True, "discrepancies": [], "log_steps": [], "table_steps": []} # Get logs and parse them logs = ( db.query(ProcessingLog).filter(ProcessingLog.file_id == file_id).order_by(ProcessingLog.timestamp.asc()).all() ) if not logs: result["discrepancies"].append("No logs found for file") result["is_valid"] = False return result expected_states = _parse_logs_to_step_states(logs) result["log_steps"] = sorted(expected_states.keys()) # Get actual steps from table actual_steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all() result["table_steps"] = sorted([s.step_name for s in actual_steps]) # Compare actual_states = {s.step_name: s for s in actual_steps} # Check for missing steps for step_name in expected_states: if step_name not in actual_states: result["discrepancies"].append(f"Step '{step_name}' missing from table") result["is_valid"] = False continue # Compare status expected = expected_states[step_name] actual = actual_states[step_name] if expected["status"] != actual.status: result["discrepancies"].append( f"Step '{step_name}' status mismatch: expected '{expected['status']}', got '{actual.status}'" ) result["is_valid"] = False # Check for extra steps in table for step_name in actual_states: if step_name not in expected_states: result["discrepancies"].append(f"Extra step '{step_name}' in table not found in logs") # Not marking as invalid since this might be intentional return result