JhHardwareWRS_BackPoint/scripts/migrate_numeric_process_names.py

from pathlib import Path
import re
import sys

from sqlalchemy import select

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

from app.database import SessionLocal  # noqa: E402
from app.models import DeviceQRCode, MoldLockFeedback, Product, ProductionReportItem, WorkSessionDevice  # noqa: E402


LEGACY_PROCESS_PATTERN = re.compile(r"0*(\d+)序")


def normalized(value: str | None) -> str | None:
    text = str(value or "").strip()
    matched = LEGACY_PROCESS_PATTERN.fullmatch(text)
    if matched:
        return str(int(matched.group(1)))
    return None


def normalize_products(db, skipped: list[str]) -> int:
    updated = 0
    products = db.scalars(
        select(Product)
        .where(Product.device_no == "")
        .order_by(Product.attendance_point_name, Product.project_no, Product.product_name, Product.process_name)
    ).all()
    for product in products:
        next_process_name = normalized(product.process_name)
        if next_process_name is None or next_process_name == product.process_name:
            continue
        next_key = {
            "attendance_point_name": product.attendance_point_name,
            "project_no": product.project_no,
            "product_name": product.product_name,
            "device_no": product.device_no,
            "process_name": next_process_name,
        }
        if db.get(Product, next_key) is not None:
            skipped.append(
                f"products: {product.attendance_point_name} / {product.project_no} / "
                f"{product.product_name} / {product.process_name}"
            )
            continue
        qr_conflict = db.get(DeviceQRCode, {
            "attendance_point_name": product.attendance_point_name,
            "device_no": product.product_name,
            "process_name": next_process_name,
        })
        if qr_conflict is not None:
            skipped.append(
                f"products: {product.attendance_point_name} / {product.project_no} / "
                f"{product.product_name} / {product.process_name} (二维码冲突)"
            )
            continue

        old_process_name = product.process_name
        point_name = product.attendance_point_name
        product_name = product.product_name

        for qr in db.scalars(select(DeviceQRCode).where(
            DeviceQRCode.attendance_point_name == point_name,
            DeviceQRCode.device_no == product_name,
            DeviceQRCode.process_name == old_process_name,
        )).all():
            qr.process_name = next_process_name

        for device in db.scalars(select(WorkSessionDevice).where(
            WorkSessionDevice.attendance_point_name == point_name,
            WorkSessionDevice.device_no == product_name,
            WorkSessionDevice.process_name == old_process_name,
        )).all():
            device.process_name = next_process_name

        for item in db.scalars(select(ProductionReportItem).where(
            ProductionReportItem.attendance_point_name == point_name,
            ProductionReportItem.product_name == product_name,
            ProductionReportItem.process_name == old_process_name,
        )).all():
            item.process_name = next_process_name

        for feedback in db.scalars(select(MoldLockFeedback).where(
            MoldLockFeedback.attendance_point_name == point_name,
            MoldLockFeedback.mold_name == product_name,
            MoldLockFeedback.process_name == old_process_name,
        )).all():
            feedback.process_name = next_process_name

        product.process_name = next_process_name
        updated += 1
    return updated


def normalize_remaining_references(db, skipped: list[str]) -> dict[str, int]:
    counts = {
        "device_qrcodes": 0,
        "work_session_devices": 0,
        "production_report_items": 0,
        "mold_lock_feedbacks": 0,
    }
    for qr in db.scalars(select(DeviceQRCode)).all():
        next_process_name = normalized(qr.process_name)
        if next_process_name is None or next_process_name == qr.process_name:
            continue
        next_key = {
            "attendance_point_name": qr.attendance_point_name,
            "device_no": qr.device_no,
            "process_name": next_process_name,
        }
        if db.get(DeviceQRCode, next_key) is not None:
            skipped.append(f"device_qrcodes: {qr.attendance_point_name} / {qr.device_no} / {qr.process_name}")
            continue
        qr.process_name = next_process_name
        counts["device_qrcodes"] += 1

    for device in db.scalars(select(WorkSessionDevice)).all():
        next_process_name = normalized(device.process_name)
        if next_process_name is None or next_process_name == device.process_name:
            continue
        device.process_name = next_process_name
        counts["work_session_devices"] += 1

    for item in db.scalars(select(ProductionReportItem)).all():
        next_process_name = normalized(item.process_name)
        if next_process_name is None or next_process_name == item.process_name:
            continue
        item.process_name = next_process_name
        counts["production_report_items"] += 1

    for feedback in db.scalars(select(MoldLockFeedback)).all():
        next_process_name = normalized(feedback.process_name)
        if next_process_name is None or next_process_name == feedback.process_name:
            continue
        feedback.process_name = next_process_name
        counts["mold_lock_feedbacks"] += 1

    return counts


def main() -> None:
    skipped: list[str] = []
    with SessionLocal() as db:
        product_updates = normalize_products(db, skipped)
        reference_updates = normalize_remaining_references(db, skipped)
        db.commit()

    reference_total = sum(reference_updates.values())
    print(
        "numeric process names migrated, "
        f"products={product_updates}, references={reference_total}, skipped={len(skipped)}"
    )
    for name, count in reference_updates.items():
        print(f"{name}={count}")
    for item in skipped[:20]:
        print(f"skipped conflict: {item}")
    if len(skipped) > 20:
        print(f"... and {len(skipped) - 20} more")


if __name__ == "__main__":
    main()