180 lines
6.8 KiB
Python
180 lines
6.8 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
kocr-summary-review-4b.py — KOCR 输出摘要质量复核（2026-09-06 P2）
|
||
=================================================================
|
||
⚠️ 独立复核工具：不接入 KOCR 自动管线（管线 v9.4 冻结，禁止覆盖 OCR 结果）。
|
||
用法：读已生成的合并凭证.xls → 本地 4B 逐条判断摘要质量 → 输出可疑清单（纯文本供人工复核）。
|
||
|
||
用法:
|
||
python3 kocr-summary-review-4b.py <合并凭证.xls> # 复核全部摘要（2B 本地免费优先）
|
||
python3 kocr-summary-review-4b.py <file.xls> --limit 30 # 只看前 30 条
|
||
python3 kocr-summary-review-4b.py <file.xls> --cloud # 强制 agnes 云复核（质量敏感）
|
||
|
||
判断维度（4B 输出 reasons）:
|
||
- placeholder: 占位符（"会计凭证"/公司名/元数据）
|
||
- truncated: 疑似截断（过短/不完整）
|
||
- suspicious: 与科目明显矛盾 / 含乱码 / 异常字符
|
||
- ok: 真实业务摘要
|
||
|
||
原则：只标记不修改。牧尘铁律——"优先OCR识别为准，不添加后处理逻辑覆盖OCR结果"。
|
||
"""
|
||
import json, os, sys, urllib.request
|
||
|
||
def _gateway_key():
|
||
"""网关 key 从 config.yaml providers.newapi-local 读（不写死在脚本里）"""
|
||
try:
|
||
import yaml
|
||
cfg = yaml.safe_load(open(os.path.expanduser("~/.hermes/config.yaml")))
|
||
return (cfg.get("providers") or {}).get("newapi-local", {}).get("api_key", "")
|
||
except Exception:
|
||
return os.environ.get("NEWAPI_KEY", "")
|
||
try:
|
||
import xlrd
|
||
except ImportError:
|
||
print("需要 xlrd: pip install xlrd")
|
||
sys.exit(1)
|
||
|
||
# 模型链：2B 本地(免费) → agnes(付费复核)。质量敏感可用 --cloud 强制云
|
||
LOCAL_URL = "http://127.0.0.1:8080/v1/chat/completions"
|
||
LOCAL_MODEL = "minicpm5-2b"
|
||
LLAMA_URL = "http://127.0.0.1:3000/v1/chat/completions" # 统一走 newapi
|
||
AGNES_MODEL = "agnes-2.5-flash"
|
||
REVIEW_PROMPT = """你是会计凭证摘要质检员。判断这条凭证摘要是否合格。
|
||
|
||
不合格情况（输出对应标记）:
|
||
- placeholder: "会计凭证"/纯公司名/制单审核等元数据占位
|
||
- truncated: 明显截断（如"收费-假日"这种半截、结尾异常）
|
||
- suspicious: 含乱码/特殊字符/与科目矛盾/明显 OCR 错误
|
||
合格: ok
|
||
|
||
凭证摘要: "{summary}"
|
||
科目: {subject}
|
||
|
||
只输出严格 JSON: {"verdict": "ok|placeholder|truncated|suspicious", "reason": "一句话"}"""
|
||
|
||
|
||
def _agnes_key():
|
||
return _gateway_key()
|
||
"""从 .env / config 读 agnes key"""
|
||
try:
|
||
with open(os.path.expanduser("~/.hermes/.env")) as f:
|
||
for line in f:
|
||
if "AGNES_API_KEY" in line:
|
||
return line.strip().split("=", 1)[1].strip()
|
||
except Exception:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def review_one(summary, subject, use_cloud=False, timeout=45):
|
||
"""调 LLM 判断摘要质量。默认 2B 本地免费；use_cloud=True 强制 agnes。
|
||
"""
|
||
prompt = REVIEW_PROMPT.replace("{summary}", summary[:80]).replace("{subject}", subject[:40])
|
||
if not use_cloud:
|
||
# 先试本地 2B
|
||
try:
|
||
payload = json.dumps({
|
||
"model": LOCAL_MODEL,
|
||
"messages": [{"role": "user", "content": prompt}],
|
||
"max_tokens": 60, "temperature": 0.1,
|
||
}).encode()
|
||
req = urllib.request.Request(
|
||
LOCAL_URL, data=payload,
|
||
headers={"Content-Type": "application/json"}, method="POST")
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
body = json.loads(resp.read())
|
||
content = (body.get("choices") or [{}])[0].get("message", {}).get("content", "")
|
||
if content.strip():
|
||
return _parse_verdict(content)
|
||
except Exception:
|
||
pass # 2B 不可用 → 云
|
||
# agnes 云（付费复核）
|
||
key = _agnes_key()
|
||
payload = json.dumps({
|
||
"model": AGNES_MODEL,
|
||
"messages": [{"role": "system", "content": prompt},
|
||
{"role": "user", "content": "请判断"}],
|
||
"max_tokens": 60, "temperature": 0.1,
|
||
}).encode()
|
||
req = urllib.request.Request(
|
||
LLAMA_URL, data=payload,
|
||
headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
|
||
method="POST")
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
body = json.loads(resp.read())
|
||
content = (body.get("choices") or [{}])[0].get("message", {}).get("content", "")
|
||
content = content.strip()
|
||
return _parse_verdict(content)
|
||
except Exception:
|
||
pass
|
||
return {"verdict": "error", "reason": "复核调用失败"}
|
||
|
||
|
||
def _parse_verdict(content):
|
||
content = content.strip()
|
||
if content.startswith("```"):
|
||
content = content.split("```")[1] if "```" in content[3:] else content
|
||
if content.startswith("json"):
|
||
content = content[4:]
|
||
start, end = content.find("{"), content.rfind("}")
|
||
if start >= 0 and end > start:
|
||
try:
|
||
return json.loads(content[start:end + 1])
|
||
except Exception:
|
||
pass
|
||
return {"verdict": "error", "reason": "输出解析失败"}
|
||
|
||
|
||
def main():
|
||
if len(sys.argv) < 2:
|
||
print("用法: python3 kocr-summary-review-4b.py <合并凭证.xls> [--limit N]")
|
||
return
|
||
path = sys.argv[1]
|
||
limit = 200
|
||
use_cloud = False
|
||
args = sys.argv[2:]
|
||
if "--limit" in args:
|
||
try:
|
||
limit = int(args[args.index("--limit") + 1])
|
||
except Exception:
|
||
pass
|
||
if "--cloud" in args:
|
||
use_cloud = True # 质量敏感时强制云复核
|
||
|
||
wb = xlrd.open_workbook(path)
|
||
ws = wb.sheet_by_name("Page1")
|
||
# 找列: 凭证摘要 / 科目名称 / 科目代码 / 凭证号
|
||
headers = [str(ws.cell_value(0, c)).strip() for c in range(ws.ncols)]
|
||
def col(keyword):
|
||
for c, h in enumerate(headers):
|
||
if keyword in h:
|
||
return c
|
||
return None
|
||
ci_sum, ci_subj, ci_code, ci_vno = col("摘要"), col("科目名称"), col("科目代码"), col("凭证号")
|
||
|
||
issues = []
|
||
checked = 0
|
||
for r in range(1, ws.nrows):
|
||
summary = str(ws.cell_value(r, ci_sum or 19)).strip()
|
||
if not summary:
|
||
continue
|
||
checked += 1
|
||
if checked > limit:
|
||
break
|
||
subject = str(ws.cell_value(r, ci_subj or 6)).strip()
|
||
vno = str(ws.cell_value(r, ci_vno or 4)).strip()
|
||
res = review_one(summary, subject, use_cloud=use_cloud)
|
||
if res.get("verdict") != "ok":
|
||
issues.append(f"记-{vno} | {res.get('verdict')} | {summary[:60]} | 科目:{subject[:20]} | {res.get('reason','')[:40]}")
|
||
|
||
print(f"复核 {checked} 条摘要, 可疑 {len(issues)} 条:")
|
||
for line in issues:
|
||
print(line)
|
||
if not issues:
|
||
print("✅ 全部摘要合格（4B 复核）")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|