取号自动填报：13,478 条申请的解析与入库

定位：一份每周都会出现的重复劳动

品类同事在 PLM 系统里提交取号申请后，系统可以导出一份 MD 格式的申请单文件：一行一条申请，@ 符号分隔十几个字段。每周的固定动作是把它整理进一张飞书表格——物料号一行一条、去掉重复、补上申请单链接——供全部门检索和追溯。手工整理一次半个多小时，还总有人名和料号抄错。

这个技能把整个动作压缩成一句「发文件，填报」：把 MD 文件丢给智能体，它解析、去重、写入表格，返回新增行数。本文把这个技能完整开源，包括全部源码。

原始数据：带毛刺的 PLM 导出

原始记录长这样（已脱敏）：

HWA_PartNumberRequest@LBJSQ-VT26090055@杨工@20155343@...@liupengpeng@...@新TOF TMF8829对应的印制板@N1370001190V01（…后面还有若干个 @ 物料号）

三个特征决定了它不能简单按 @ 切分了事：

最近一次运行的数据量：13,478 条申请记录，展开 22,219 行（每条申请的每个物料号一行）。

技能展开：两条硬规则

技能说明（SKILL.md）里除了字段映射，真正保证数据质量的是两条硬规则：

规则一：料号合法性双道校验。第 8 个 @ 之后的每个候选字段，必须同时满足「不含中文/百分号/括号/分号/冒号/逗号」且「不含空格/方括号」才算料号，否则判定为被 @ 切碎的申请原因碎片，跳过。合法字符集只有字母、数字、点和横线。这条规则上线前，混光电流参数里的 "36mA@14.2%" 一类碎片曾把脏数据带进表格。

规则二：申请单号 + 物料号联合去重。写入前先拉取表格全量已有行，同一申请的同一料号只记录一次。所以同一份文件重复提交、或两周文件之间有重叠申请，都不会产生重复行——本周 22,219 行展开里，21,703 行因已存在被跳过，只有 516 行真正入库。

结果：本周一次真实运行

运行记录：解析 13,478 条 → 展开 22,219 行 → 表格已有 21,917 行 → 新增 516 行，跳过 21,703 行。表格现共 22,432 行数据。截图见文首，申请人已脱敏。

技能文件开源

SKILL.md（技能说明全文）：

---
name: md-sheet-fill
label: MD 取号数据填报
description: |
  将 MD 文件中的取号申请数据（HWA_PartNumberRequest 格式）解析后填入飞书电子表格。
  用户提供 MD 文件内容或要求"提取/填入/填报取号申请单"时使用。
  ⚠️ 必须通过 parse_fill_sheet.py 脚本解析和写入，禁止自行拆分字段或手动填表。

# MD 取号数据填报
目标表格
链接：https://hisense.feishu.cn/sheets/FbjVsbn6ehxcsStyfP4ciooRnpw?sheet=7i0rov
Sheet：「取号申请单」（7i0rov）
列（A-F）：物料号 | 申请标题 | 申请原因 | 申请人 | 申请单号 | 链接
使用流程（严格按步骤执行）
Step 1: 获取 MD 内容
用户提供 MD 文件内容（多条 HWA_PartNumberRequest 开头的记录）时，保存为文件或直接传入脚本。无法获取内容时向用户索取。
Step 2: 执行脚本
```bash
python3 scripts/parse_fill_sheet.py --file <md文件路径>
# 或直接传内容：
python3 scripts/parse_fill_sheet.py --content "<MD内容>"
# 只解析不写入（检查用）：
python3 scripts/parse_fill_sheet.py --file <md文件路径> --dry-run
```
Step 3: 反馈结果
严格按脚本输出反馈：
解析信息：解析到 N 条申请记录，展开 M 行（物料号×申请）
去重结果：新增 X 行，跳过（已存在）Y 行——已存在的不重复写入
写入完成：给出表格链接
失败：原样反馈错误信息
解析规则（脚本已实现，智能体不得自行解析）
字段（@ 分隔位置）
对应列
说明
第 1 个 @ 后
E 申请单号
如 LBJSQ-VT26050711
第 3 个 @ 后
D 申请人
如 liumengjiao
第 5 个 @ 后
B 申请标题
如 半成品申请单批量取号
第 6 个 @ 后
F 链接
申请单 ID 前拼接 https://plm.hisense.com/3dspace/common/emxNavigator.jsp?isPopup=true&mode=tree&objectId=
第 7 个 @ 后
C 申请原因
如 原型机55Q61QAVT基础上更换长江64G EMMC
第 8 个 @ 后及以后
A 物料号
每个 @ 一个料号，每个料号一行；去掉 "-" 后的后缀（20323684-012000 → 20323684）
物料号过滤（2026-08-11 修复）：部分记录的申请原因内部含 @ 字符（如 "…电流36mA@14.2%(混合调光…)"），按 @ 拆分后原因会被切成多段混入料号位置。脚本对第 8 个 @ 后的每个字段做两道校验，任一不过即跳过：
含中文/百分号/括号/分号/冒号/逗号等符号 → 原因片段，跳过
含空格/方括号等非料号字符（如 "144HZ HDMI3][Earc 4K"）→ 原因片段，跳过
合法料号字符集：字母/数字/点/横线（如 N9010000052V01、RSAG8.074.58785V09、20121336-012000）。
去重规则：申请单号 + 物料号 组合已存在于表格则跳过（同一申请的同一料号只记录一次）。
禁止事项
❌ 禁止自行拆分 @ 字段或手动填表（必须用脚本）
❌ 禁止修改脚本的字段映射或去重逻辑
❌ 禁止跳过去重检查直接写入

scripts/parse_fill_sheet.py（脚本全文，凭据已替换为占位符）：

#!/usr/bin/env python3
"""
MD 取号申请数据解析 → 飞书电子表格填报脚本
用法:
  python3 parse_fill_sheet.py --file <md文件>          # 从文件读取 MD 内容
  python3 parse_fill_sheet.py --content "<MD内容>"      # 直接传 MD 内容
  python3 parse_fill_sheet.py --file <md文件> --dry-run # 只解析不写入
解析规则：
每条记录以 HWA_PartNumberRequest 开头，到下一条 HWA_PartNumberRequest 截止
字段以 @ 分隔：idx1=申请单号, idx3=申请人, idx5=申请标题, idx6=申请单ID(拼链接),
  idx7=申请原因, idx8 及以后=物料号（去 "-" 后缀）
写入表格：每个物料号一行（物料号|申请标题|申请原因|申请人|申请单号|链接）
去重：申请单号+物料号 已存在于表格则跳过
"""
import argparse
import json
import re
import requests
import sys
SPREADSHEET_TOKEN = "FbjVsbn6ehxcsStyfP4ciooRnpw"
SHEET_ID = "7i0rov"
SHEET_RANGE = f"{SHEET_ID}!A:F"
FEISHU_APP_ID = "cli_xxxxxxxxxxxx"
FEISHU_APP_SECRET = "YOUR_APP_SECRET"
# 明显非料号特征：含中文、百分号、括号、分号、冒号、逗号、句号等符号的字段视为原因片段
NON_MATERIAL_PATTERN = re.compile(r'[\u4e00-\u9fff%（）();；:：。，,、]')
# 合法料号字符集：字母/数字/点/横线（如 N9010000052V01、RSAG8.074.58785V09、20121336-012000）。
# 原因字段内嵌 @ 拆出的 ASCII 片段（含空格/方括号，如 "144HZ HDMI3][Earc 4K"）会被该检查拦下
VALID_MATERIAL_PATTERN = re.compile(r'^[A-Za-z0-9.\-]+$')
PLM_LINK_PREFIX = "https://plm.hisense.com/3dspace/common/emxNavigator.jsp?isPopup=true&mode=tree&objectId="
# 字段列映射（表格列 A-F）
COL_MATERIAL = 0    # A 物料号
COL_TITLE = 1       # B 申请标题
COL_REASON = 2      # C 申请原因
COL_APPLICANT = 3   # D 申请人
COL_APP_NO = 4      # E 申请单号
COL_LINK = 5        # F 链接
def get_tenant_token():
    resp = requests.post(
        "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
        json={"app_id": FEISHU_APP_ID, "app_secret": FEISHU_APP_SECRET},
        timeout=15
    ).json()
    token = resp.get("tenant_access_token")
    if not token:
        print(f"❌ 获取飞书 token 失败: {resp.get('msg')}")
        sys.exit(1)
    return token
def parse_md(content):
    """解析 MD 内容 → 记录列表 [{app_no, applicant, title, reason, link, materials: []}]"""
    records = []
    # 每条记录以 HWA_PartNumberRequest 开头
    parts = content.split("HWA_PartNumberRequest")
    for part in parts:
        part = part.strip()
        if not part or not part.startswith("@"):
            continue
        fields = part.split("@")
        # fields[0] 为空（@ 前是标记本身），fields[1..] 对应字段
        if len(fields) < 9:
            continue
        app_no = fields[1].strip()
        applicant = fields[3].strip()
        title = fields[5].strip()
        app_id = fields[6].strip()
        reason = fields[7].strip()
        link = PLM_LINK_PREFIX + app_id
        # 物料号：fields[8:] 及以后，去 "-" 后缀。
        # ⚠️ 部分记录的原因字段内部含 @ 字符（如 "...电流36mA@14.2%(混合调光...)"），
        # 导致 split('@') 后原因被拆开、料号起始位置前移——只提取符合料号格式的字段，
        # 跳过含中文/百分号/括号等明显非料号的片段。
        materials = []
        for m in fields[8:]:
            m = m.strip()
            if not m:
                continue
            if NON_MATERIAL_PATTERN.search(m):
                continue  # 原因片段（含中文/符号），跳过
            if not VALID_MATERIAL_PATTERN.match(m):
                continue  # 含空格/方括号等非料号字符的片段（如 "144HZ HDMI3][Earc 4K"），跳过
            clean = m.split("-")[0]  # 20121336-012000 → 20121336
            if clean not in materials:
                materials.append(clean)
        if app_no and materials:
            records.append({
                "app_no": app_no,
                "applicant": applicant,
                "title": title,
                "reason": reason,
                "link": link,
                "materials": materials,
            })
    return records
def read_existing(token):
    """读取表格现有数据，返回 (行数, 已存在键集合 {(app_no, material)})"""
    resp = requests.get(
        f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{SPREADSHEET_TOKEN}/values/{SHEET_RANGE}",
        headers={"Authorization": f"Bearer {token}"},
        timeout=30
    ).json()
    if resp.get("code") != 0:
        print(f"❌ 读取表格失败: {resp.get('msg')}")
        sys.exit(1)
    values = resp.get("data", {}).get("valueRange", {}).get("values", [])
    existing = set()
    for row in values:
        if len(row) >= 5:
            material = str(row[COL_MATERIAL]).strip() if row[COL_MATERIAL] else ""
            app_no = str(row[COL_APP_NO]).strip() if row[COL_APP_NO] else ""
            if material and app_no:
                existing.add((app_no, material))
    return len(values), existing
def append_rows(token, rows):
    """追加写入表格（rows: list[list]，每个物料号一行）"""
    if not rows:
        return 0
    # 获取当前最大行（读 A:F 末尾），追加从下一行开始
    resp = requests.get(
        f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{SPREADSHEET_TOKEN}/values/{SHEET_RANGE}",
        headers={"Authorization": f"Bearer {token}"},
        timeout=30
    ).json()
    values = resp.get("data", {}).get("valueRange", {}).get("values", [])
    start_row = len(values) + 1  # 表头占一行，下一行开始追加
    range_str = f"{SHEET_ID}!A{start_row}:F{start_row + len(rows) - 1}"
    body = {
        "valueRange": {
            "range": range_str,
            "values": rows,
        }
    }
    resp = requests.put(
        f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{SPREADSHEET_TOKEN}/values",
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        json=body,
        timeout=30
    ).json()
    if resp.get("code") != 0:
        print(f"❌ 写入表格失败: {resp.get('msg')}")
        sys.exit(1)
    return len(rows)
def main():
    parser = argparse.ArgumentParser(description="MD 取号申请数据解析 → 飞书表格填报")
    parser.add_argument("--file", help="MD 内容文件路径")
    parser.add_argument("--content", help="MD 内容（直接传参）")
    parser.add_argument("--dry-run", action="store_true", help="只解析不写入")
    args = parser.parse_args()
    if args.file:
        with open(args.file, "r", encoding="utf-8") as f:
            content = f.read()
    elif args.content:
        content = args.content
    else:
        print("❌ 必须提供 --file 或 --content")
        sys.exit(1)
    records = parse_md(content)
    if not records:
        print("❌ 未解析到任何记录（MD 内容格式不正确）")
        sys.exit(1)
    print(f"📋 解析到 {len(records)} 条申请记录")
    # 展开为行（每个物料号一行）
    new_rows = []
    for r in records:
        for m in r["materials"]:
            new_rows.append([
                m, r["title"], r["reason"], r["applicant"], r["app_no"], r["link"],
            ])
    print(f"   共展开 {len(new_rows)} 行（物料号×申请）")
    token = get_tenant_token()
    existing_count, existing = read_existing(token)
    print(f"   表格现有 {existing_count} 行（含表头）")
    # 去重：申请单号+物料号 已存在则跳过
    to_insert = []
    skipped = 0
    for row in new_rows:
        key = (row[COL_APP_NO], row[COL_MATERIAL])
        if key in existing:
            skipped += 1
        else:
            to_insert.append(row)
            existing.add(key)
    if args.dry_run:
        print(f"\n🔍 DRY-RUN 结果：将新增 {len(to_insert)} 行，跳过（已存在）{skipped} 行")
        for row in to_insert[:10]:
            print(f"   {row}")
        if len(to_insert) > 10:
            print(f"   ... 共 {len(to_insert)} 行")
        return
    if not to_insert:
        print(f"\n✅ 无需写入：全部 {skipped} 行已存在于表格中")
        return
    inserted = append_rows(token, to_insert)
    print(f"\n✅ 写入完成：新增 {inserted} 行，跳过（已存在）{skipped} 行")
    print(f"   表格链接: https://hisense.feishu.cn/sheets/{SPREADSHEET_TOKEN}?sheet={SHEET_ID}")
if __name__ == "__main__":
    main()

价值与可复用方法

对业务同事，变化是「每周半小时的手工整理 + 抄错风险」变成「发个文件，看一眼新增行数」。对想复刻的人，这个模式的三要素是：位置化解析 + 合法性白名单校验 + 联合主键去重。它不依赖任何高级能力，一个标准库 Python 脚本就够——自动化的价值不在技术复杂度，在于把「必然重复的动作」变成「必然正确的动作」。