zkxw/backend/scripts/seed_qc_materials.py

#!/usr/bin/env python3
"""质检物料库测试数据（biz_qc_material + 入库批次 + 变动记录），已存在编码/入库编号则跳过。"""
from __future__ import annotations

import sys
from decimal import Decimal
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 qc_material as material_svc
from app.crud import qc_material_stock_batch as batch_svc

# (material_code, material_name, spec_model, unit, warning_qty, order_num, remark,
#  stock_in_no, stock_in_date, expire_date, catalog_no, batch_no, manufacturer,
#  unit_price, qty)
QC_MATERIAL_SEEDS = [
    (
        "QC-MAT-001",
        "DMEM培养基",
        "500ml/瓶",
        "瓶",
        Decimal("5"),
        1,
        "细胞培养基础培养基",
        "RK-20260601-001",
        "2026-06-01",
        "2027-06-01",
        "CAT-DMEM-500",
        "BN20260501",
        "Gibco",
        Decimal("128.00"),
        Decimal("20"),
    ),
    (
        "QC-MAT-002",
        "胎牛血清",
        "500ml/瓶",
        "瓶",
        Decimal("3"),
        2,
        "细胞培养添加物",
        "RK-20260601-002",
        "2026-06-01",
        "2027-03-01",
        "CAT-FBS-500",
        "BN20260415",
        "Gibco",
        Decimal("2680.00"),
        Decimal("6"),
    ),
    (
        "QC-MAT-003",
        "青霉素-链霉素双抗",
        "100ml/瓶",
        "瓶",
        Decimal("5"),
        3,
        "细胞培养抗生素",
        "RK-20260601-003",
        "2026-06-02",
        "2026-12-02",
        "CAT-PS-100",
        "BN20260520",
        "HyClone",
        Decimal("186.00"),
        Decimal("10"),
    ),
    (
        "QC-MAT-004",
        "胰酶消化液",
        "100ml/瓶",
        "瓶",
        Decimal("4"),
        4,
        "贴壁细胞消化",
        "RK-20260601-004",
        "2026-06-02",
        "2026-11-02",
        "CAT-TRY-100",
        "BN20260518",
        "Sigma",
        Decimal("95.00"),
        Decimal("8"),
    ),
    (
        "QC-MAT-005",
        "台盼蓝染色液",
        "50ml/瓶",
        "瓶",
        Decimal("2"),
        5,
        "细胞活力检测",
        "RK-20260601-005",
        "2026-06-03",
        "2027-06-03",
        "CAT-TB-50",
        "BN20260525",
        "Bio-Rad",
        Decimal("72.00"),
        Decimal("5"),
    ),
]


def seed(tenant_no: str = "zkxw", user_id: str = "d3000000000000000000000000000001") -> None:
    db = SessionLocal()
    created_material = 0
    skipped_material = 0
    created_batch = 0
    skipped_batch = 0
    try:
        for row in QC_MATERIAL_SEEDS:
            (
                code,
                name,
                spec,
                unit,
                warning_qty,
                order_num,
                remark,
                stock_in_no,
                stock_in_date,
                expire_date,
                catalog_no,
                batch_no,
                manufacturer,
                unit_price,
                qty,
            ) = row
            amount = (unit_price * qty).quantize(Decimal("0.01"))

            material = None
            if material_svc.code_exists(db, code, tenant_no):
                skipped_material += 1
                from app.models.models import BizQcMaterial

                material = (
                    db.query(BizQcMaterial)
                    .filter(
                        BizQcMaterial.material_code == code,
                        BizQcMaterial.tenant_no == tenant_no,
                        BizQcMaterial.del_flag == 0,
                    )
                    .first()
                )
            else:
                material = material_svc.create_material(
                    db,
                    material_code=code,
                    material_name=name,
                    spec_model=spec,
                    unit=unit,
                    stock_qty=Decimal("0"),
                    warning_qty=warning_qty,
                    status="1",
                    tenant_no=tenant_no,
                    order_num=order_num,
                    remark=remark,
                    user_id=user_id,
                )
                created_material += 1

            if not material:
                print(f"跳过批次 {stock_in_no}：物料 {code} 不存在")
                continue

            if batch_svc.stock_in_no_exists(db, stock_in_no, tenant_no):
                skipped_batch += 1
                continue

            batch_svc.stock_in_one(
                db,
                material.id,
                stock_in_no=stock_in_no,
                stock_in_date=stock_in_date,
                expire_date=expire_date,
                catalog_no=catalog_no,
                batch_no=batch_no,
                manufacturer=manufacturer,
                unit_price=unit_price,
                amount=amount,
                qty=qty,
                remark="测试数据初始化",
                tenant_no=tenant_no,
                user_id=user_id,
            )
            created_batch += 1

        print(
            f"质检物料库测试数据：物料新增 {created_material} 条、跳过 {skipped_material} 条；"
            f"入库批次新增 {created_batch} 条、跳过 {skipped_batch} 条。"
        )
    finally:
        db.close()


if __name__ == "__main__":
    seed()