From 2ae5621090a7e30b16e133a1fa6b121e5684a80e Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Tue, 28 Jul 2026 11:17:17 +0800 Subject: [PATCH] feat(write_attachments): add --append mode, fix batch timing and SQL Server param limit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add --append flag: diff source-table 总排号 against existing Common.Attachment SN and classify/write only the missing ones. --limit caps the per-run append count, --order sets the direction. Backed by new fetch_existing_attachment_sns() in db.py. - Fix batch elapsed-time accounting: summarize_results() summed each task's per-item elapsed_ms, which overcounts under ThreadPoolExecutor concurrency (cumulative work time, not real wall-clock — 100 tasks on 8 workers reported ~5x the actual runtime). Callers now time classify_batch() via perf_counter and pass wall_clock_ms; both the write_attachments [汇总] line and main.py --summary report wall-clock separately from the cumulative sum. - Format durations >=1s in seconds (88851.4 ms -> 88.85 s) in the human-readable [汇总] line; structured JSON --summary fields stay in ms. - Chunk all IN (...) lists to 2000 items to respect SQL Server's 2100 bind-parameter hard limit (previously --append --limit 5000 failed at the fetch step with "COUNT 字段不正确"). Applied to fetch_params_by_ids (sn / id paths) and to the DELETE inside upsert_attachments. Co-Authored-By: Claude --- README.md | 19 +++++++-- classifier.py | 22 +++++++--- db.py | 99 +++++++++++++++++++++++++++++++------------- main.py | 8 +++- write_attachments.py | 60 +++++++++++++++++++++++++-- 5 files changed, 166 insertions(+), 42 deletions(-) diff --git a/README.md b/README.md index 69139fc..97c06a1 100644 --- a/README.md +++ b/README.md @@ -173,9 +173,13 @@ python main.py --sn 26B742,26B743 --summary "meta": {"elapsed_ms": 1476.6, "attempts": 1, "llm": {"model": "deepseek-v4-flash", "elapsed_ms": 1472.8, "usage": {"prompt_tokens": 1297, "completion_tokens": 15, "total_tokens": 1312, "prompt_cache_hit_tokens": 1280, "prompt_cache_miss_tokens": 17, "cache_hit_rate": 0.9869}}} ``` -`--summary` 打印的整批汇总字段与 `meta.llm.usage` 对应：`count` / `total_elapsed_ms` / `llm_calls` / -各类 `total_*_tokens` / `total_cache_hit_tokens` / `total_cache_miss_tokens` / `avg_cache_hit_rate` -（按 token 加权的整体命中率）。 +`--summary` 打印的整批汇总字段与 `meta.llm.usage` 对应：`count` / `wall_clock_ms` / `total_elapsed_ms` / +`llm_calls` / 各类 `total_*_tokens` / `total_cache_hit_tokens` / `total_cache_miss_tokens` / +`avg_cache_hit_rate`（按 token 加权的整体命中率）。 + +两个耗时字段的区别（`classify_batch` 是并发执行的）： +- `wall_clock_ms`：整批分类的**真实墙钟**耗时（在 `classify_batch` 外层用 `perf_counter` 计时），即你实际等的时间。 +- `total_elapsed_ms`：各结果自身耗时 `elapsed_ms` 的**累加**（相当于把这些任务串行跑的总时长），并发下会明显大于墙钟；二者之比 ≈ 并发增益（≈ `max_workers`）。 ## 对话日志 @@ -265,6 +269,11 @@ python write_attachments.py --range 800,805 # 组合：区间内降序、再取前 3 个 python write_attachments.py --range 800,805 --order desc --limit 3 +# 追加模式：仅补写源表中存在、但 Common.Attachment 还没有的总排号 +# --limit 此时是"本次最多追加多少个 SN"的上限；--order 决定从哪一端补起 +python write_attachments.py --append --limit 10 +python write_attachments.py --append --order desc --limit 10 + # 指定总排号文件（每行一个，键类型 sn） python write_attachments.py --ids-file ids.txt @@ -293,7 +302,9 @@ python write_attachments.py --mode fine --enable-other ``` 运行结束后，除上面的 `ok/empty_param/not_found/error` 与"将写入行数/跳过数"统计外，还会打印一行 -`[汇总]`：本批总耗时、LLM 调用次数、token 总量（prompt/completion/total）与加权缓存命中率，便于核算成本。 +`[汇总]`：本批**总耗时**（整批分类的真实墙钟）、**LLM 累计**（各任务自身耗时的累加，并发下大于墙钟， +括号标注并发数 `max_workers`）、LLM 调用次数、token 总量（prompt/completion/total）与加权缓存命中率， +便于核算成本与并发效率。 落库约定（与 `db.py` 常量、`write_attachments.py` 的转换逻辑保持一致）： diff --git a/classifier.py b/classifier.py index 26100a5..4190500 100644 --- a/classifier.py +++ b/classifier.py @@ -265,12 +265,21 @@ def classify_single( )[0] -def summarize_results(results: list[dict[str, Any]]) -> dict[str, Any]: +def summarize_results( + results: list[dict[str, Any]], + wall_clock_ms: float | None = None, +) -> dict[str, Any]: """汇总一批 classify_batch 结果的运行统计。 - 从每条结果的 meta 里累加：总耗时、实际 LLM 调用次数、各类 token 总量、 - 以及整体缓存命中率（按 prompt_cache_hit_tokens / prompt_tokens 加权）。 - 供 CLI 打印批统计用，不改变 classify_batch 的返回结构。 + 从每条结果的 meta 里累加：各结果 elapsed_ms 的总和、实际 LLM 调用次数、 + 各类 token 总量，以及整体缓存命中率（按 prompt_cache_hit_tokens / + prompt_tokens 加权）。供 CLI 打印批统计用，不改变 classify_batch 的返回结构。 + + 注意耗时两个字段的区别（classify_batch 是并发执行的）： + - total_elapsed_ms：各结果自身耗时(elapsed_ms)的**累加**，即"累计工作量" + （相当于把这些任务串行跑的总时长），并发下会明显大于真实墙钟； + - wall_clock_ms：调用方传入的整批分类**真实墙钟**耗时（在 classify_batch + 外层用 perf_counter 计时），二者之比 ≈ 并发增益（≈ max_workers）。 """ total_elapsed = 0.0 llm_calls = 0 @@ -291,7 +300,7 @@ def summarize_results(results: list[dict[str, Any]]) -> dict[str, Any]: sum_hit += u.get("prompt_cache_hit_tokens", 0) or 0 sum_miss += u.get("prompt_cache_miss_tokens", 0) or 0 avg_cache_hit_rate = round(sum_hit / sum_prompt, 4) if sum_prompt else 0.0 - return { + out = { "count": len(results), "total_elapsed_ms": round(total_elapsed, 1), "llm_calls": llm_calls, @@ -302,3 +311,6 @@ def summarize_results(results: list[dict[str, Any]]) -> dict[str, Any]: "total_cache_miss_tokens": sum_miss, "avg_cache_hit_rate": avg_cache_hit_rate, } + if wall_clock_ms is not None: + out["wall_clock_ms"] = round(float(wall_clock_ms), 1) + return out diff --git a/db.py b/db.py index 65ff2f3..28fbff3 100644 --- a/db.py +++ b/db.py @@ -43,11 +43,24 @@ __all__ = [ "fetch_params_by_ids", "fetch_param_by_id", "fetch_all_ids", + "fetch_existing_attachment_sns", "upsert_attachments", "init_schema", ] +# SQL Server 单条语句最多 2100 个绑定参数，WHERE col IN (?, ?, ...) 每项占 1 个， +# 超过会报 "COUNT 字段不正确或语法错误"。这里把超长 IN 列表按 2000 一批分片查询 +# （2000 < 2100，留安全余量；对 PostgreSQL 的 65535 上限也完全够用）。 +_PARAM_CHUNK = 2000 + + +def _chunked(seq: list, size: int = _PARAM_CHUNK): + """把列表切成 <= size 的连续片段，用于分批 IN 查询。""" + for i in range(0, len(seq), size): + yield seq[i : i + size] + + def _build_source_table(db_cfg: dict[str, Any]): """按配置动态构造源表 Core 结构（带 quote，保证跨库标识符正确引用）。""" schema = db_cfg["schema"] @@ -102,14 +115,15 @@ def fetch_params_by_ids( def _fill_sn() -> None: if not sn_items: return - stmt = select(sn_c, param_c).where(sn_c.in_(sn_items)) with engine.connect() as conn: - for row in conn.execute(stmt): - ident = str(row[0]) - entry = result.get(ident) - if entry is not None: - entry["param"] = row[1] - entry["sn"] = ident + for chunk in _chunked(sn_items): + stmt = select(sn_c, param_c).where(sn_c.in_(chunk)) + for row in conn.execute(stmt): + ident = str(row[0]) + entry = result.get(ident) + if entry is not None: + entry["param"] = row[1] + entry["sn"] = ident def _fill_id() -> None: if not id_items: @@ -119,14 +133,15 @@ def fetch_params_by_ids( logger.warning( "未配置 database.id_field，--id 将退化为按总排号列 [%s] 查询", sn_col ) - stmt = select(sn_c, param_c).where(sn_c.in_(id_items)) with engine.connect() as conn: - for row in conn.execute(stmt): - ident = str(row[0]) - entry = result.get(ident) - if entry is not None: - entry["param"] = row[1] - entry["sn"] = ident + for chunk in _chunked(id_items): + stmt = select(sn_c, param_c).where(sn_c.in_(chunk)) + for row in conn.execute(stmt): + ident = str(row[0]) + entry = result.get(ident) + if entry is not None: + entry["param"] = row[1] + entry["sn"] = ident return id_c = src.c[id_field] @@ -144,17 +159,18 @@ def fetch_params_by_ids( if not id_ints: return - stmt = select(id_c, sn_c, param_c).where(id_c.in_(id_ints)) with engine.connect() as conn: - for row in conn.execute(stmt): - idv = row[0] - orig = id_int_map.get(idv) - if orig is None: - continue - entry = result.get(orig) - if entry is not None: - entry["param"] = row[2] - entry["sn"] = row[1] + for chunk in _chunked(id_ints): + stmt = select(id_c, sn_c, param_c).where(id_c.in_(chunk)) + for row in conn.execute(stmt): + idv = row[0] + orig = id_int_map.get(idv) + if orig is None: + continue + entry = result.get(orig) + if entry is not None: + entry["param"] = row[2] + entry["sn"] = row[1] try: _fill_sn() @@ -227,6 +243,26 @@ def fetch_all_ids( raise DatabaseError(f"查询总排号失败: {e}") from e +def fetch_existing_attachment_sns(db_cfg: dict[str, Any]) -> set[str]: + """读取 Common.Attachment 中已存在的全部 SN（distinct）。 + + 用于"追加"模式做差集：源表总排号集合 减去 这里返回的 SN 集合， + 得到"源表有、但目标表还没写过"的总排号，只对这部分分类写入。 + + 返回 set[str]（空表返回空集合）；SN 统一转成 str，便于与源表总排号 + （同样是 str）直接做集合比较。NULL 值会被过滤掉。 + """ + engine = get_engine(db_cfg) + try: + with engine.connect() as conn: + stmt = select(Attachment.SN).distinct() + return { + str(row[0]) for row in conn.execute(stmt) if row[0] is not None + } + except SQLAlchemyError as e: + raise DatabaseError(f"读取已有附件 SN 失败: {e}") from e + + def upsert_attachments( db_cfg: dict[str, Any], rows: list[tuple[str, str, str]], @@ -237,6 +273,9 @@ def upsert_attachments( 对出现的每个 SN 先 DELETE 其旧行，再批量 INSERT——保证重跑总是反映 最新分类结果，不会因唯一约束（复合主键 (SN,大类,小类)）冲突而失败。 + DELETE 的 IN 列表按 _PARAM_CHUNK 分片执行，避免 SN 过多时单条语句超过 + SQL Server 的 2100 绑定参数上限；整个"删 + 插"在同一事务内完成，语义不变。 + 返回 (deleted_rows, inserted_rows) 计数。 """ if not rows: @@ -247,10 +286,14 @@ def upsert_attachments( try: with Session(engine) as session: - del_res = session.execute( - delete(Attachment).where(Attachment.SN.in_(distinct_sn)) - ) - deleted = del_res.rowcount if del_res.rowcount is not None else 0 + # 分片 DELETE：单条 IN (...) 最多 _PARAM_CHUNK 个参数，避免触发 2100 上限 + deleted = 0 + for chunk in _chunked(distinct_sn): + del_res = session.execute( + delete(Attachment).where(Attachment.SN.in_(chunk)) + ) + if del_res.rowcount: + deleted += del_res.rowcount session.execute( insert(Attachment), [ diff --git a/main.py b/main.py index 1298c51..64e07d0 100644 --- a/main.py +++ b/main.py @@ -62,6 +62,7 @@ import argparse import json import logging import sys +import time from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent / "src")) @@ -166,6 +167,10 @@ def main() -> None: # 只能从"关"打开成"开"，不支持反过来用命令行强制关闭配置文件里已打开的设置。 enable_other = bool(args.enable_other) or bool(cfg["business"].get("enable_other_category", False)) + # 整批分类的真实墙钟：在并发 classify_batch 外层计时（perf_counter）。 + # summarize_results 累加的 total_elapsed_ms 是各任务自身耗时的总和（并发下的 + # 累计工作量，会明显大于墙钟），二者之比 ≈ 并发增益；wall_clock_ms 才是真实墙钟。 + t0 = time.perf_counter() results = classify_batch( db_cfg=cfg["database"], llm_cfg=cfg["llm"], @@ -175,6 +180,7 @@ def main() -> None: log_dir=log_dir, enable_other=enable_other, ) + wall_ms = (time.perf_counter() - t0) * 1000 if args.pretty: print(json.dumps(results, ensure_ascii=False, indent=2)) @@ -183,7 +189,7 @@ def main() -> None: print(json.dumps(r, ensure_ascii=False)) if args.summary: - summary = summarize_results(results) + summary = summarize_results(results, wall_clock_ms=wall_ms) print("--- 运行汇总 ---", file=sys.stderr) print(json.dumps(summary, ensure_ascii=False, indent=2), file=sys.stderr) diff --git a/write_attachments.py b/write_attachments.py index 85a57cc..91b33f7 100644 --- a/write_attachments.py +++ b/write_attachments.py @@ -15,6 +15,8 @@ python write_attachments.py --limit 10 --order desc # 按真实 ID 降序取前 10 个 python write_attachments.py --range 800,805 # 真实 ID 在 [800,805] 闭区间内 python write_attachments.py --range 800,805 --order desc --limit 3 # 组合：区间内降序取前3 + python write_attachments.py --append --limit 10 # 追加模式：仅补写源表中还没写过的前 10 个总排号 + python write_attachments.py --append --order desc --limit 10 # 追加模式：从最大 ID 端开始补 10 个 python write_attachments.py --ids-file ids.txt # 指定总排号（文件，每行一个） python write_attachments.py --sn 26B742 # 单个总排号写入（键类型 sn） python write_attachments.py --sn 26B742,26B743 # 多个总排号（逗号分隔）写入 @@ -35,6 +37,7 @@ from __future__ import annotations import argparse import logging import sys +import time from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent / "src")) @@ -49,12 +52,28 @@ from db import ( # noqa: E402 SENTINEL_MINOR, NO_MINOR, fetch_all_ids, + fetch_existing_attachment_sns, get_engine, init_schema, upsert_attachments, ) +def _format_duration(ms: float | None) -> str: + """把毫秒耗时格式化为人类友好的时长字符串。 + + 超过 1 秒（>=1000ms）按秒输出（保留 2 位小数），否则按毫秒（保留 1 位小数）。 + 例：88851.4 -> "88.85 s"，458.9 -> "458.9 ms"。None / 异常值回退为 "0 ms"。 + """ + try: + v = float(ms) # type: ignore[arg-type] + except (TypeError, ValueError): + return "0 ms" + if v >= 1000: + return f"{v / 1000:.2f} s" + return f"{v:.1f} ms" + + def convert_results_to_rows(results: list[dict], mode: str) -> tuple[list[tuple[str, str, str]], list[str]]: """将 classify_batch 的结果列表转成可写入 Common.Attachment 的行。 @@ -146,6 +165,13 @@ def main() -> None: "--ids-file", type=str, default=None, help="包含总排号的文本文件路径，每行一个总排号（键类型 sn）；可单独使用或与 --id/--sn 合并", ) + parser.add_argument( + "--append", action="store_true", + help="追加模式：对比源表全部总排号与 Common.Attachment 已有 SN，仅对" + "「源表存在、Attachment 尚未写入」的总排号分类写入。--limit 此时是" + "「本次最多追加多少个 SN」的上限，--order 决定从哪一端（升/降序）补起；" + "与 --id/--sn/--ids-file 互斥", + ) parser.add_argument( "--mode", type=str, default=None, choices=["coarse", "fine"], help="分类粒度，不指定则用配置文件 business.default_mode", @@ -194,6 +220,9 @@ def main() -> None: # 1) 取总排号列表 if args.id or args.sn or args.ids_file: + if args.append: + print("[错误] --append 不能与 --id/--sn/--ids-file 同时使用", file=sys.stderr) + sys.exit(1) ids = _parse_ids(args) if not ids: print("[错误] --id / --sn / --ids-file 未解析到任何有效的标识符", file=sys.stderr) @@ -201,6 +230,22 @@ def main() -> None: 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) + elif args.append: + # 追加模式：源表全部总排号 − Common.Attachment 已有 SN = 待补写集合。 + # 全表拉取源表总排号（按真实 ID 排序），再做差集；--order 控制补写方向， + # --limit 此时是"本次最多追加多少个 SN"的上限。 + all_sns = fetch_all_ids(cfg["database"], order=args.order) + existing = fetch_existing_attachment_sns(cfg["database"]) + missing = [s for s in all_sns if s not in existing] + print( + f"[追加] 源表 {len(all_sns)} 个总排号，Common.Attachment 已有 " + f"{len(existing)} 个，待追加 {len(missing)} 个（order={args.order}）" + ) + if args.limit is not None: + missing = missing[: args.limit] + print(f"[追加] 应用上限 --limit {args.limit}，本次将处理 {len(missing)} 个") + ids = [(s, "sn") for s in missing] + logger.info("追加模式：待处理 %d 个总排号", len(ids)) else: # 全表扫描路径：--range/--order/--limit 仅在此生效 if args.range: @@ -233,6 +278,10 @@ def main() -> None: # 2) 分类 logger.info("开始分类 (mode=%s, enable_other=%s)", mode, enable_other) + # 整批分类的真实墙钟：在并发 classify_batch 外层计时（perf_counter）。 + # 注意 summarize_results 里累加的 total_elapsed_ms 是各任务自身耗时的总和 + # （并发下的"累计工作量"，会明显大于这里测得的墙钟），二者之比 ≈ 并发增益。 + t0 = time.perf_counter() results = classify_batch( db_cfg=cfg["database"], llm_cfg=cfg["llm"], @@ -242,6 +291,7 @@ def main() -> None: log_dir=log_dir, enable_other=enable_other, ) + wall_ms = (time.perf_counter() - t0) * 1000 # 3) 转换 rows, skipped = convert_results_to_rows(results, mode) @@ -261,12 +311,14 @@ def main() -> None: f"跳过(无附件判定=null) {len(skipped)} 个" ) - # 运行汇总：总耗时/token 总量/缓存命中率，从每条结果的 meta 累加。 + # 运行汇总：墙钟耗时 + 各任务累计工作量 + token 总量/缓存命中率。 # 走 stdout 与上面的 [统计] 一致；不属于落库动作本身，仅作诊断输出。 - summary = summarize_results(results) + summary = summarize_results(results, wall_clock_ms=wall_ms) print( - f"[汇总] 总耗时 {summary['total_elapsed_ms']} ms, " - f"LLM 调用 {summary['llm_calls']} 次, " + f"[汇总] 总耗时 {_format_duration(summary['wall_clock_ms'])}, " + f"LLM 累计 {_format_duration(summary['total_elapsed_ms'])} " + f"(并发 {cfg['business']['max_workers']}), " + f"调用 {summary['llm_calls']} 次, " f"token: prompt={summary['total_prompt_tokens']} " f"completion={summary['total_completion_tokens']} " f"total={summary['total_tokens']}, "