zkxw/backend/scripts/seed_production_materials.py

#!/usr/bin/env python3
"""批量初始化生产物料示例数据（物料主数据 + 入库批次），已存在编码则跳过。"""
from __future__ import annotations

import sys
from datetime import date, timedelta
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 production_material as mat_svc
from app.crud import production_material_stock_batch as batch_svc


# (material_type, material_code, material_name, spec_model, unit, warning_qty, remark)
SAMPLE_MATERIALS = [
    (
        "reagent",
        "R001",
        "75% 乙醇溶液",
        "500mL/瓶",
        "瓶",
        10.0,
        "实验室常用消毒剂",
    ),
    (
        "reagent",
        "R002",
        "PBS 缓冲液",
        "1×, pH 7.4",
        "瓶",
        5.0,
        "细胞培养用磷酸盐缓冲液",
    ),
    (
        "reagent",
        "R003",
        "DMEM 高糖培养基",
        "500mL/瓶, 含双抗",
        "瓶",
        8.0,
        "细胞培养基础培养基",
    ),
    (
        "consumable",
        "C001",
        "15mL 离心管",
        "50支/袋",
        "袋",
        20.0,
        "实验室常规耗材",
    ),
    (
        "consumable",
        "C002",
        "T25 细胞培养瓶",
        "1个/包",
        "包",
        15.0,
        "贴壁细胞培养用",
    ),
]

# 入库批次示例数据（与 SAMPLE_MATERIALS 一一对应）
TODAY = date.today()
BATCH_ITEMS = [
    # R001: 乙醇，入库 20 瓶
    dict(
        material_id=None,  # 创建后填充
        stockInNo="RK2025061501",
        stockInDate=TODAY.isoformat(),
        expireDate=(TODAY + timedelta(days=730)).isoformat(),  # 2年后过期
        catalogNo="SP-ETH-500",
        batchNo="B20250601",
        manufacturer="国药集团化学试剂有限公司",
        storageLocation="试剂柜-A-01",
        unitPrice=Decimal("15.00"),
        amount=Decimal("300.00"),
        qty=Decimal("20"),
    ),
    # R002: PBS，入库 10 瓶
    dict(
        material_id=None,
        stockInNo="RK2025061502",
        stockInDate=TODAY.isoformat(),
        expireDate=(TODAY + timedelta(days=365)).isoformat(),
        catalogNo="SP-PBS-500",
        batchNo="B20250602",
        manufacturer="Gibco",
        storageLocation="试剂柜-B-02",
        unitPrice=Decimal("85.00"),
        amount=Decimal("850.00"),
        qty=Decimal("10"),
    ),
    # R003: DMEM，入库 8 瓶
    dict(
        material_id=None,
        stockInNo="RK2025061503",
        stockInDate=TODAY.isoformat(),
        expireDate=(TODAY + timedelta(days=180)).isoformat(),
        catalogNo="SP-DMEM-500",
        batchNo="B20250603",
        manufacturer="Gibco",
        storageLocation="试剂柜-B-03",
        unitPrice=Decimal("120.00"),
        amount=Decimal("960.00"),
        qty=Decimal("8"),
    ),
    # C001: 离心管，入库 50 袋
    dict(
        material_id=None,
        stockInNo="RK2025061504",
        stockInDate=TODAY.isoformat(),
        expireDate=(TODAY + timedelta(days=1825)).isoformat(),  # 5年后过期
        catalogNo="SC-CF15-50",
        batchNo="B20250604",
        manufacturer="Axygen",
        storageLocation="耗材柜-C-01",
        unitPrice=Decimal("45.00"),
        amount=Decimal("2250.00"),
        qty=Decimal("50"),
    ),
    # C002: 培养瓶，入库 30 包
    dict(
        material_id=None,
        stockInNo="RK2025061505",
        stockInDate=TODAY.isoformat(),
        expireDate=(TODAY + timedelta(days=1095)).isoformat(),  # 3年后过期
        catalogNo="SC-CF25-1",
        batchNo="B20250605",
        manufacturer="Corning",
        storageLocation="耗材柜-D-02",
        unitPrice=Decimal("28.00"),
        amount=Decimal("840.00"),
        qty=Decimal("30"),
    ),
]


def seed(
    tenant_no: str = "zkxw",
    user_id: str = "d3000000000000000000000000000001",
) -> None:
    db = SessionLocal()
    mat_created = 0
    mat_skipped = 0
    batch_created = 0
    batch_errors = 0
    try:
        for i, (mt, code, name, spec, unit, warn_qty, remark) in enumerate(SAMPLE_MATERIALS):
            exists = mat_svc.code_exists(db, code, mt, tenant_no)
            if exists:
                # 找到已有物料的 ID
                rows, _ = mat_svc.list_materials(
                    db, page=1, page_size=1, keyword=code, material_type=mt, tenant_no=tenant_no
                )
                mat_id = rows[0]["id"] if rows else None
                print(f"  跳过物料：{code} - {name}（已存在，ID: {mat_id}）")
                mat_skipped += 1
                if mat_id:
                    BATCH_ITEMS[i]["material_id"] = mat_id
            else:
                try:
                    obj = mat_svc.create_material(
                        db,
                        material_type=mt,
                        material_code=code,
                        material_name=name,
                        spec_model=spec,
                        unit=unit,
                        warning_qty=Decimal(str(warn_qty)),
                        status="1",
                        remark=remark,
                        tenant_no=tenant_no,
                        user_id=user_id,
                    )
                    BATCH_ITEMS[i]["material_id"] = obj.id
                    print(f"  新增物料：{code} - {name}（ID: {obj.id}）")
                    mat_created += 1
                except Exception as e:
                    print(f"  错误（物料）：{code} - {name}，原因：{e}")
                    continue

        # 入库批次
        print("\n--- 开始入库批次 ---")
        for item in BATCH_ITEMS:
            mat_id = item.get("material_id")
            if not mat_id:
                print(f"  跳过批次：物料ID不存在")
                continue
            stock_in_no = item["stockInNo"]
            try:
                batch_svc.batch_stock_in(
                    db,
                    items=[item],
                    tenant_no=tenant_no,
                    user_id=user_id,
                )
                print(f"  新增批次：{stock_in_no}（物料ID: {mat_id}，数量: {item['qty']}）")
                batch_created += 1
            except Exception as e:
                print(f"  错误（批次）：{stock_in_no}，原因：{e}")
                batch_errors += 1

        print(
            f"\n完成：物料新增 {mat_created} 条、跳过 {mat_skipped} 条；"
            f"批次新增 {batch_created} 条、错误 {batch_errors} 条。"
        )
    finally:
        db.close()


if __name__ == "__main__":
    seed()