zkxw/backend/scripts/reset_process_categories.py

#!/usr/bin/env python3
"""重置工艺工序分类（biz_process_category）：清空指定租户数据后插入 5 条初始数据。

用法（在 backend 目录）：
  source venv/bin/activate
  python scripts/reset_process_categories.py
  python scripts/reset_process_categories.py --tenant-no zkxw

注意：biz_process / biz_process_route_step / biz_production_task_step 等表
通过 category_id 引用本表，重置后上述关联 id 会失效（字段可空，不会阻断脚本）。
"""
from __future__ import annotations

import argparse
import sys
from pathlib import Path

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

from sqlalchemy import text

from app.core.database import SessionLocal, engine
from app.crud import process_category as process_category_svc

ADMIN_ID = "d3000000000000000000000000000001"

# (category_code, category_name, order_num, remark)
PROCESS_CATEGORIES = [
    ("PC-XBYP", "细胞培养", 1, "细胞扩增与培养相关工序"),
    ("PC-YBJS", "样本接收", 2, "样本接收、登记与初检"),
    ("PC-ZJJC", "质检检测", 3, "过程质控与放行检测"),
    ("PC-DCGL", "冻存管理", 4, "冻存、复苏与出入库"),
    ("PC-BZZF", "包装发运", 5, "成品包装与发运"),
]


def reset_process_categories(tenant_no: str = "zkxw", user_id: str = ADMIN_ID) -> None:
    with engine.begin() as conn:
        result = conn.execute(
            text("DELETE FROM `biz_process_category` WHERE `tenant_no` = :tenant_no"),
            {"tenant_no": tenant_no},
        )
        deleted = result.rowcount or 0

    db = SessionLocal()
    created = 0
    try:
        for code, name, order_num, remark in PROCESS_CATEGORIES:
            process_category_svc.create_process_category(
                db,
                category_code=code,
                category_name=name,
                tenant_no=tenant_no,
                order_num=order_num,
                remark=remark,
                user_id=user_id,
            )
            created += 1
        print(
            f"工艺工序分类重置完成（租户 {tenant_no}）："
            f"删除 {deleted} 条，新增 {created} 条。"
        )
        for code, name, _, _ in PROCESS_CATEGORIES:
            print(f"  - {code} / {name}")
    finally:
        db.close()


def main() -> None:
    parser = argparse.ArgumentParser(description="重置 biz_process_category 表数据")
    parser.add_argument("--tenant-no", default="zkxw", help="租户编号，默认 zkxw")
    parser.add_argument("--user-id", default=ADMIN_ID, help="create_by / update_by 用户 id")
    args = parser.parse_args()
    reset_process_categories(tenant_no=args.tenant_no, user_id=args.user_id)


if __name__ == "__main__":
    main()