JhHardwareWRS_BackPoint/scripts/migrate_report_allocations.py

from pathlib import Path
import re
import sys

from sqlalchemy import text

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from app.database import engine  # noqa: E402


TABLE_NAME = "production_report_allocations"
IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
REQUIRED_COLUMNS = {
    "id": {"column_type": "bigint", "is_nullable": "NO", "extra_contains": ("auto_increment",)},
    "report_id": {"column_type": "bigint", "is_nullable": "NO"},
    "report_item_id": {"column_type": "bigint", "is_nullable": "YES"},
    "attendance_point_name": {
        "column_type": "varchar(128)",
        "is_nullable": "NO",
        "default": "empty_string",
    },
    "employee_phone": {"column_type": "varchar(20)", "is_nullable": "NO", "default": "empty_string"},
    "allocation_date": {"column_type": "date", "is_nullable": "NO"},
    "day_minutes": {"column_type": "decimal(10,2)", "is_nullable": "NO", "default": "zero"},
    "overtime_minutes": {"column_type": "decimal(10,2)", "is_nullable": "NO", "default": "zero"},
    "night_minutes": {"column_type": "decimal(10,2)", "is_nullable": "NO", "default": "zero"},
    "effective_minutes": {"column_type": "decimal(10,2)", "is_nullable": "NO", "default": "zero"},
    "good_qty": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
    "defect_qty": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
    "scrap_qty": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
    "changeover_count": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
    "reference_wage": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
    "created_at": {
        "column_type": "datetime",
        "is_nullable": "NO",
        "default": "current_timestamp",
    },
    "updated_at": {
        "column_type": "datetime",
        "is_nullable": "NO",
        "default": "current_timestamp",
        "extra_contains": ("on update current_timestamp",),
    },
}
REQUIRED_INDEXES = {
    "idx_report_allocations_report": ("report_id",),
    "idx_report_allocations_item": ("report_item_id",),
    "idx_report_allocations_date": ("allocation_date",),
    "idx_report_allocations_point_date": ("attendance_point_name", "allocation_date"),
    "idx_report_allocations_employee_date": ("employee_phone", "allocation_date"),
}
REQUIRED_FOREIGN_KEYS = {
    "report_id": ("production_reports", "id", "CASCADE"),
    "report_item_id": ("production_report_items", "id", "CASCADE"),
}
IndexMetadata = dict[str, tuple[str, ...] | int]


def quote_identifier(identifier: str) -> str:
    if not IDENTIFIER_RE.fullmatch(identifier):
        raise ValueError(f"unexpected SQL identifier: {identifier}")
    return f"`{identifier}`"


def table_exists(conn) -> bool:
    return bool(
        conn.execute(
            text(
                """
                SELECT COUNT(1)
                FROM information_schema.tables
                WHERE table_schema = DATABASE()
                  AND table_name = :table_name
                """
            ),
            {"table_name": TABLE_NAME},
        ).scalar_one()
    )


def existing_columns(conn) -> dict[str, dict[str, str | None]]:
    rows = conn.execute(
        text(
            """
            SELECT column_name, column_type, is_nullable, column_default, extra
            FROM information_schema.columns
            WHERE table_schema = DATABASE()
              AND table_name = :table_name
            """
        ),
        {"table_name": TABLE_NAME},
    ).all()
    return {
        row[0]: {
            "column_type": row[1],
            "is_nullable": row[2],
            "column_default": row[3],
            "extra": row[4],
        }
        for row in rows
    }


def existing_indexes(conn) -> dict[str, IndexMetadata]:
    rows = conn.execute(
        text(
            """
            SELECT index_name, non_unique, column_name
            FROM information_schema.statistics
            WHERE table_schema = DATABASE()
              AND table_name = :table_name
            ORDER BY index_name, seq_in_index
            """
        ),
        {"table_name": TABLE_NAME},
    ).all()
    indexes: dict[str, dict[str, object]] = {}
    for index_name, non_unique, column_name in rows:
        metadata = indexes.setdefault(index_name, {"columns": [], "non_unique": int(non_unique)})
        metadata["columns"].append(column_name)
    return {
        index_name: {
            "columns": tuple(metadata["columns"]),
            "non_unique": metadata["non_unique"],
        }
        for index_name, metadata in indexes.items()
    }


def existing_foreign_keys(conn) -> dict[str, tuple[str, str, str]]:
    rows = conn.execute(
        text(
            """
            SELECT k.column_name, k.referenced_table_name, k.referenced_column_name, r.delete_rule
            FROM information_schema.key_column_usage k
            JOIN information_schema.referential_constraints r
              ON r.constraint_schema = k.constraint_schema
             AND r.constraint_name = k.constraint_name
            WHERE k.table_schema = DATABASE()
              AND k.table_name = :table_name
              AND k.referenced_table_name IS NOT NULL
            ORDER BY k.constraint_name, k.ordinal_position
            """
        ),
        {"table_name": TABLE_NAME},
    ).all()
    return {row[0]: (row[1], row[2], row[3]) for row in rows}


def find_equivalent_index(indexes: dict[str, IndexMetadata], columns: tuple[str, ...]) -> str | None:
    for index_name, metadata in indexes.items():
        if index_name != "PRIMARY" and metadata["columns"] == columns and metadata["non_unique"] == 1:
            return index_name
    return None


def create_index(conn, index_name: str, columns: tuple[str, ...]) -> None:
    if index_name not in REQUIRED_INDEXES or REQUIRED_INDEXES[index_name] != columns:
        raise ValueError(f"unexpected index definition: {index_name} {columns}")

    indexes = existing_indexes(conn)
    if index_name in indexes:
        return

    equivalent_index = find_equivalent_index(indexes, columns)
    if equivalent_index:
        conn.execute(
            text(
                f"ALTER TABLE {quote_identifier(TABLE_NAME)} "
                f"RENAME INDEX {quote_identifier(equivalent_index)} TO {quote_identifier(index_name)}"
            )
        )
        return

    column_sql = ", ".join(quote_identifier(column) for column in columns)
    conn.execute(
        text(
            f"CREATE INDEX {quote_identifier(index_name)} "
            f"ON {quote_identifier(TABLE_NAME)} ({column_sql})"
        )
    )


def cleanup_duplicate_fk_indexes(conn) -> None:
    indexes = existing_indexes(conn)
    duplicates = [
        index_name
        for index_name, metadata in indexes.items()
        if (
            metadata["non_unique"] == 1
            and (
                (metadata["columns"] == ("report_id",) and index_name != "idx_report_allocations_report")
                or (
                    metadata["columns"] == ("report_item_id",)
                    and index_name != "idx_report_allocations_item"
                )
            )
        )
    ]
    for index_name in duplicates:
        conn.execute(
            text(
                f"ALTER TABLE {quote_identifier(TABLE_NAME)} "
                f"DROP INDEX {quote_identifier(index_name)}"
            )
        )


def ensure_schema(conn) -> None:
    conn.execute(
        text(
            """
            CREATE TABLE IF NOT EXISTS production_report_allocations (
                id BIGINT NOT NULL AUTO_INCREMENT,
                report_id BIGINT NOT NULL,
                report_item_id BIGINT NULL,
                attendance_point_name VARCHAR(128) NOT NULL DEFAULT '',
                employee_phone VARCHAR(20) NOT NULL DEFAULT '',
                allocation_date DATE NOT NULL,
                day_minutes DECIMAL(10, 2) NOT NULL DEFAULT 0,
                overtime_minutes DECIMAL(10, 2) NOT NULL DEFAULT 0,
                night_minutes DECIMAL(10, 2) NOT NULL DEFAULT 0,
                effective_minutes DECIMAL(10, 2) NOT NULL DEFAULT 0,
                good_qty DECIMAL(12, 2) NOT NULL DEFAULT 0,
                defect_qty DECIMAL(12, 2) NOT NULL DEFAULT 0,
                scrap_qty DECIMAL(12, 2) NOT NULL DEFAULT 0,
                changeover_count DECIMAL(12, 2) NOT NULL DEFAULT 0,
                reference_wage DECIMAL(12, 2) NOT NULL DEFAULT 0,
                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                PRIMARY KEY (id),
                KEY idx_report_allocations_report (report_id),
                KEY idx_report_allocations_item (report_item_id),
                KEY idx_report_allocations_date (allocation_date),
                KEY idx_report_allocations_point_date (attendance_point_name, allocation_date),
                KEY idx_report_allocations_employee_date (employee_phone, allocation_date),
                CONSTRAINT fk_report_allocations_report
                    FOREIGN KEY (report_id) REFERENCES production_reports (id)
                    ON DELETE CASCADE,
                CONSTRAINT fk_report_allocations_item
                    FOREIGN KEY (report_item_id) REFERENCES production_report_items (id)
                    ON DELETE CASCADE
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
            """
        )
    )

    for index_name, columns in REQUIRED_INDEXES.items():
        create_index(conn, index_name, columns)
    cleanup_duplicate_fk_indexes(conn)


def normalized_default(value: str | None) -> str | None:
    if value is None:
        return None
    lowered = str(value).strip().lower()
    if lowered in {"current_timestamp()", "current_timestamp"}:
        return "current_timestamp"
    try:
        return "zero" if float(lowered) == 0 else lowered
    except ValueError:
        return lowered


def column_matches(column: dict[str, str | None], expected: dict[str, object]) -> list[str]:
    mismatches: list[str] = []
    actual_type = str(column["column_type"]).lower()
    actual_nullable = str(column["is_nullable"]).upper()
    actual_default = normalized_default(column["column_default"])
    actual_extra = str(column["extra"] or "").lower()

    if actual_type != expected["column_type"]:
        mismatches.append(f"column_type={column['column_type']}")
    if actual_nullable != expected["is_nullable"]:
        mismatches.append(f"is_nullable={column['is_nullable']}")

    expected_default = expected.get("default")
    if expected_default == "empty_string" and actual_default != "":
        mismatches.append(f"column_default={column['column_default']!r}")
    elif expected_default == "zero" and actual_default != "zero":
        mismatches.append(f"column_default={column['column_default']!r}")
    elif expected_default == "current_timestamp" and actual_default != "current_timestamp":
        mismatches.append(f"column_default={column['column_default']!r}")

    for extra_text in expected.get("extra_contains", ()):
        if extra_text not in actual_extra:
            mismatches.append(f"extra={column['extra']!r}")

    return mismatches


def verify_schema(conn) -> None:
    if not table_exists(conn):
        raise RuntimeError(f"{TABLE_NAME} table is missing")

    columns = existing_columns(conn)
    missing_columns = sorted(set(REQUIRED_COLUMNS) - set(columns))
    if missing_columns:
        raise RuntimeError(f"{TABLE_NAME} missing columns: {', '.join(missing_columns)}")

    incompatible_columns = [
        f"{column_name}({', '.join(mismatches)})"
        for column_name, expected in REQUIRED_COLUMNS.items()
        if (mismatches := column_matches(columns[column_name], expected))
    ]
    if incompatible_columns:
        raise RuntimeError(f"{TABLE_NAME} incompatible columns: {', '.join(incompatible_columns)}")

    indexes = existing_indexes(conn)
    missing_indexes = [index_name for index_name in REQUIRED_INDEXES if index_name not in indexes]
    if missing_indexes:
        raise RuntimeError(f"{TABLE_NAME} missing indexes: {', '.join(missing_indexes)}")

    wrong_indexes = [
        f"{index_name}({', '.join(indexes[index_name]['columns'])})"
        for index_name, columns in REQUIRED_INDEXES.items()
        if indexes.get(index_name, {}).get("columns") != columns
    ]
    if wrong_indexes:
        raise RuntimeError(f"{TABLE_NAME} incompatible indexes: {', '.join(wrong_indexes)}")

    unique_required_indexes = [
        index_name
        for index_name in REQUIRED_INDEXES
        if indexes[index_name]["non_unique"] != 1
    ]
    if unique_required_indexes:
        raise RuntimeError(f"{TABLE_NAME} required indexes are unique: {', '.join(unique_required_indexes)}")

    duplicate_single_column_indexes = [
        index_name
        for index_name, metadata in indexes.items()
        if (
            metadata["non_unique"] == 1
            and (
                (metadata["columns"] == ("report_id",) and index_name != "idx_report_allocations_report")
                or (
                    metadata["columns"] == ("report_item_id",)
                    and index_name != "idx_report_allocations_item"
                )
            )
        )
    ]
    if duplicate_single_column_indexes:
        raise RuntimeError(
            f"{TABLE_NAME} duplicate FK indexes: {', '.join(sorted(duplicate_single_column_indexes))}"
        )

    unique_single_column_indexes = [
        index_name
        for index_name, metadata in indexes.items()
        if (
            metadata["non_unique"] != 1
            and (
                metadata["columns"] == ("report_id",)
                or metadata["columns"] == ("report_item_id",)
            )
        )
    ]
    if unique_single_column_indexes:
        raise RuntimeError(
            f"{TABLE_NAME} unique FK indexes are incompatible: {', '.join(sorted(unique_single_column_indexes))}"
        )

    foreign_keys = existing_foreign_keys(conn)
    missing_foreign_keys = [
        column
        for column, definition in REQUIRED_FOREIGN_KEYS.items()
        if foreign_keys.get(column) != definition
    ]
    if missing_foreign_keys:
        raise RuntimeError(f"{TABLE_NAME} missing foreign keys: {', '.join(missing_foreign_keys)}")


def main() -> None:
    with engine.begin() as conn:
        ensure_schema(conn)
        verify_schema(conn)

    print("report allocations schema migrated")


if __name__ == "__main__":
    main()