"""add_production_tracking_and_qa_extensions Revision ID: a1b2c3d4e5f7 Revises: 6d9170ee6113 Create Date: 2026-09-21 12:50:00.000000 Manual migration for Production Tracking & QA Checksheet modules. Creates: delay_reason_master, chassis_station_log, production_hold, qa_station_template_map, qa_shortage_log Alters: chassis_master (+3 cols), station_master (+1 col), qa_checksheet_template (+1 col), qa_checksheet_instance (+5 cols), qa_checkpoint_response (+4 cols) Seeds: 9 default delay reason codes """ from typing import Sequence, Union from alembic import op import sqlalchemy as sa from sqlalchemy.dialects.postgresql import UUID import uuid # revision identifiers, used by Alembic. revision: str = 'a1b2c3d4e5f7' down_revision: Union[str, None] = '6d9170ee6113' branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None def upgrade() -> None: conn = op.get_bind() inspector = sa.inspect(conn) existing_tables = set(inspector.get_table_names()) # ── NEW TABLE: delay_reason_master ─────────────────────── if 'delay_reason_master' not in existing_tables: op.create_table( 'delay_reason_master', sa.Column('delay_reason_id', sa.UUID(), nullable=False, default=uuid.uuid4), sa.Column('reason_code', sa.String(50), nullable=False), sa.Column('reason_label', sa.String(200), nullable=False), sa.Column('category', sa.String(50), nullable=False, server_default='OPERATIONAL'), sa.Column('is_active', sa.Boolean(), nullable=False, server_default=sa.text('true')), sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()), sa.PrimaryKeyConstraint('delay_reason_id'), sa.UniqueConstraint('reason_code'), ) # ── NEW TABLE: chassis_station_log ─────────────────────── if 'chassis_station_log' not in existing_tables: op.create_table( 'chassis_station_log', sa.Column('log_id', sa.UUID(), nullable=False, default=uuid.uuid4), sa.Column('chassis_id', sa.UUID(), nullable=False), sa.Column('station_id', sa.UUID(), nullable=False), sa.Column('event_type', sa.String(20), nullable=False), sa.Column('scanned_at', sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column('scanned_by', sa.UUID(), nullable=True), sa.Column('scan_mode', sa.String(20), nullable=False, server_default='QR_SCAN'), sa.Column('elapsed_minutes', sa.Integer(), nullable=True), sa.Column('is_overdue', sa.Boolean(), nullable=False, server_default=sa.text('false')), sa.Column('delay_reason_id', sa.UUID(), nullable=True), sa.Column('delay_remarks', sa.Text(), nullable=True), sa.Column('delay_logged_by', sa.UUID(), nullable=True), sa.Column('delay_logged_at', sa.DateTime(timezone=True), nullable=True), sa.Column('status', sa.String(20), nullable=False, server_default='AT_STATION'), sa.Column('remarks', sa.Text(), nullable=True), sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()), sa.PrimaryKeyConstraint('log_id'), sa.ForeignKeyConstraint(['chassis_id'], ['chassis_master.chassis_id']), sa.ForeignKeyConstraint(['station_id'], ['station_master.station_id']), sa.ForeignKeyConstraint(['scanned_by'], ['login_users.user_id']), sa.ForeignKeyConstraint(['delay_reason_id'], ['delay_reason_master.delay_reason_id']), sa.ForeignKeyConstraint(['delay_logged_by'], ['login_users.user_id']), ) # ── NEW TABLE: production_hold ─────────────────────────── if 'production_hold' not in existing_tables: op.create_table( 'production_hold', sa.Column('hold_id', sa.UUID(), nullable=False, default=uuid.uuid4), sa.Column('chassis_id', sa.UUID(), nullable=False), sa.Column('station_id', sa.UUID(), nullable=True), sa.Column('hold_type', sa.String(50), nullable=False), sa.Column('hold_reason', sa.Text(), nullable=False), sa.Column('raised_by', sa.UUID(), nullable=False), sa.Column('raised_at', sa.DateTime(timezone=True), server_default=sa.func.now()), sa.Column('resolved_by', sa.UUID(), nullable=True), sa.Column('resolved_at', sa.DateTime(timezone=True), nullable=True), sa.Column('resolution_notes', sa.Text(), nullable=True), sa.Column('status', sa.String(20), nullable=False, server_default='ACTIVE'), sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()), sa.PrimaryKeyConstraint('hold_id'), sa.ForeignKeyConstraint(['chassis_id'], ['chassis_master.chassis_id']), sa.ForeignKeyConstraint(['station_id'], ['station_master.station_id']), sa.ForeignKeyConstraint(['raised_by'], ['login_users.user_id']), sa.ForeignKeyConstraint(['resolved_by'], ['login_users.user_id']), ) # ── NEW TABLE: qa_station_template_map ─────────────────── if 'qa_station_template_map' not in existing_tables: op.create_table( 'qa_station_template_map', sa.Column('map_id', sa.UUID(), nullable=False, default=uuid.uuid4), sa.Column('station_id', sa.UUID(), nullable=False), sa.Column('template_id', sa.UUID(), nullable=False), sa.Column('sort_order', sa.Integer(), server_default='1'), sa.Column('is_mandatory', sa.Boolean(), nullable=False, server_default=sa.text('true')), sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()), sa.PrimaryKeyConstraint('map_id'), sa.ForeignKeyConstraint(['station_id'], ['station_master.station_id']), sa.ForeignKeyConstraint(['template_id'], ['qa_checksheet_template.template_id']), sa.UniqueConstraint('station_id', 'template_id'), ) # ── NEW TABLE: qa_shortage_log ─────────────────────────── if 'qa_shortage_log' not in existing_tables: op.create_table( 'qa_shortage_log', sa.Column('shortage_id', sa.UUID(), nullable=False, default=uuid.uuid4), sa.Column('chassis_id', sa.UUID(), nullable=False), sa.Column('station_id', sa.UUID(), nullable=False), sa.Column('bom_snapshot_id', sa.UUID(), nullable=True), sa.Column('rm_id', sa.UUID(), nullable=False), sa.Column('required_qty', sa.Numeric(14, 3), nullable=False, server_default='1'), sa.Column('fitted_qty', sa.Numeric(14, 3), nullable=False, server_default='0'), sa.Column('status', sa.String(20), nullable=False, server_default='OPEN'), sa.Column('noted_by', sa.UUID(), nullable=False), sa.Column('noted_at', sa.DateTime(timezone=True), server_default=sa.func.now()), sa.Column('resolved_by', sa.UUID(), nullable=True), sa.Column('resolved_at', sa.DateTime(timezone=True), nullable=True), sa.Column('remarks', sa.Text(), nullable=True), sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()), sa.PrimaryKeyConstraint('shortage_id'), sa.ForeignKeyConstraint(['chassis_id'], ['chassis_master.chassis_id']), sa.ForeignKeyConstraint(['station_id'], ['station_master.station_id']), sa.ForeignKeyConstraint(['bom_snapshot_id'], ['chassis_bom_snapshot.id']), sa.ForeignKeyConstraint(['rm_id'], ['rm_master.rm_id']), sa.ForeignKeyConstraint(['noted_by'], ['login_users.user_id']), sa.ForeignKeyConstraint(['resolved_by'], ['login_users.user_id']), ) # ── ALTER chassis_master: add production tracking columns ─ cm_cols = {c['name'] for c in inspector.get_columns('chassis_master')} if 'production_hold_status' not in cm_cols: op.add_column('chassis_master', sa.Column('production_hold_status', sa.String(20), nullable=False, server_default='NONE')) if 'station_arrived_at' not in cm_cols: op.add_column('chassis_master', sa.Column('station_arrived_at', sa.DateTime(timezone=True), nullable=True)) if 'station_completed_at' not in cm_cols: op.add_column('chassis_master', sa.Column('station_completed_at', sa.DateTime(timezone=True), nullable=True)) # ── ALTER station_master: add alert_threshold_min ───────── sm_cols = {c['name'] for c in inspector.get_columns('station_master')} if 'alert_threshold_min' not in sm_cols: op.add_column('station_master', sa.Column('alert_threshold_min', sa.Integer(), nullable=True)) # ── ALTER qa_checksheet_template: add station_id FK ─────── qt_cols = {c['name'] for c in inspector.get_columns('qa_checksheet_template')} if 'station_id' not in qt_cols: op.add_column('qa_checksheet_template', sa.Column('station_id', sa.UUID(), nullable=True)) op.create_foreign_key('fk_qa_template_station', 'qa_checksheet_template', 'station_master', ['station_id'], ['station_id']) if 'description' not in qt_cols: op.add_column('qa_checksheet_template', sa.Column('description', sa.Text(), nullable=True)) # ── ALTER qa_checksheet_instance: add chassis_id, station_id, station_log_id, etc. ─ qi_cols = {c['name'] for c in inspector.get_columns('qa_checksheet_instance')} if 'chassis_id' not in qi_cols: op.add_column('qa_checksheet_instance', sa.Column('chassis_id', sa.UUID(), nullable=True)) op.create_foreign_key('fk_qa_instance_chassis', 'qa_checksheet_instance', 'chassis_master', ['chassis_id'], ['chassis_id']) if 'station_id' not in qi_cols: op.add_column('qa_checksheet_instance', sa.Column('station_id', sa.UUID(), nullable=True)) op.create_foreign_key('fk_qa_instance_station', 'qa_checksheet_instance', 'station_master', ['station_id'], ['station_id']) if 'station_log_id' not in qi_cols: op.add_column('qa_checksheet_instance', sa.Column('station_log_id', sa.UUID(), nullable=True)) op.create_foreign_key('fk_qa_instance_station_log', 'qa_checksheet_instance', 'chassis_station_log', ['station_log_id'], ['log_id']) if 'completed_at' not in qi_cols: op.add_column('qa_checksheet_instance', sa.Column('completed_at', sa.DateTime(timezone=True), nullable=True)) if 'passed_by' not in qi_cols: op.add_column('qa_checksheet_instance', sa.Column('passed_by', sa.UUID(), nullable=True)) op.create_foreign_key('fk_qa_instance_passed_by', 'qa_checksheet_instance', 'login_users', ['passed_by'], ['user_id']) if 'station_qa_status' not in qi_cols: op.add_column('qa_checksheet_instance', sa.Column('station_qa_status', sa.String(20), nullable=True, server_default='PENDING')) if 'overall_result' not in qi_cols: op.add_column('qa_checksheet_instance', sa.Column('overall_result', sa.String(20), nullable=True, server_default='PENDING')) if 'inspector_id' not in qi_cols: op.add_column('qa_checksheet_instance', sa.Column('inspector_id', sa.UUID(), nullable=True)) op.create_foreign_key('fk_qa_instance_inspector', 'qa_checksheet_instance', 'login_users', ['inspector_id'], ['user_id']) if 'shift' not in qi_cols: op.add_column('qa_checksheet_instance', sa.Column('shift', sa.String(20), nullable=True)) if 'remarks' not in qi_cols: op.add_column('qa_checksheet_instance', sa.Column('remarks', sa.Text(), nullable=True)) if 'rework_notes' not in qi_cols: op.add_column('qa_checksheet_instance', sa.Column('rework_notes', sa.Text(), nullable=True)) # ── ALTER qa_checkpoint_response: add na_reason, is_flagged, flagged_by, flagged_at ─ qr_cols = {c['name'] for c in inspector.get_columns('qa_checkpoint_response')} if 'na_reason' not in qr_cols: op.add_column('qa_checkpoint_response', sa.Column('na_reason', sa.Text(), nullable=True)) if 'is_flagged' not in qr_cols: op.add_column('qa_checkpoint_response', sa.Column('is_flagged', sa.Boolean(), server_default=sa.text('false'))) if 'flagged_by' not in qr_cols: op.add_column('qa_checkpoint_response', sa.Column('flagged_by', sa.UUID(), nullable=True)) op.create_foreign_key('fk_qa_response_flagged_by', 'qa_checkpoint_response', 'login_users', ['flagged_by'], ['user_id']) if 'flagged_at' not in qr_cols: op.add_column('qa_checkpoint_response', sa.Column('flagged_at', sa.DateTime(timezone=True), nullable=True)) if 'rework_status' not in qr_cols: op.add_column('qa_checkpoint_response', sa.Column('rework_status', sa.String(20), nullable=True)) if 'after_rework_observation' not in qr_cols: op.add_column('qa_checkpoint_response', sa.Column('after_rework_observation', sa.Text(), nullable=True)) if 'after_rework_verified_by' not in qr_cols: op.add_column('qa_checkpoint_response', sa.Column('after_rework_verified_by', sa.UUID(), nullable=True)) op.create_foreign_key('fk_qa_response_rework_verifier', 'qa_checkpoint_response', 'login_users', ['after_rework_verified_by'], ['user_id']) # ── SEED: delay reason codes ───────────────────────────── delay_reason_table = sa.table( 'delay_reason_master', sa.column('delay_reason_id', sa.UUID()), sa.column('reason_code', sa.String), sa.column('reason_label', sa.String), sa.column('category', sa.String), ) seeds = [ ('PAYMENT_HOLD', 'Client Payment Hold', 'COMMERCIAL'), ('QA_REWORK', 'QA Rework Required', 'QUALITY'), ('PART_SHORTAGE', 'Part Shortage', 'MATERIAL'), ('MANPOWER', 'Manpower Unavailable', 'OPERATIONAL'), ('MACHINE_DOWN', 'Machine Breakdown', 'OPERATIONAL'), ('DESIGN_CHANGE', 'Design Change Request', 'OPERATIONAL'), ('CLIENT_HOLD', 'Client-Requested Hold', 'COMMERCIAL'), ('WAITING_MATERIAL', 'Waiting for Material', 'MATERIAL'), ('OTHER', 'Other Reason', 'OPERATIONAL'), ] for code, label, cat in seeds: # Use INSERT ... ON CONFLICT DO NOTHING to be idempotent op.execute( sa.text( "INSERT INTO delay_reason_master (delay_reason_id, reason_code, reason_label, category) " "VALUES (gen_random_uuid(), :code, :label, :cat) " "ON CONFLICT (reason_code) DO NOTHING" ).bindparams(code=code, label=label, cat=cat) ) def downgrade() -> None: # Drop new columns from existing tables (reverse order) op.drop_column('qa_checkpoint_response', 'after_rework_verified_by') op.drop_column('qa_checkpoint_response', 'after_rework_observation') op.drop_column('qa_checkpoint_response', 'rework_status') op.drop_column('qa_checkpoint_response', 'flagged_at') op.drop_column('qa_checkpoint_response', 'flagged_by') op.drop_column('qa_checkpoint_response', 'is_flagged') op.drop_column('qa_checkpoint_response', 'na_reason') op.drop_column('qa_checksheet_instance', 'rework_notes') op.drop_column('qa_checksheet_instance', 'remarks') op.drop_column('qa_checksheet_instance', 'shift') op.drop_column('qa_checksheet_instance', 'inspector_id') op.drop_column('qa_checksheet_instance', 'overall_result') op.drop_column('qa_checksheet_instance', 'station_qa_status') op.drop_column('qa_checksheet_instance', 'passed_by') op.drop_column('qa_checksheet_instance', 'completed_at') op.drop_column('qa_checksheet_instance', 'station_log_id') op.drop_column('qa_checksheet_instance', 'station_id') op.drop_column('qa_checksheet_instance', 'chassis_id') op.drop_column('qa_checksheet_template', 'description') op.drop_column('qa_checksheet_template', 'station_id') op.drop_column('station_master', 'alert_threshold_min') op.drop_column('chassis_master', 'station_completed_at') op.drop_column('chassis_master', 'station_arrived_at') op.drop_column('chassis_master', 'production_hold_status') # Drop new tables op.drop_table('qa_shortage_log') op.drop_table('qa_station_template_map') op.drop_table('production_hold') op.drop_table('chassis_station_log') op.drop_table('delay_reason_master')