zkxw/backend/scripts/seed_cell_product_specs.py

#!/usr/bin/env python3
"""批量初始化细胞产品规格（biz_cell_product_spec），已存在编码则跳过。"""
from __future__ import annotations

import sys
from datetime import datetime
from pathlib import Path

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

from app.core.database import SessionLocal
from app.crud import pilot_service as ps_svc
from app.models.models import BizCellProductSpec

# 旧版仅细胞数规格，执行时软删除
LEGACY_SPEC_CODES = [
    "CPS-100W",
    "CPS-200W",
    "CPS-500W",
    "CPS-1000W",
    "CPS-2000W",
    "CPS-5000W",
]

# (spec_code, spec_name, order_num, remark) — 规格名称：细胞数/体积
CELL_PRODUCT_SPECS = [
    ("CPS-1.1E6-5ML", "1.1×10⁶/5ml", 1, "低剂量规格"),
    ("CPS-2.2E6-10ML", "2.2×10⁶/10ml", 2, "低剂量规格"),
    ("CPS-5.5E6-20ML", "5.5×10⁶/20ml", 3, "标准规格"),
    ("CPS-1.0E7-50ML", "1.0×10⁷/50ml", 4, "标准规格"),
    ("CPS-2.5E7-50ML", "2.5×10⁷/50ml", 5, "高剂量规格"),
    ("CPS-5.5E7-100ML", "5.5×10⁷/100ml", 6, "高剂量规格"),
]


def cleanup_legacy_specs(db, tenant_no: str = "zkxw") -> int:
    rows = (
        db.query(BizCellProductSpec)
        .filter(
            BizCellProductSpec.spec_code.in_(LEGACY_SPEC_CODES),
            BizCellProductSpec.tenant_no == tenant_no,
            BizCellProductSpec.del_flag == 0,
        )
        .all()
    )
    for row in rows:
        row.del_flag = 1
        row.update_time = datetime.now()
    if rows:
        db.commit()
    return len(rows)


def seed(tenant_no: str = "zkxw", user_id: str = "d3000000000000000000000000000001") -> None:
    db = SessionLocal()
    created = 0
    skipped = 0
    try:
        removed = cleanup_legacy_specs(db, tenant_no)
        if removed:
            print(f"已软删除旧版规格 {removed} 条。")

        for code, name, order_num, remark in CELL_PRODUCT_SPECS:
            if ps_svc.cell_product_spec_code_exists(db, code, tenant_no):
                skipped += 1
                continue
            ps_svc.create_cell_product_spec(
                db,
                spec_code=code,
                spec_name=name,
                tenant_no=tenant_no,
                order_num=order_num,
                remark=remark,
                user_id=user_id,
            )
            created += 1
        print(f"细胞产品规格初始化完成：新增 {created} 条，跳过 {skipped} 条（编码已存在）。")
    finally:
        db.close()


if __name__ == "__main__":
    seed()