3. 一物一码生成与打印方案

一物一码生成与打印方案

本方案面向关键零部件（以某型号液压阀体为例），解决三个核心问题：打印内容与数据库记录不一致、标签重复打印、补打时的追溯关联。核心思路是：数据库先生成唯一码并落库，打印动作只消费已落库的码，每次打印/补打都写入打印日志，标签上的码与数据库记录通过 serial_no 强绑定。

1. 编码结构与生成规则

沿用前例的编码规则，本方案在码中嵌入校验位，便于打印后扫码自校验。以液压阀体为例，码长 22 位：

完整码示例：HVL3250714B0001287。校验位计算：对前 17 位字符取 ASCII 码，按位置权重 1~17 加权求和，再对 10 取模。

def calc_check_digit(body: str) -> str:
    # body 为前 17 位
    total = sum((i + 1) * ord(ch) for i, ch in enumerate(body))
    return str(total % 10)

def build_serial(line: str, date: str, shift: str, seq: int) -> str:
    body = f"HV{line}{date}{shift}{seq:06d}"   # 17 位
    return body + calc_check_digit(body)

# 示例
print(build_serial("L3", "250714", "B", 128))
# 输出: HVL3250714B0001287

2. 生成与落库：先入库，后打印

关键约束：任何标签上的码，必须先在数据库中作为一条唯一记录存在。打印程序不允许自行拼码，只能从待打印池中取码。这样从源头杜绝“打印内容与数据库不一致”。

CREATE TABLE serial_master (
    serial_no     VARCHAR(22) PRIMARY KEY,
    part_no       VARCHAR(32) NOT NULL,
    line_code     VARCHAR(4)  NOT NULL,
    prod_date     DATE        NOT NULL,
    shift_code    CHAR(1)     NOT NULL,
    seq_no        INT         NOT NULL,
    status        VARCHAR(12) NOT NULL DEFAULT 'CREATED', -- CREATED/PRINTED/BOUND/SCRAPPED
    created_at    DATETIME    NOT NULL,
    UNIQUE KEY uk_line_date_shift_seq (line_code, prod_date, shift_code, seq_no)
);

CREATE TABLE print_log (
    log_id        BIGINT AUTO_INCREMENT PRIMARY KEY,
    serial_no     VARCHAR(22) NOT NULL,
    print_type    VARCHAR(8)  NOT NULL,   -- FIRST / REPRINT
    reason_code   VARCHAR(16) NULL,       -- 补打原因
    printer_id    VARCHAR(32) NOT NULL,
    operator_id   VARCHAR(32) NOT NULL,
    printed_at    DATETIME    NOT NULL,
    KEY idx_serial (serial_no)
);

生成一批码时，用唯一索引 uk_line_date_shift_seq 保证同一产线、日期、班次下流水号不重复；若并发插入冲突，捕获重复键异常后重试下一个流水号。

def generate_batch(conn, part_no, line, date, shift, count):
    cur = conn.cursor()
    cur.execute(
        "SELECT COALESCE(MAX(seq_no),0) FROM serial_master "
        "WHERE line_code=%s AND prod_date=%s AND shift_code=%s",
        (line, date, shift))
    start = cur.fetchone()[0] + 1
    created = []
    for i in range(count):
        seq = start + i
        sn = build_serial(line, date, shift, seq)
        try:
            cur.execute(
                "INSERT INTO serial_master(serial_no,part_no,line_code,"
                "prod_date,shift_code,seq_no,status,created_at) "
                "VALUES(%s,%s,%s,%s,%s,%s,'CREATED',NOW())",
                (sn, part_no, line, date, shift, seq))
            created.append(sn)
        except IntegrityError:
            continue   # 流水号被占用，跳过
    conn.commit()
    return created

3. 打印与补打：状态机与日志

打印动作把状态从 CREATED 推进到 PRINTED；补打不改变状态，只追加一条 REPRINT 日志。这样“首次打印”与“补打”在数据上可区分，且补打次数可统计。

def print_label(conn, serial_no, printer_id, operator_id,
                print_type="FIRST", reason_code=None):
    cur = conn.cursor()
    cur.execute("SELECT status FROM serial_master WHERE serial_no=%s FOR UPDATE",
                (serial_no,))
    row = cur.fetchone()
    if row is None:
        raise ValueError("码不存在，禁止打印")
    status = row[0]

    if print_type == "FIRST":
        if status != "CREATED":
            raise ValueError(f"非首次打印状态: {status}")
        cur.execute("UPDATE serial_master SET status='PRINTED' "
                    "WHERE serial_no=%s", (serial_no,))
    else:  # REPRINT
        if status not in ("PRINTED", "BOUND"):
            raise ValueError("仅已打印/已绑定的码可补打")
        if not reason_code:
            raise ValueError("补打必须填写原因")

    cur.execute(
        "INSERT INTO print_log(serial_no,print_type,reason_code,"
        "printer_id,operator_id,printed_at) VALUES(%s,%s,%s,%s,%s,NOW())",
        (serial_no, print_type, reason_code, printer_id, operator_id))
    conn.commit()

4. 打印内容与数据库一致性校验

标签模板中的可变字段（码、零件号、日期、班次）全部来自 serial_master 查询结果，模板本身不拼接业务字段。打印前做一次“回读校验”：把即将发送给打印机的数据与数据库记录逐字段比对，不一致则拒绝打印。

def verify_before_print(conn, serial_no, label_data: dict):
    cur = conn.cursor()
    cur.execute("SELECT part_no,line_code,prod_date,shift_code "
                "FROM serial_master WHERE serial_no=%s", (serial_no,))
    db = cur.fetchone()
    if db is None:
        return False, "码未落库"
    expect = {
        "serial_no": serial_no,
        "part_no":   db[0],
        "line_code": db[1],
        "prod_date": str(db[2]),
        "shift_code": db[3],
    }
    for k, v in expect.items():
        if str(label_data.get(k)) != str(v):
            return False, f"字段不一致: {k} 标签={label_data.get(k)} 库={v}"
    # 校验位自检
    if calc_check_digit(serial_no[:17]) != serial_no[17]:
        return False, "校验位错误"
    return True, "OK"

5. 补打与追溯关联

补打产生的多张标签共享同一个 serial_no，因此无论哪张标签被扫描，追溯查询都指向同一条主记录。查询某码的全部打印历史：

SELECT p.print_type, p.reason_code, p.printer_id,
       p.operator_id, p.printed_at
FROM print_log p
WHERE p.serial_no = 'HVL3250714B0001287'
ORDER BY p.printed_at;

若后续该码在装配工位被绑定到产品，绑定记录只写一次 serial_no，不区分标签版本，从而保证“补打不产生新的追溯分支”。

6. 验证与通过条件

• 唯一性：同一产线、日期、班次下 seq_no 不重复，插入冲突被唯一索引拦截。
• 一致性：verify_before_print 对 5 个字段逐一比对，任一不符即拒绝打印。
• 防重复打印：首次打印要求状态为 CREATED，已 PRINTED 的码再次走首次打印会抛异常。
• 补打可追溯：每次补打写入 print_log，含原因、打印机、操作员、时间；同一码补打次数可统计。
• 追溯不分支：所有标签共享同一 serial_no，绑定与查询均以该字段为准。