169 lines
8.0 KiB
Python
169 lines
8.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
布莱迪压力表 - 订单附件识别批处理脚本
|
||
读取 Excel 中"新参数"字段，调用大模型判断每条订单是否携带附件，
|
||
并对附件类型(资料/配件)做分类，结果写回 Excel。
|
||
|
||
用法:
|
||
pip install openpyxl anthropic
|
||
export ANTHROPIC_API_KEY=sk-...
|
||
python attachment_classifier.py 输入.xlsx 输出.xlsx
|
||
"""
|
||
import os
|
||
import sys
|
||
import json
|
||
import time
|
||
import concurrent.futures as cf
|
||
import openpyxl
|
||
from anthropic import Anthropic
|
||
|
||
# ============ 配置 ============
|
||
MODEL = "claude-sonnet-5" # 分类任务用 Sonnet 足够，性价比高
|
||
MAX_WORKERS = 8 # 并发线程数，按你的 API 速率上限调整
|
||
MAX_RETRY = 3 # 单条失败重试次数
|
||
PARAM_COL_NAME = "新参数" # 参数所在列名(合同工作表)
|
||
CONTRACT_SHEET = None # None=自动取第一个含"新参数"列的表；也可写死表名
|
||
|
||
client = Anthropic() # 自动读取 ANTHROPIC_API_KEY
|
||
|
||
# ============ 提示词 ============
|
||
SYSTEM_PROMPT = """你是布莱迪压力表公司的订单参数审核助手。你的任务是判断一条订单的"技术参数"文本中，除了压力表产品本体之外，是否还要求交付"附件"。
|
||
|
||
【背景】公司生产压力表（用于化工行业）。参数文本里大部分内容是描述产品本体的规格，这些都不是附件。
|
||
|
||
【什么是产品本体（不算附件）】
|
||
表壳直径/材质、精度等级、防护等级(IP)、过压保护、弹性元件材质、接头材质、接头螺纹、安装方式、表盘材质/刻度/颜色、量程、充油、玻璃、指针、调零装置、缓冲钉/阻尼钉、结构代号(如513/531)、卡盘尺寸、以及各种打标/示值/红线标识要求。
|
||
|
||
【附件 = 产品本体之外，随货交付的东西，分两类】
|
||
|
||
一、资料类（交付给客户的技术文件）：
|
||
合格证、检测报告、检验报告、出厂报告、材质证明/材质单/材质报告、说明书(使用/安装/维护)、标定证书/校验证书、制造记录/检验记录、图纸、含技术信息的二维码/标签。
|
||
※ 注意：装箱单、送货单、贴箱单、发货章、包装/箱体标注等，属于物流包装指令，不算资料。
|
||
|
||
二、配件类（随货的独立实物零件）：
|
||
针型阀、二/三阀组、冷凝管/冷凝圈/冷凝弯、表弯管/缓冲弯、虹吸管/存水弯、过压保护器/保护罐、散热器、焊接接头/连接接头、法兰隔膜/隔膜、活接、卡箍、支架、单独提供的垫片、转接件等，凡是随表一起出货的独立零件都算。
|
||
※ 注意：位号牌、铭牌、标牌、号牌属于标识件，不算配件。缓冲钉/阻尼钉是打进接头的工艺，不算配件。
|
||
|
||
【判断步骤】
|
||
1. 逐项扫描文本，剔除所有"产品本体"描述。
|
||
2. 看剩下的内容里是否出现资料类或配件类的东西。
|
||
3. 只要出现任一类，has_attachment 就是"是"；两类都没有就是"否"。
|
||
|
||
【输出要求】
|
||
只输出一个 JSON 对象，不要任何解释、不要 markdown 代码块：
|
||
{"has_attachment":"是或否","types":["资料"或"配件"或两者],"items":["具体识别到的附件名称"],"reason":"一句话依据"}
|
||
- 若 has_attachment 为"否"，types 和 items 都为空数组 []。
|
||
- types 只能包含 "资料" 和/或 "配件"。"""
|
||
|
||
# few-shot：正例 + 陷阱负例，显著提升边界准确率
|
||
FEWSHOT = [
|
||
("表壳直径:100mm 表壳材质:304SS 精度等级:1.6级 接头螺纹:M20×1.5 带缓冲钉 513结构",
|
||
'{"has_attachment":"否","types":[],"items":[],"reason":"全部为产品本体规格，缓冲钉是工艺不算配件"}'),
|
||
("表壳直径:63mm 精度等级:2.5级 带不锈钢位号牌",
|
||
'{"has_attachment":"否","types":[],"items":[],"reason":"位号牌为标识件，不算配件"}'),
|
||
("表壳直径:100mm 张贴好装箱单并注明订单号 送货单盖发货章",
|
||
'{"has_attachment":"否","types":[],"items":[],"reason":"装箱单送货单属物流指令，不算资料"}'),
|
||
("表壳直径:100mm 提供出厂检测报告 合格证随货",
|
||
'{"has_attachment":"是","types":["资料"],"items":["出厂检测报告","合格证"],"reason":"要求交付检测报告和合格证"}'),
|
||
("表壳直径:100mm 针型阀材质:304SS 配316L表弯管",
|
||
'{"has_attachment":"是","types":["配件"],"items":["针型阀","表弯管"],"reason":"随货提供针型阀和表弯管独立零件"}'),
|
||
("出厂检验报告 仪表使用说明书 配冷凝圈 过压保护器",
|
||
'{"has_attachment":"是","types":["资料","配件"],"items":["检验报告","使用说明书","冷凝圈","过压保护器"],"reason":"既有技术文件又有独立零件"}'),
|
||
]
|
||
|
||
|
||
def build_messages(text: str):
|
||
msgs = []
|
||
for u, a in FEWSHOT:
|
||
msgs.append({"role": "user", "content": f'请判断以下订单参数是否携带附件：\n"""\n{u}\n"""'})
|
||
msgs.append({"role": "assistant", "content": a})
|
||
msgs.append({"role": "user", "content": f'请判断以下订单参数是否携带附件：\n"""\n{text}\n"""'})
|
||
return msgs
|
||
|
||
|
||
def parse_json(raw: str) -> dict:
|
||
raw = raw.strip()
|
||
if raw.startswith("```"):
|
||
raw = raw.strip("`")
|
||
if raw.startswith("json"):
|
||
raw = raw[4:]
|
||
s, e = raw.find("{"), raw.rfind("}")
|
||
if s != -1 and e != -1:
|
||
raw = raw[s:e + 1]
|
||
return json.loads(raw)
|
||
|
||
|
||
def classify(text: str) -> dict:
|
||
if not text or not str(text).strip():
|
||
return {"has_attachment": "否", "types": [], "items": [], "reason": "空参数"}
|
||
last_err = None
|
||
for _ in range(MAX_RETRY):
|
||
try:
|
||
resp = client.messages.create(
|
||
model=MODEL,
|
||
max_tokens=400,
|
||
system=SYSTEM_PROMPT,
|
||
messages=build_messages(str(text)),
|
||
)
|
||
return parse_json(resp.content[0].text)
|
||
except Exception as ex: # noqa
|
||
last_err = ex
|
||
time.sleep(1.5)
|
||
return {"has_attachment": "错误", "types": [], "items": [],
|
||
"reason": f"调用失败: {last_err}"}
|
||
|
||
|
||
def find_sheet_and_col(wb):
|
||
for ws in wb.worksheets:
|
||
if CONTRACT_SHEET and ws.title != CONTRACT_SHEET:
|
||
continue
|
||
headers = [c.value for c in ws[1]]
|
||
if PARAM_COL_NAME in headers:
|
||
return ws, headers.index(PARAM_COL_NAME) + 1
|
||
raise RuntimeError(f'未找到含"{PARAM_COL_NAME}"列的工作表')
|
||
|
||
|
||
def main(in_path: str, out_path: str):
|
||
wb = openpyxl.load_workbook(in_path)
|
||
ws, col = find_sheet_and_col(wb)
|
||
max_row = ws.max_row
|
||
print(f"工作表: {ws.title} 参数列: 第{col}列 数据行: {max_row - 1}")
|
||
|
||
# 追加结果列
|
||
base = ws.max_column
|
||
ws.cell(row=1, column=base + 1, value="是否携带附件")
|
||
ws.cell(row=1, column=base + 2, value="附件类型")
|
||
ws.cell(row=1, column=base + 3, value="识别明细")
|
||
ws.cell(row=1, column=base + 4, value="判断依据")
|
||
|
||
rows = list(range(2, max_row + 1))
|
||
results = {}
|
||
|
||
with cf.ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
|
||
fut = {pool.submit(classify, ws.cell(row=r, column=col).value): r for r in rows}
|
||
done = 0
|
||
for f in cf.as_completed(fut):
|
||
r = fut[f]
|
||
results[r] = f.result()
|
||
done += 1
|
||
if done % 50 == 0:
|
||
print(f" 进度 {done}/{len(rows)}")
|
||
|
||
for r in rows:
|
||
d = results[r]
|
||
ws.cell(row=r, column=base + 1, value=d.get("has_attachment", ""))
|
||
ws.cell(row=r, column=base + 2, value="、".join(d.get("types", [])))
|
||
ws.cell(row=r, column=base + 3, value="、".join(d.get("items", [])))
|
||
ws.cell(row=r, column=base + 4, value=d.get("reason", ""))
|
||
|
||
wb.save(out_path)
|
||
n_yes = sum(1 for d in results.values() if d.get("has_attachment") == "是")
|
||
print(f"完成 -> {out_path} 携带附件 {n_yes}/{len(rows)}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) < 3:
|
||
print("用法: python attachment_classifier.py 输入.xlsx 输出.xlsx")
|
||
sys.exit(1)
|
||
main(sys.argv[1], sys.argv[2])
|