3. 原料入库批次信息采集与关联

入库批次采集的对象与关联关系

原料入库时，需要把四类信息绑定到同一个入库批次记录上：批次号（追溯主键）、供应商（来源责任方）、质检报告（合规凭证）、入库单（数量与时间凭证）。四者通过入库批次记录聚合，任一缺失都会造成追溯断链。下图展示采集字段与关联方向。

采集表结构与约束

用一张入库批次表承载关联，关键约束决定数据一致性：批次号唯一、质检结论必须合格、入库数量与批次数量一致。以下建表语句可直接在 SQLite 中执行验证。

CREATE TABLE supplier (
    supplier_id   TEXT PRIMARY KEY,
    supplier_name TEXT NOT NULL,
    license_no    TEXT NOT NULL
);

CREATE TABLE inbound_batch (
    batch_no      TEXT PRIMARY KEY,          -- 批次号，全局唯一
    material_id   TEXT NOT NULL,             -- 原料编码
    supplier_id   TEXT NOT NULL,
    qc_report_no  TEXT NOT NULL,
    qc_result     TEXT NOT NULL CHECK (qc_result IN ('合格','不合格')),
    inbound_no    TEXT NOT NULL,             -- 入库单号
    qty           REAL NOT NULL CHECK (qty > 0),
    in_time       TEXT NOT NULL,
    FOREIGN KEY (supplier_id) REFERENCES supplier(supplier_id)
);

CREATE INDEX idx_batch_material ON inbound_batch(material_id);
CREATE INDEX idx_batch_inbound  ON inbound_batch(inbound_no);

关联录入与一致性校验

录入时先校验批次号格式与唯一性，再校验质检结论，最后写入并回查。下面用 Python 标准库 sqlite3 完成一次完整录入与多批次一致性检查。

import sqlite3, re

def parse_batch_no(batch_no):
    # 原料码-供应商码-日期(YYYYMMDD)-序号
    m = re.fullmatch(r"([A-Z]{2}\d{3})-([A-Z]{3})-(\d{8})-(\d{3})", batch_no)
    if not m:
        raise ValueError(f"批次号格式非法: {batch_no}")
    return m.groups()

def insert_batch(conn, rec):
    parse_batch_no(rec["batch_no"])
    if rec["qc_result"] != "合格":
        raise ValueError("质检不合格，禁止入库")
    cur = conn.execute("SELECT 1 FROM inbound_batch WHERE batch_no=?", (rec["batch_no"],))
    if cur.fetchone():
        raise ValueError(f"批次号重复: {rec['batch_no']}")
    conn.execute(
        "INSERT INTO inbound_batch VALUES (?,?,?,?,?,?,?,?)",
        (rec["batch_no"], rec["material_id"], rec["supplier_id"],
         rec["qc_report_no"], rec["qc_result"], rec["inbound_no"],
         rec["qty"], rec["in_time"]))
    conn.commit()

def check_consistency(conn, material_id):
    rows = conn.execute(
        "SELECT batch_no, supplier_id, qty, inbound_no FROM inbound_batch WHERE material_id=?",
        (material_id,)).fetchall()
    batch_nos = [r[0] for r in rows]
    assert len(batch_nos) == len(set(batch_nos)), "存在重复批次号"
    for r in rows:
        assert r[2] > 0, f"{r[0]} 数量异常"
        assert r[3], f"{r[0]} 缺少入库单号"
    return rows

conn = sqlite3.connect(":memory:")
conn.executescript(open("schema.sql").read())
conn.execute("INSERT INTO supplier VALUES ('SUP01','晨光乳业原料部','SC123456')")

insert_batch(conn, {"batch_no":"RM001-SUP-20240510-001","material_id":"RM001",
    "supplier_id":"SUP01","qc_report_no":"QC20240510-01","qc_result":"合格",
    "inbound_no":"IN20240510-01","qty":500.0,"in_time":"2024-05-10 09:20"})
insert_batch(conn, {"batch_no":"RM001-SUP-20240512-002","material_id":"RM001",
    "supplier_id":"SUP01","qc_report_no":"QC20240512-03","qc_result":"合格",
    "inbound_no":"IN20240512-02","qty":300.0,"in_time":"2024-05-12 14:05"})

rows = check_consistency(conn, "RM001")
print("批次数量:", len(rows))
for r in rows:
    print(r)

预期输出为两条记录，批次号分别为 RM001-SUP-20240510-001 与 RM001-SUP-20240512-002，数量 500 与 300，均带入库单号。若把第二条的批次号改成与第一条相同，insert_batch 会抛出“批次号重复”；若把 qc_result 改为“不合格”，则抛出“质检不合格，禁止入库”。