zkxw/backend/scripts/seed_market_materials.py

#!/usr/bin/env python3
"""市场物料库测试数据（biz_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 material as material_svc
from app.crud import material_stock_batch as batch_svc
from app.models.material import BizMaterial

# (material_code, material_name, category, spec_model, unit, warning_qty, order_num, remark,
#  stock_in_no, stock_in_date, expire_date, catalog_no, batch_no, manufacturer,
#  unit_price, qty)
MARKET_MATERIAL_SEEDS = [
    (
        "MKT-MAT-001",
        "品牌宣传册",
        "xccyw",
        "A4 彩色 20页",
        "册",
        Decimal("50"),
        1,
        "市场宣传用物料",
        "MKT-RK-20260615-001",
        "2026-06-10",
        "2028-06-10",
        "CAT-BRO-A4-20",
        "BN20260601",
        "安徽印务有限公司",
        Decimal("12.50"),
        Decimal("200"),
    ),
    (
        "MKT-MAT-002",
        "细胞存储服务手册",
        "xccyw",
        "A5 精装",
        "本",
        Decimal("30"),
        2,
        "客户咨询配套资料",
        "MKT-RK-20260615-002",
        "2026-06-10",
        "2028-06-10",
        "CAT-MAN-A5",
        "BN20260602",
        "安徽印务有限公司",
        Decimal("18.00"),
        Decimal("100"),
    ),
    (
        "MKT-MAT-003",
        "采集专用管套装",
        "ccyw",
        "50ml×6支/套",
        "套",
        Decimal("20"),
        3,
        "采集业务现场使用",
        "MKT-RK-20260615-003",
        "2026-06-11",
        "2027-06-11",
        "CAT-TUBE-50-6",
        "BN20260603",
        "康宁生物耗材",
        Decimal("85.00"),
        Decimal("50"),
    ),
    (
        "MKT-MAT-004",
        "知情同意书范本",
        "ccyw",
        "A4 单页",
        "份",
        Decimal("100"),
        4,
        "采集业务合规文件",
        "MKT-RK-20260615-004",
        "2026-06-11",
        "2027-12-31",
        "CAT-ICF-A4",
        "BN20260604",
        "法务文印中心",
        Decimal("0.80"),
        Decimal("500"),
    ),
    (
        "MKT-MAT-005",
        "会议展架易拉宝",
        "xccyw",
        "80×200cm 含支架",
        "个",
        Decimal("5"),
        5,
        "展会及会议布置",
        "MKT-RK-20260615-005",
        "2026-06-12",
        "2029-06-12",
        "CAT-ROLL-80200",
        "BN20260605",
        "合肥广告制作",
        Decimal("120.00"),
        Decimal("10"),
    ),
]


def _get_material_by_code(db, code: str, tenant_no: str) -> BizMaterial | None:
    return (
        db.query(BizMaterial)
        .filter(
            BizMaterial.material_code == code,
            BizMaterial.tenant_no == tenant_no,
            BizMaterial.del_flag == 0,
        )
        .first()
    )


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 MARKET_MATERIAL_SEEDS:
            (
                code,
                name,
                category,
                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 = _get_material_by_code(db, code, tenant_no)
            if material:
                skipped_material += 1
            else:
                material_svc.create_material(
                    db,
                    {
                        "materialCode": code,
                        "materialName": name,
                        "category": category,
                        "specModel": spec,
                        "unit": unit,
                        "warningQty": warning_qty,
                        "orderNum": order_num,
                        "remark": remark,
                        "status": "1",
                    },
                    tenant_no=tenant_no,
                    user_id=user_id,
                )
                created_material += 1
                material = _get_material_by_code(db, code, tenant_no)

            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()