zkxw/backend/scripts/generate_production_materials_seed_sql.py

#!/usr/bin/env python3
"""从 生产物料.txt 生成期初导入 SQL（biz_production_material*）。"""
from __future__ import annotations

import re
from collections import OrderedDict
from decimal import Decimal, ROUND_HALF_UP
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
TXT_PATH = ROOT / "生产物料.txt"
OUT_PATH = ROOT / "docs/sql/seed_production_materials_init.sql"

TENANT = "zkxw"
USER_ID = "d3000000000000000000000000000001"
TS = "2026-06-18 10:00:00"
STOCK_IN_DATE = "2026-06-18"
EXPIRE_DATE = "2029-06-18"
REMARK = "期初库存导入（生产物料.txt）"

MONTHS = ["2", "3", "4", "5", "6", "7", "8", "9"]

REAGENT_KEYWORDS = (
    "培养基",
    "血清",
    "胰酶",
    "冻存",
    "PBS",
    "DPBS",
    "DMSO",
    "双抗",
    "链霉素",
    "试剂",
    "添加剂",
    "激活剂",
    "活化剂",
    "因子",
    "消毒液",
    "酒精",
    "新洁尔灭",
    "除孢子",
    "鲎试剂",
    "台盼蓝",
    "Ficoll",
    "DEPC",
    "血替",
    "白蛋白",
    "HES",
    "氯化钠",
    "沉降液",
    "MSC",
    "NK",
    "Cryosure",
    "保存液",
    "羟乙基",
    "青霉素",
    "青/链霉素",
    "Rh血型",
    "琼脂",
    "血平皿",
)


def parse_num(raw: str | None) -> Decimal | None:
    if raw is None:
        return None
    s = str(raw).strip()
    if not s:
        return None
    try:
        return Decimal(s)
    except Exception:
        return None


def q4(val: Decimal) -> str:
    return format(val.quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP), "f")


def sql_str(val: str | None) -> str:
    if val is None:
        return "NULL"
    s = str(val).replace("\\", "\\\\").replace("'", "''")
    return f"'{s}'"


def norm_spec(spec: str) -> str | None:
    s = (spec or "").strip()
    if not s or s == "0":
        return None
    return s


def norm_batch(batch: str, row_no: int) -> str:
    s = (batch or "").strip()
    if not s or s in {"0", "/"}:
        return f"INIT-ROW{row_no:04d}"
    return s


def classify_type(name: str) -> str:
    for kw in REAGENT_KEYWORDS:
        if kw in name:
            return "reagent"
    return "consumable"


def calc_balance_and_in_qty(cols: dict[str, str]) -> tuple[Decimal, Decimal]:
    opening = parse_num(cols.get("期初")) or Decimal("0")
    balance: Decimal | None = None
    total_in = opening

    for m in MONTHS:
        v_in = parse_num(cols.get(f"{m}月入库")) or Decimal("0")
        v_out = parse_num(cols.get(f"{m}月出库")) or Decimal("0")
        v_bal = parse_num(cols.get(f"{m}月结存"))

        if v_in == 0 and v_out == 0 and v_bal is None:
            continue

        if balance is None:
            balance = opening

        total_in += v_in
        if v_bal is not None:
            balance = v_bal
        else:
            balance = balance + v_in - v_out

    if balance is None:
        balance = opening

    if total_in <= 0 and balance <= 0:
        in_qty = Decimal("0")
    else:
        in_qty = max(total_in, balance)

    return balance, in_qty


def parse_rows() -> list[dict]:
    text = TXT_PATH.read_text(encoding="utf-8")
    lines = [ln.rstrip("\n") for ln in text.splitlines() if ln.strip()]
    headers = lines[0].split("\t")
    rows: list[dict] = []
    for idx, line in enumerate(lines[1:], start=2):
        parts = line.split("\t")
        if len(parts) < len(headers):
            parts.extend([""] * (len(headers) - len(parts)))
        cols = dict(zip(headers, parts))
        name = (cols.get("名称") or "").strip()
        if not name:
            continue
        remaining, in_qty = calc_balance_and_in_qty(cols)
        rows.append(
            {
                "row_no": idx,
                "name": name,
                "spec": norm_spec(cols.get("规格型号", "")),
                "unit": (cols.get("单位") or "").strip() or "个",
                "batch_no": norm_batch(cols.get("批号", ""), idx),
                "remaining": remaining,
                "in_qty": in_qty,
                "material_type": classify_type(name),
            }
        )
    return rows


def build_materials(rows: list[dict]) -> OrderedDict[tuple, dict]:
    materials: OrderedDict[tuple, dict] = OrderedDict()
    for row in rows:
        key = (row["name"], row["spec"], row["unit"])
        if key not in materials:
            seq = len(materials) + 1
            code = f"ZKXW-PROD-{seq:05d}"
            mat_id = f"pm{seq:032d}"
            materials[key] = {
                "key": key,
                "seq": seq,
                "code": code,
                "mat_id": mat_id,
                "name": row["name"],
                "spec": row["spec"],
                "unit": row["unit"],
                "material_type": row["material_type"],
                "stock_qty": Decimal("0"),
                "batches": [],
            }
        mat = materials[key]
        mat["stock_qty"] += row["remaining"]
        if row["material_type"] == "reagent":
            mat["material_type"] = "reagent"
        batch_seq = len(mat["batches"]) + 1
        batch_id = f"pb{mat['seq']:03d}{batch_seq:029d}"
        stock_in_no = f"INIT-20260618-{mat['seq']:05d}-{batch_seq:02d}"
        mat["batches"].append(
            {
                **row,
                "batch_id": batch_id,
                "stock_in_no": stock_in_no,
                "usage_id": f"pu{mat['seq']:03d}{batch_seq:029d}",
                "material_code": mat["code"],
                "material_type": mat["material_type"],
                "mat_id": mat["mat_id"],
            }
        )
    return materials


def gen_unit_inserts(units: set[str]) -> list[str]:
    # 生产物料单位直接存 unit_name；此处仅确保 biz_item_unit 存在常用单位
    unit_code_map = {
        "个": "ge",
        "件": "jian",
        "套": "tao",
        "台": "tai",
        "支": "zhi",
        "张": "zhang",
        "片": "pian",
        "把": "ba",
        "双": "shuang",
        "盒": "he",
        "包": "bao",
        "袋": "dai",
        "瓶": "ping",
        "桶": "tong",
        "卷": "juan",
        "箱": "xiang",
        "L": "l",
    }
    lines: list[str] = []
    uid = 0x60
    for unit in sorted(units):
        code = unit_code_map.get(unit)
        if not code:
            continue
        uid += 1
        unit_id = f"e400000000000000000000000000{uid:04x}"
        lines.append(
            f"INSERT INTO `biz_item_unit` (`id`, `unit_code`, `unit_name`, `tenant_no`, `order_num`, `remark`, `create_by`, `create_dept_id`, `create_time`, `update_by`, `update_time`, `del_flag`)\n"
            f"SELECT {sql_str(unit_id)}, {sql_str(code)}, {sql_str(unit)}, {sql_str(TENANT)}, 500, '生产物料导入', {sql_str(USER_ID)}, '', {sql_str(TS)}, {sql_str(USER_ID)}, {sql_str(TS)}, 0 FROM DUAL\n"
            f"WHERE NOT EXISTS (SELECT 1 FROM `biz_item_unit` WHERE `unit_code`={sql_str(code)} AND `tenant_no`={sql_str(TENANT)} AND `del_flag`=0);"
        )
    return lines


def gen_material_insert(mat: dict) -> str:
    spec_sql = sql_str(mat["spec"]) if mat["spec"] else "NULL"
    return (
        f"INSERT INTO `biz_production_material` (`id`, `material_type`, `material_code`, `material_name`, `spec_model`, `unit`, `stock_qty`, `warning_qty`, `status`, `tenant_no`, `order_num`, `remark`, `create_by`, `create_dept_id`, `create_time`, `update_by`, `update_time`, `del_flag`)\n"
        f"SELECT {sql_str(mat['mat_id'])}, {sql_str(mat['material_type'])}, {sql_str(mat['code'])}, {sql_str(mat['name'])}, {spec_sql}, {sql_str(mat['unit'])}, {q4(mat['stock_qty'])}, 0.0000, '1', {sql_str(TENANT)}, {mat['seq']}, {sql_str(REMARK)}, {sql_str(USER_ID)}, '', {sql_str(TS)}, {sql_str(USER_ID)}, {sql_str(TS)}, 0 FROM DUAL\n"
        f"WHERE NOT EXISTS (SELECT 1 FROM `biz_production_material` WHERE `material_code`={sql_str(mat['code'])} AND `tenant_no`={sql_str(TENANT)} AND `del_flag`=0);"
    )


def gen_batch_insert(batch: dict) -> str:
    spec_sql = sql_str(batch["spec"]) if batch["spec"] else "NULL"
    return (
        f"INSERT INTO `biz_production_material_stock_batch` (`id`, `material_id`, `material_type`, `material_code`, `material_name`, `spec_model`, `unit`, `stock_in_no`, `stock_in_date`, `expire_date`, `catalog_no`, `batch_no`, `manufacturer`, `unit_price`, `amount`, `in_qty`, `remaining_qty`, `tenant_no`, `order_num`, `remark`, `create_by`, `create_dept_id`, `create_time`, `update_by`, `update_time`, `del_flag`)\n"
        f"SELECT {sql_str(batch['batch_id'])}, (SELECT `id` FROM `biz_production_material` WHERE `material_code`={sql_str(batch['material_code'])} AND `tenant_no`={sql_str(TENANT)} AND `del_flag`=0 LIMIT 1), {sql_str(batch['material_type'])}, {sql_str(batch['material_code'])}, {sql_str(batch['name'])}, {spec_sql}, {sql_str(batch['unit'])}, {sql_str(batch['stock_in_no'])}, {sql_str(STOCK_IN_DATE)}, {sql_str(EXPIRE_DATE)}, {sql_str(batch['material_code'])}, {sql_str(batch['batch_no'])}, '期初导入', 0.00, 0.00, {q4(batch['in_qty'])}, {q4(batch['remaining'])}, {sql_str(TENANT)}, {batch['row_no']}, {sql_str(REMARK)}, {sql_str(USER_ID)}, '', {sql_str(TS)}, {sql_str(USER_ID)}, {sql_str(TS)}, 0 FROM DUAL\n"
        f"WHERE NOT EXISTS (SELECT 1 FROM `biz_production_material_stock_batch` WHERE `stock_in_no`={sql_str(batch['stock_in_no'])} AND `tenant_no`={sql_str(TENANT)} AND `del_flag`=0);"
    )


def gen_usage_insert(batch: dict) -> str:
    if batch["in_qty"] <= 0:
        return ""
    return (
        f"INSERT INTO `biz_production_material_usage` (`id`, `material_id`, `material_type`, `material_code`, `material_name`, `batch_id`, `stock_in_no`, `catalog_no`, `batch_no`, `manufacturer`, `unit_price`, `amount`, `change_type`, `change_qty`, `usage_unit`, `stock_before`, `stock_after`, `source_type`, `tenant_no`, `remark`, `create_by`, `create_dept_id`, `create_time`, `update_by`, `update_time`, `del_flag`)\n"
        f"SELECT {sql_str(batch['usage_id'])}, (SELECT `id` FROM `biz_production_material` WHERE `material_code`={sql_str(batch['material_code'])} AND `tenant_no`={sql_str(TENANT)} AND `del_flag`=0 LIMIT 1), {sql_str(batch['material_type'])}, {sql_str(batch['material_code'])}, {sql_str(batch['name'])}, {sql_str(batch['batch_id'])}, {sql_str(batch['stock_in_no'])}, {sql_str(batch['material_code'])}, {sql_str(batch['batch_no'])}, '期初导入', 0.00, 0.00, 'in', {q4(batch['in_qty'])}, {sql_str(batch['unit'])}, 0.0000, {q4(batch['in_qty'])}, 'manual_in', {sql_str(TENANT)}, {sql_str(REMARK)}, {sql_str(USER_ID)}, '', {sql_str(TS)}, {sql_str(USER_ID)}, {sql_str(TS)}, 0 FROM DUAL\n"
        f"WHERE NOT EXISTS (SELECT 1 FROM `biz_production_material_usage` WHERE `id`={sql_str(batch['usage_id'])});"
    )


def main() -> None:
    rows = parse_rows()
    materials = build_materials(rows)
    units = {m["unit"] for m in materials.values()}

    parts = [
        "-- 生产物料期初数据导入（来源：生产物料.txt）",
        "-- 剩余数量：按台账逐月推算（期初 + 各月入库 - 出库，优先采用各月结存列）",
        "-- 幂等：按 material_code / stock_in_no 跳过已存在记录",
        f"-- 执行：mysql -u root -p zkxw < docs/sql/seed_production_materials_init.sql",
        "",
        "SET NAMES utf8mb4 COLLATE utf8mb4_unicode_ci;",
        "",
        "-- ── 物品单位 biz_item_unit（确保本批物料所需单位存在）──",
        "",
        *gen_unit_inserts(units),
        "",
        "-- ── 物料主表 biz_production_material ──────────────────────",
        "",
    ]

    for mat in materials.values():
        parts.append(gen_material_insert(mat))
        parts.append("")

    parts.append("-- ── 入库批次 biz_production_material_stock_batch ─────────")
    parts.append("")

    all_batches: list[dict] = []
    for mat in materials.values():
        all_batches.extend(mat["batches"])

    for batch in all_batches:
        parts.append(gen_batch_insert(batch))
        parts.append("")

    parts.append("-- ── 变动记录 biz_production_material_usage（期初入库）──────")
    parts.append("")

    for batch in all_batches:
        usage_sql = gen_usage_insert(batch)
        if usage_sql:
            parts.append(usage_sql)
            parts.append("")

    OUT_PATH.write_text("\n".join(parts), encoding="utf-8")
    print(f"Generated {OUT_PATH}")
    print(f"  txt rows: {len(rows)}")
    print(f"  materials: {len(materials)}")
    print(f"  batches: {len(all_batches)}")
    nonzero = sum(1 for r in rows if r["remaining"] > 0)
    print(f"  batches with remaining > 0: {nonzero}")


if __name__ == "__main__":
    main()