#!/usr/bin/env python3 # -*- coding: utf-8 -*- """基于源表总排号,调用 classify_batch 分类,将结果按约定写入 Common.Attachment。 落库约定(详见项目记忆 MEMORY.md,与 db.py 中的常量保持一致): - has_attachment=true (status=ok): 每个 type 一行。fine 模式 type 形如 "大类:细分",按首个冒号拆成 (大类, 小类); coarse 模式 type 是大类,小类统一填 '无'。 - has_attachment=false (无附件): 写哨兵行 (SN, '无附件', '无')。 - has_attachment=null (not_found / llm_error / db_error): 不写(属"无法确定/失败")。 用法: python write_attachments.py # 全量写入 python write_attachments.py --limit 10 # 仅前 10 个总排号(测试) python write_attachments.py --ids-file ids.txt # 指定总排号(文件,每行一个) python write_attachments.py --sn 26B742 # 单个总排号写入(键类型 sn) python write_attachments.py --sn 26B742,26B743 # 多个总排号(逗号分隔)写入 python write_attachments.py --id 802 # 单个数据库真实 ID 写入(键类型 id) python write_attachments.py --id 802,803 # 多个真实 ID(逗号分隔)写入 python write_attachments.py --mode coarse # 粗分类写入 python write_attachments.py --dry-run # 只打印将写入的行,不落库 python write_attachments.py --config other.yaml 指定方式(--id / --sn / --ids-file 可混用): --id 数据库真实 ID 列(id_field),如 802 --sn 总排号列(id_column),如 26B742(即原先 --id 的语义) --ids-file 每行一个总排号(键类型 sn) 无论用哪种方式,结果均按回查到的总排号写入 Common.Attachment.SN。 """ from __future__ import annotations import argparse import logging import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent / "src")) from classifier import classify_batch, summarize_results # noqa: E402 from config_loader import ConfigError, load_config # noqa: E402 from db import ( # noqa: E402 ATTACHMENT_SCHEMA, ATTACHMENT_TABLE, DatabaseError, SENTINEL_MAJOR, SENTINEL_MINOR, NO_MINOR, fetch_all_ids, upsert_attachments, ) def convert_results_to_rows(results: list[dict], mode: str) -> tuple[list[tuple[str, str, str]], list[str]]: """将 classify_batch 的结果列表转成可写入 Common.Attachment 的行。 返回 (rows, skipped): rows - list[(SN, MajorCategory, MinorCategory)] skipped- list[总排号],has_attachment 为 None(not_found/失败)未写入的 """ rows: list[tuple[str, str, str]] = [] skipped: list[str] = [] for r in results: zph = r["zong_pai_hao"] has = r["has_attachment"] if has is None: skipped.append(zph) continue if has is True: for t in r["types"]: t = t.strip() if not t: continue if mode == "fine" and ":" in t: major, minor = t.split(":", 1) major = major.strip() minor = minor.strip() or NO_MINOR else: major = t minor = NO_MINOR rows.append((zph, major, minor)) else: # has is False -> 无附件 rows.append((zph, SENTINEL_MAJOR, SENTINEL_MINOR)) return rows, skipped def _parse_ids(args: argparse.Namespace) -> list[tuple[str, str]]: """从 --id / --sn / --ids-file 解析出 (标识符, 键类型) 列表。 --id -> 键类型 "id"(数据库真实 ID 列) --sn -> 键类型 "sn"(总排号列,即原先 --id 的语义) --ids-file -> 每行一个总排号,键类型 "sn" 三者可同时提供、合并后返回(顺序:--id, --sn, --ids-file)。 --ids-file 文件不存在时直接报错退出。 """ items: list[tuple[str, str]] = [] if args.id: items.extend((s.strip(), "id") for s in args.id.split(",") if s.strip()) if args.sn: items.extend((s.strip(), "sn") for s in args.sn.split(",") if s.strip()) if args.ids_file: p = Path(args.ids_file) if not p.exists(): print(f"[错误] 总排号文件不存在: {p.resolve()}", file=sys.stderr) sys.exit(1) with p.open("r", encoding="utf-8") as f: items.extend((line.strip(), "sn") for line in f if line.strip()) return items def main() -> None: parser = argparse.ArgumentParser( description="将订单附件分类结果写入 Common.Attachment", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument("--config", type=str, default="config.yaml") parser.add_argument( "--id", type=str, default=None, help="数据库真实 ID(如 802),单个或逗号分隔的多个;提供后按真实 ID 列查询," "不再全表扫描(--limit 此时无效)", ) parser.add_argument( "--sn", type=str, default=None, help="总排号(如 26B742),单个或逗号分隔的多个(即原先 --id 的语义);" "提供后直接按总排号写入,不再全表扫描(--limit 此时无效)", ) parser.add_argument( "--limit", type=int, default=None, help="仅处理前 N 个总排号(按总排号排序),测试用;与 --id/--sn/--ids-file 互斥", ) parser.add_argument( "--ids-file", type=str, default=None, help="包含总排号的文本文件路径,每行一个总排号(键类型 sn);可单独使用或与 --id/--sn 合并", ) parser.add_argument( "--mode", type=str, default=None, choices=["coarse", "fine"], help="分类粒度,不指定则用配置文件 business.default_mode", ) parser.add_argument("--log-dir", type=str, default=None) parser.add_argument("--enable-other", action="store_true") parser.add_argument( "--dry-run", action="store_true", help="只打印将要写入/跳过的统计,不连接目标表写入", ) args = parser.parse_args() try: cfg = load_config(args.config) except ConfigError as e: print(f"[配置错误] {e}", file=sys.stderr) sys.exit(1) logging.basicConfig( level=getattr(logging, cfg["business"]["log_level"], logging.INFO), format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", stream=sys.stderr, ) logger = logging.getLogger("write_attachments") mode = args.mode or cfg["business"]["default_mode"] log_dir = args.log_dir or cfg["business"]["log_dir"] # enable_other_category 是可选配置项,缺省按"关闭";--enable-other 只能从关到开 enable_other = bool(args.enable_other) or bool( cfg["business"].get("enable_other_category", False) ) # 1) 取总排号列表 if args.id or args.sn or args.ids_file: ids = _parse_ids(args) if not ids: print("[错误] --id / --sn / --ids-file 未解析到任何有效的标识符", file=sys.stderr) sys.exit(1) n_id = sum(1 for _, k in ids if k == "id") n_sn = sum(1 for _, k in ids if k == "sn") logger.info("从 --id/--sn/--ids-file 读取 %d 个标识符(id=%d, sn=%d)", len(ids), n_id, n_sn) else: sn_list = fetch_all_ids(cfg["database"], limit=args.limit) # 全表扫描得到的是总排号,键类型统一为 sn ids = [(s, "sn") for s in sn_list] logger.info( "源表读取 %d 个总排号%s", len(ids), f"(limit={args.limit})" if args.limit else "", ) if not ids: print("[提示] 没有可处理的总排号,退出") return # 2) 分类 logger.info("开始分类 (mode=%s, enable_other=%s)", mode, enable_other) results = classify_batch( db_cfg=cfg["database"], llm_cfg=cfg["llm"], id_list=ids, max_workers=cfg["business"]["max_workers"], mode=mode, log_dir=log_dir, enable_other=enable_other, ) # 3) 转换 rows, skipped = convert_results_to_rows(results, mode) # 统计各状态 n_ok = sum(1 for r in results if r["status"] == "ok") n_empty = sum(1 for r in results if r["status"] == "empty_param") n_notfound = sum(1 for r in results if r["status"] == "not_found") n_err = sum(1 for r in results if r["status"].endswith("error")) print( f"[统计] 输入 {len(results)} 个总排号: " f"ok={n_ok} empty_param={n_empty} not_found={n_notfound} error={n_err}" ) print( f"[统计] 将写入 {len(rows)} 行, " f"跳过(无附件判定=null) {len(skipped)} 个" ) # 运行汇总:总耗时/token 总量/缓存命中率,从每条结果的 meta 累加。 # 走 stdout 与上面的 [统计] 一致;不属于落库动作本身,仅作诊断输出。 summary = summarize_results(results) print( f"[汇总] 总耗时 {summary['total_elapsed_ms']} ms, " f"LLM 调用 {summary['llm_calls']} 次, " f"token: prompt={summary['total_prompt_tokens']} " f"completion={summary['total_completion_tokens']} " f"total={summary['total_tokens']}, " f"缓存命中率 {summary['avg_cache_hit_rate']} " f"(hit={summary['total_cache_hit_tokens']} " f"miss={summary['total_cache_miss_tokens']})" ) if args.dry_run: print("[dry-run] 不落库。示例行(前 10):") for row in rows[:10]: print(" ", row) return # 4) 写入 target = f"{ATTACHMENT_SCHEMA}.{ATTACHMENT_TABLE}" try: deleted, inserted = upsert_attachments(cfg["database"], rows) print(f"[完成] 已写入 {target}: 删除旧行 {deleted}, 插入新行 {inserted}") except DatabaseError as e: print(f"[写入失败] {e}", file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()