#!/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()