332 lines
12 KiB
Python
332 lines
12 KiB
Python
#!/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()
|