From 70b6fb37670951b762726feefdaadafeefd2a7fa Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Fri, 24 Jul 2026 16:27:16 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20--limit=20=E5=85=A8=E8=A1=A8=E6=89=AB?= =?UTF-8?q?=E6=8F=8F=E6=94=AF=E6=8C=81=E6=8C=89=E7=9C=9F=E5=AE=9E=20ID=20?= =?UTF-8?q?=E6=8E=92=E5=BA=8F/=E5=8D=87=E9=99=8D=E5=BA=8F/=E8=8C=83?= =?UTF-8?q?=E5=9B=B4=E8=BF=87=E6=BB=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 此前 --limit 仅按总排号排序取前 N 个。现扩展为： - 排序按数据库真实 ID 列（config.id_field），新增 --order {asc,desc}（默认 asc） - 新增 --range START,END 按真实 ID 闭区间过滤（与 --order/--limit 可组合） - 上述三参数仅全表扫描生效，与 --id/--sn/--ids-file 互斥 db.py: fetch_all_ids 新增 order/id_min/id_max 参数，用派生表 （内层 DISTINCT 取 (总排号,ID) 配对，外层按 ID 排序）规避 SQL Server "SELECT DISTINCT 时 ORDER BY 列须在选择列表" 的限制； id_field 未配置时降级按总排号排序并告警。 write_attachments.py: 新增 --order/--range CLI 参数并接线，含 --range 格式校验（须为 START,END 两个整数）；模块 docstring 补示例。 README.md: 写入用法块与项目结构注释补充 --order/--range 说明。 自测（小数据量，只读+部分 dry-run）：--limit 3 升/降序、--range 800,805 升序、--range 800,805 --order desc --limit 3 均验证返回总排号按真实 ID 正确排序且在范围内；dry-run 不落库；--range 格式错误正确报错退出。 Co-Authored-By: WorkBuddy --- README.md | 17 +++++++++++---- db.py | 52 +++++++++++++++++++++++++++++++++++++------- write_attachments.py | 41 +++++++++++++++++++++++++++++----- 3 files changed, 92 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 7875e6f..fbc289f 100644 --- a/README.md +++ b/README.md @@ -235,12 +235,21 @@ JSON 的括号、引号、转义更容易被模型写错。约定 LLM 只输出" `Common.Attachment` 表，供下游系统按总排号查询该订单的附件类目。 ```bash -# 全量写入（按总排号遍历源表全部记录） +# 全量写入（按真实 ID 升序遍历源表全部记录） python write_attachments.py -# 仅前 N 个总排号（测试用，避免一次性消耗大量 LLM 额度） +# 仅前 N 个（按真实 ID 升序，测试用，避免一次性消耗大量 LLM 额度） python write_attachments.py --limit 10 +# 降序取前 N 个 +python write_attachments.py --limit 10 --order desc + +# 按真实 ID 闭区间过滤：仅处理 ID 在 [800,805] 的记录 +python write_attachments.py --range 800,805 + +# 组合：区间内降序、再取前 3 个 +python write_attachments.py --range 800,805 --order desc --limit 3 + # 指定总排号文件（每行一个，键类型 sn） python write_attachments.py --ids-file ids.txt @@ -252,7 +261,7 @@ python write_attachments.py --sn 26B742,26B743,26B744 python write_attachments.py --id 802 python write_attachments.py --id 802,803 -# --id / --sn / --ids-file 可混用；提供任一后不再全表扫描（--limit 此时无效） +# --id / --sn / --ids-file 可混用；提供任一后不再全表扫描（--limit/--order/--range 此时无效） # 粗分类写入（小类统一填占位值）；不指定 --mode 则用配置文件 business.default_mode python write_attachments.py --mode coarse @@ -288,7 +297,7 @@ python write_attachments.py --mode fine --enable-other ``` ├── config.yaml # 配置文件 ├── main.py # 命令行入口：分类并输出 JSON Lines（支持 --summary 在 stderr 打印批汇总；直接 import 同目录各模块） -├── write_attachments.py # 写入入口：分类结果落库到 Common.Attachment（全量/--limit/--id/--sn/--ids-file/--mode/--dry-run/--enable-other/--config；运行后打印 token/缓存汇总） +├── write_attachments.py # 写入入口：分类结果落库到 Common.Attachment（全表扫描支持 --limit/--order/--range 按真实 ID 排序与范围过滤；--id/--sn/--ids-file/--mode/--dry-run/--enable-other/--config；运行后打印 token/缓存汇总） ├── requirements.txt ├── config_loader.py # YAML 配置读取与校验 ├── db.py # SQL Server 查询与落库（fetch_params_by_ids 支持按 id/sn 双键查询、fetch_all_ids / upsert_attachments，pyodbc） diff --git a/db.py b/db.py index 432bce0..5ae72a6 100644 --- a/db.py +++ b/db.py @@ -143,21 +143,57 @@ SENTINEL_MINOR = "无" NO_MINOR = "无" -def fetch_all_ids(db_cfg: dict[str, Any], limit: int | None = None) -> list[str]: - """拉取源表全部总排号（去重、排除 NULL），用于全量分类写入。 +def fetch_all_ids( + db_cfg: dict[str, Any], + limit: int | None = None, + order: str = "asc", + id_min: int | None = None, + id_max: int | None = None, +) -> list[str]: + """拉取源表总排号列表，供全量/范围分类写入。 - limit 用于测试时只取前 N 个（按总排号排序）。返回的总排号列表可直接 - 喂给 classify_batch。 + 排序按数据库真实 ID 列（config.id_field）；未配置 id_field 时降级为按总排号 + 列排序并告警。order 取 "asc" / "desc"。id_min / id_max 按真实 ID 列做闭区间 + 过滤（含端点）；任一为 None 表示不限制该侧。limit 取排序+过滤后的前 N 条。 + + 返回的仍是总排号列表，可直接喂给 classify_batch（键类型统一为 sn）。 + 用派生表（内层 DISTINCT 取 (总排号, ID) 配对，外层按 ID 排序）规避 SQL Server + "SELECT DISTINCT 时 ORDER BY 列须出现在选择列表"的限制。 """ schema = db_cfg["schema"] table = db_cfg["table"] - id_col = db_cfg["id_column"] + sn_col = db_cfg["id_column"] # 总排号列（返回列） + id_field = db_cfg.get("id_field") # 真实 ID 列（排序/过滤用） qualified = f"[{schema}].[{table}]" - sql = f"SELECT DISTINCT [{id_col}] FROM {qualified} WHERE [{id_col}] IS NOT NULL" - params: list[Any] = [] + + order_dir = "DESC" if order == "desc" else "ASC" + inner_where = [f"[{sn_col}] IS NOT NULL"] + inner_params: list[Any] = [] + + if id_field: + inner_where.append(f"[{id_field}] IS NOT NULL") + if id_min is not None: + inner_where.append(f"[{id_field}] >= ?") + inner_params.append(int(id_min)) + if id_max is not None: + inner_where.append(f"[{id_field}] <= ?") + inner_params.append(int(id_max)) + order_expr = "[id]" # 派生表别名 + inner_select = f"SELECT DISTINCT [{sn_col}] AS sn, [{id_field}] AS id" + else: + logger.warning( + "未配置 database.id_field，--limit/--range 将按总排号列排序（无法按真实 ID 排序/范围过滤）" + ) + order_expr = "[sn]" + inner_select = f"SELECT DISTINCT [{sn_col}] AS sn" + + inner_sql = f"{inner_select} FROM {qualified} WHERE {' AND '.join(inner_where)}" + sql = f"SELECT [sn] FROM ({inner_sql}) AS t ORDER BY {order_expr} {order_dir}" + params = list(inner_params) if limit is not None: - sql += f" ORDER BY [{id_col}] OFFSET 0 ROWS FETCH NEXT ? ROWS ONLY" + sql += " OFFSET 0 ROWS FETCH NEXT ? ROWS ONLY" params.append(int(limit)) + conn_str = _build_conn_str(db_cfg) try: with pyodbc.connect(conn_str, timeout=db_cfg["connect_timeout"]) as conn: diff --git a/write_attachments.py b/write_attachments.py index 27e7992..0e26940 100644 --- a/write_attachments.py +++ b/write_attachments.py @@ -10,8 +10,11 @@ - has_attachment=null (not_found / llm_error / db_error): 不写（属"无法确定/失败"）。 用法: - python write_attachments.py # 全量写入 - python write_attachments.py --limit 10 # 仅前 10 个总排号（测试） + python write_attachments.py # 全量写入（按真实 ID 升序） + python write_attachments.py --limit 10 # 仅前 10 个（按真实 ID 升序，测试） + 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 --ids-file ids.txt # 指定总排号（文件，每行一个） python write_attachments.py --sn 26B742 # 单个总排号写入（键类型 sn） python write_attachments.py --sn 26B742,26B743 # 多个总排号（逗号分隔）写入 @@ -126,7 +129,16 @@ def main() -> None: ) parser.add_argument( "--limit", type=int, default=None, - help="仅处理前 N 个总排号（按总排号排序），测试用；与 --id/--sn/--ids-file 互斥", + help="取排序+范围过滤后的前 N 条总排号，测试用；与 --id/--sn/--ids-file 互斥", + ) + parser.add_argument( + "--order", type=str, default="asc", choices=["asc", "desc"], + help="全表扫描时的排序方向（按真实 ID 列排序），默认 asc；与 --id/--sn/--ids-file 互斥", + ) + parser.add_argument( + "--range", type=str, default=None, + help="按真实 ID 列的值域过滤，闭区间，格式 START,END（如 800,805）；" + "与 --order/--limit 可组合；与 --id/--sn/--ids-file 互斥", ) parser.add_argument( "--ids-file", type=str, default=None, @@ -174,12 +186,29 @@ def main() -> None: 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) + # 全表扫描路径：--range/--order/--limit 仅在此生效 + if args.range: + parts = [p.strip() for p in args.range.split(",")] + if len(parts) != 2 or not parts[0] or not parts[1]: + print("[错误] --range 需为 START,END 两个数字，如 800,805", file=sys.stderr) + sys.exit(1) + try: + id_min, id_max = int(parts[0]), int(parts[1]) + except ValueError: + print("[错误] --range 的 START/END 必须为整数", file=sys.stderr) + sys.exit(1) + else: + id_min = id_max = None + sn_list = fetch_all_ids( + cfg["database"], limit=args.limit, order=args.order, + id_min=id_min, id_max=id_max, + ) # 全表扫描得到的是总排号，键类型统一为 sn ids = [(s, "sn") for s in sn_list] logger.info( - "源表读取 %d 个总排号%s", len(ids), - f"（limit={args.limit}）" if args.limit else "", + "源表读取 %d 个总排号（order=%s%s%s）", len(ids), args.order, + f", range=[{id_min},{id_max}]" if args.range else "", + f", limit={args.limit}" if args.limit else "", ) if not ids: