ProductionDatabase/attachment_classifier
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: --id 改为真实 ID 列，新增 --sn 对应总排号

Browse Source 
此前接口 --id 实际按总排号列查询，与数据库真实 ID 字段语义混淆。
现明确区分两种键类型：
- --id  -> 数据库真实 ID 列（config id_field）
- --sn  -> 总排号列（config id_column，即原先 --id 的语义）
- --ids-file 视为总排号（键类型 sn）

db.py: fetch_params_by_ids 改为接受 (标识符, 键类型) 列表，返回
  {标识符: {param, sn}}；id 键回取对应总排号；新增 id_field 配置读取
classifier.py: classify_batch/classify_single 透传键类型，输出 zong_pai_hao
  一律为回查到的总排号，not_found 时回退输入标识符便于追溯
main.py / write_attachments.py: 新增 --sn，--id 改指真实 ID，二者可混用；
  全表扫描回退仍按总排号(sn)查询
README.md: 对齐 --id/--sn 语义并补充 id_field 配置说明

注：config.yaml 含密钥被 .gitignore 忽略，id_field 仅存于本地配置；
db.py 在未配置 id_field 时优雅降级为按总排号查询并告警。

Co-Authored-By: WorkBuddy <workbuddy@tencent.com>
This commit is contained in:
Misaka_Company
2026-07-24 16:00:21 +08:00
parent 622f348cf1
commit ee566e3fbd
5 changed files with 236 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
README.md
View File

@@ -20,7 +20,9 @@ pip install -r requirements.txt
编辑 `config.yaml`，填入以下三部分（首次使用需替换为真实值，**请勿将含真实 编辑 `config.yaml`，填入以下三部分（首次使用需替换为真实值，**请勿将含真实
数据库密码 / API Key 的配置文件提交到版本库**）： 数据库密码 / API Key 的配置文件提交到版本库**）：
- `database`：SQL Server 连接信息（含 `schema`）、表名、字段名 - `database`：SQL Server 连接信息（含 `schema`）、表名、字段名。其中
`id_column` 为**总排号**列（接口 `--sn` 使用），`id_field` 为数据库**真实 ID**
列（接口 `--id` 使用）；二者需按源表实际列名填写
- `llm`：OpenAI 兼容接口的 `base_url`、`api_key`、`model` - `llm`：OpenAI 兼容接口的 `base_url`、`api_key`、`model`
- `business`：并发数、日志级别、默认分类模式(`default_mode`)、日志目录(`log_dir`)、 - `business`：并发数、日志级别、默认分类模式(`default_mode`)、日志目录(`log_dir`)、
是否启用"其他"兜底类目(`enable_other_category`，默认关闭) 是否启用"其他"兜底类目(`enable_other_category`，默认关闭)
@@ -28,32 +30,41 @@ pip install -r requirements.txt
## 使用 ## 使用
```bash ```bash
# 指定方式有两种键类型，可混用：
# --sn 按总排号列查询（如 26B742），即原先 --id 的语义
# --id 按数据库真实 ID 列查询（如 802）
# 无论用哪种，输出 JSON 的 zong_pai_hao 一律为回查到的总排号
# 单个总排号，默认粗分类 # 单个总排号，默认粗分类
python main.py --id 26B742 python main.py --sn 26B742
# 批量，逗号分隔 # 批量，逗号分隔
python main.py --id 26B742,26B743,26B744 python main.py --sn 26B742,26B743,26B744
# 按数据库真实 ID 指定（单个 / 多个）
python main.py --id 802
python main.py --id 802,803,804
# 批量，文件输入（每行一个总排号） # 批量，文件输入（每行一个总排号）
python main.py --ids-file ids.txt python main.py --ids-file ids.txt
# 精分类：输出"大类:细分类目"组合 # 精分类：输出"大类:细分类目"组合
python main.py --id 26B742 --mode fine python main.py --sn 26B742 --mode fine
# 指定其他配置文件 # 指定其他配置文件
python main.py --id 26B742 --config other_config.yaml python main.py --id 802 --config other_config.yaml
# 临时覆盖日志目录 # 临时覆盖日志目录
python main.py --id 26B742 --log-dir /tmp/debug_logs python main.py --sn 26B742 --log-dir /tmp/debug_logs
# 格式化输出（默认是紧凑的 JSON Lines，每行一条） # 格式化输出（默认是紧凑的 JSON Lines，每行一条）
python main.py --id 26B742,26B743 --pretty python main.py --sn 26B742,26B743 --pretty
# 允许模型使用"其他"兜底类目 # 允许模型使用"其他"兜底类目
python main.py --id 26B742 --mode fine --enable-other python main.py --sn 26B742 --mode fine --enable-other
# 额外在 stderr 打印本批运行汇总（耗时/token/缓存命中率）；stdout 仍只输出干净 JSON Lines # 额外在 stderr 打印本批运行汇总（耗时/token/缓存命中率）；stdout 仍只输出干净 JSON Lines
python main.py --id 26B742,26B743 --summary python main.py --sn 26B742,26B743 --summary
``` ```
## 分类粒度：coarse / fine ## 分类粒度：coarse / fine
@@ -230,12 +241,18 @@ python write_attachments.py
# 仅前 N 个总排号（测试用，避免一次性消耗大量 LLM 额度） # 仅前 N 个总排号（测试用，避免一次性消耗大量 LLM 额度）
python write_attachments.py --limit 10 python write_attachments.py --limit 10
# 指定总排号文件（每行一个） # 指定总排号文件（每行一个，键类型 sn）
python write_attachments.py --ids-file ids.txt python write_attachments.py --ids-file ids.txt
# 直接指定总排号：单个，或逗号分隔的多个（与 --ids-file 可合并，提供后不再全表扫描） # 按总排号指定（键类型 sn，即原先 --id 的语义）：单个或逗号分隔多个
python write_attachments.py --id 26B742 python write_attachments.py --sn 26B742
python write_attachments.py --id 26B742,26B743,26B744 python write_attachments.py --sn 26B742,26B743,26B744
# 按数据库真实 ID 指定（键类型 id）：单个或逗号分隔多个
python write_attachments.py --id 802
python write_attachments.py --id 802,803
# --id / --sn / --ids-file 可混用；提供任一后不再全表扫描（--limit 此时无效）
# 粗分类写入（小类统一填占位值）；不指定 --mode 则用配置文件 business.default_mode # 粗分类写入（小类统一填占位值）；不指定 --mode 则用配置文件 business.default_mode
python write_attachments.py --mode coarse python write_attachments.py --mode coarse
@@ -271,10 +288,10 @@ python write_attachments.py --mode fine --enable-other
``` ```
├── config.yaml # 配置文件 ├── config.yaml # 配置文件
├── main.py # 命令行入口：分类并输出 JSON Lines（支持 --summary 在 stderr 打印批汇总；直接 import 同目录各模块） ├── main.py # 命令行入口：分类并输出 JSON Lines（支持 --summary 在 stderr 打印批汇总；直接 import 同目录各模块）
├── write_attachments.py # 写入入口：分类结果落库到 Common.Attachment（全量/--limit/--ids-file/--mode/--dry-run/--enable-other/--config；运行后打印 token/缓存汇总） ├── write_attachments.py # 写入入口：分类结果落库到 Common.Attachment（全量/--limit/--id/--sn/--ids-file/--mode/--dry-run/--enable-other/--config；运行后打印 token/缓存汇总）
├── requirements.txt ├── requirements.txt
├── config_loader.py # YAML 配置读取与校验 ├── config_loader.py # YAML 配置读取与校验
├── db.py # SQL Server 查询与落库（fetch_params_by_ids / fetch_all_ids / upsert_attachments，pyodbc） ├── db.py # SQL Server 查询与落库（fetch_params_by_ids 支持按 id/sn 双键查询、fetch_all_ids / upsert_attachments，pyodbc）
├── prompts.py # 提示词与细分类目枚举（唯一需要改分类边界时编辑的文件） ├── prompts.py # 提示词与细分类目枚举（唯一需要改分类边界时编辑的文件）
├── llm_client.py # LLM 调用 (OpenAI 兼容接口) ├── llm_client.py # LLM 调用 (OpenAI 兼容接口)
├── parser.py # LLM 输出格式校验与清洗 → 结构化数据 ├── parser.py # LLM 输出格式校验与清洗 → 结构化数据

72
classifier.py
View File

@@ -15,9 +15,9 @@
日志模块的核心约定：不因为"这次调用失败了"就跳过记录，失败的调用恰恰最 日志模块的核心约定：不因为"这次调用失败了"就跳过记录，失败的调用恰恰最
需要被记下来供排查。 需要被记下来供排查。
批量处理(classify_batch)在此基础上先查数据库拿到 {总排号: 新参数} 映射， 批量处理(classify_batch)在此基础上先按 (标识符, 键类型) 查数据库，回取出
对查不到的总排号直接标记 status="not_found"，不发起LLM调用（也不生成日志， {标识符: {param, 总排号}} 映射；对查不到的标识符直接标记 status="not_found"，
因为根本没有"新参数"可供判断，没有对话可记）。 不发起LLM调用（也不生成日志，因为根本没有"新参数"可供判断，没有对话可记）。
enable_other 控制是否允许模型使用"其他"兜底类目（默认 False），从 enable_other 控制是否允许模型使用"其他"兜底类目（默认 False），从
classify_batch/classify_single 一路透传到 llm_client.classify_raw（决定 classify_batch/classify_single 一路透传到 llm_client.classify_raw（决定
@@ -174,7 +174,7 @@ def classify_param_text(
def classify_batch( def classify_batch(
db_cfg: dict[str, Any], db_cfg: dict[str, Any],
llm_cfg: dict[str, Any], llm_cfg: dict[str, Any],
zong_pai_hao_list: list[str], id_list: list[tuple[str, str]],
max_workers: int = 8, max_workers: int = 8,
mode: str = "coarse", mode: str = "coarse",
log_dir: str | Path | None = None, log_dir: str | Path | None = None,
@@ -182,43 +182,56 @@ def classify_batch(
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
"""批量分类入口：查库 -> 并发调用LLM -> 组装结果列表。 """批量分类入口：查库 -> 并发调用LLM -> 组装结果列表。
id_list: list[(标识符, 键类型)]，键类型取值：
"sn" -> 标识符为总排号，按总排号列查询；
"id" -> 标识符为数据库真实 ID，按 id_field 列查询并回取总排号。
两种键类型可在同一次调用中混用。
log_dir 若提供，会为每个实际发起LLM调用的总排号在该目录下生成一个日志 log_dir 若提供，会为每个实际发起LLM调用的总排号在该目录下生成一个日志
文件；查库失败(not_found)或参数为空(empty_param)的总排号不生成日志文件， 文件；查库失败(not_found)或参数为空(empty_param)的总排号不生成日志文件，
因为它们本就没有与LLM的对话内容可记。 因为它们本就没有与LLM的对话内容可记。
返回结果顺序与输入 zong_pai_hao_list 一致，即使某些查询失败或格式错误也会 返回结果顺序与输入 id_list 一致，即使某些查询失败或格式错误也会
补全为对应 status 的占位结果，保证"输入N个总排号，输出N条结果"。 补全为对应 status 的占位结果，保证"输入N条，输出N条"。结果中的
zong_pai_hao 一律为回查到的总排号（键类型为 id 时由数据库回取；键类型为
sn 时即输入本身；not_found 时回退为输入标识符以便追溯）。
""" """
# 去重但保留顺序，避免用户传入重复总排号导致重复查询/调用 # 去重但保留顺序，避免用户传入重复标识符导致重复查询/调用
seen: set[str] = set() seen: set[str] = set()
unique_ids: list[str] = [] unique_ids: list[tuple[str, str]] = []
for zph in zong_pai_hao_list: for item in id_list:
if zph not in seen: idt = item[0]
seen.add(zph) if idt not in seen:
unique_ids.append(zph) seen.add(idt)
unique_ids.append(item)
try: try:
param_map = fetch_params_by_ids(db_cfg, unique_ids) param_map = fetch_params_by_ids(db_cfg, unique_ids)
except DatabaseError as e: except DatabaseError as e:
logger.error("数据库查询失败，本批次全部标记为 db_error: %s", e) logger.error("数据库查询失败，本批次全部标记为 db_error: %s", e)
return [_empty_result(zph, "db_error") for zph in zong_pai_hao_list] return [_empty_result(idt, "db_error") for idt, _ in id_list]
llm_client = LLMClient(llm_cfg) llm_client = LLMClient(llm_cfg)
results_by_id: dict[str, dict[str, Any]] = {} results_by_id: dict[str, dict[str, Any]] = {}
ids_to_classify = [zph for zph in unique_ids if param_map.get(zph) is not None] # 待分类项：(标识符, 对应总排号)；not_found(未查到) 直接占位
for zph in unique_ids: ids_to_classify: list[tuple[str, str]] = []
if param_map.get(zph) is None: for idt, _ in unique_ids:
results_by_id[zph] = _empty_result(zph, "not_found") entry = param_map.get(idt)
if entry is not None and entry["param"] is not None:
ids_to_classify.append((idt, entry["sn"]))
else:
# 未查到：zong_pai_hao 回退为输入标识符（可能是 ID 或 SN）
results_by_id[idt] = _empty_result(idt, "not_found")
def _worker(zph: str) -> tuple[str, dict[str, Any]]: def _worker(idt: str, sn: str | None) -> tuple[str, dict[str, Any]]:
order_logger = OrderLogger(log_dir, zph, mode) if log_dir is not None else None order_logger = OrderLogger(log_dir, sn, mode) if log_dir is not None else None
status, has_attachment, types, meta = classify_param_text( status, has_attachment, types, meta = classify_param_text(
llm_client, param_map[zph], mode=mode, order_logger=order_logger, llm_client, param_map[idt]["param"], mode=mode, order_logger=order_logger,
enable_other=enable_other, enable_other=enable_other,
) )
return zph, { return idt, {
"zong_pai_hao": zph, "zong_pai_hao": sn,
"status": status, "status": status,
"has_attachment": has_attachment, "has_attachment": has_attachment,
"types": types, "types": types,
@@ -227,26 +240,27 @@ def classify_batch(
if ids_to_classify: if ids_to_classify:
with ThreadPoolExecutor(max_workers=max_workers) as pool: with ThreadPoolExecutor(max_workers=max_workers) as pool:
futures = [pool.submit(_worker, zph) for zph in ids_to_classify] futures = [pool.submit(_worker, idt, sn) for idt, sn in ids_to_classify]
for fut in as_completed(futures): for fut in as_completed(futures):
zph, result = fut.result() idt, result = fut.result()
results_by_id[zph] = result results_by_id[idt] = result
# 按原始输入顺序（含重复项）展开最终结果 # 按原始输入顺序（含重复项）展开最终结果
return [results_by_id[zph] for zph in zong_pai_hao_list] return [results_by_id[idt] for idt, _ in id_list]
def classify_single( def classify_single(
db_cfg: dict[str, Any], db_cfg: dict[str, Any],
llm_cfg: dict[str, Any], llm_cfg: dict[str, Any],
zong_pai_hao: str, identifier: str,
key: str = "sn",
mode: str = "coarse", mode: str = "coarse",
log_dir: str | Path | None = None, log_dir: str | Path | None = None,
enable_other: bool = False, enable_other: bool = False,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""单个总排号分类的便捷封装。""" """单个总排号/ID 分类的便捷封装。key 取值 "sn"(总排号) 或 "id"(真实ID)。"""
return classify_batch( return classify_batch(
db_cfg, llm_cfg, [zong_pai_hao], max_workers=1, mode=mode, log_dir=log_dir, db_cfg, llm_cfg, [(identifier, key)], max_workers=1, mode=mode, log_dir=log_dir,
enable_other=enable_other, enable_other=enable_other,
)[0] )[0]

117
db.py
View File

@@ -32,60 +32,101 @@ def _build_conn_str(db_cfg: dict[str, Any]) -> str:
def fetch_params_by_ids( def fetch_params_by_ids(
db_cfg: dict[str, Any], zong_pai_hao_list: list[str] db_cfg: dict[str, Any], id_keys: list[tuple[str, str]]
) -> dict[str, str | None]: ) -> dict[str, dict[str, Any | None]]:
"""按总排号批量查询新参数字段。 """按 (标识符, 键类型) 批量查询新参数字段。
返回 dict：{总排号: 新参数文本}。数据库中不存在的总排号，其值为 None id_keys: list[(identifier, key)]，key 取值：
（而不是直接从结果里省略该 key），方便调用方区分"没查到"和"查到但内容为空"。 "sn" -> 按总排号列（配置 id_column）查询，标识符即总排号；
"id" -> 按数据库真实 ID 列（配置 id_field）查询，并回取对应的总排号。
一个总排号只对应一条记录（业务已确认为一对一关系）；如果实际数据出现 返回 dict：{identifier: {"param": 新参数文本|None, "sn": 总排号|None}}。
重复总排号，取查询结果的第一条并记录一条 WARNING 日志，不中断整体流程。 标识符作为 key 原样保留（便于回查）；未查到的标识符其 param/sn 为 None，
与"查到但内容为空"区分开。sn 为对应的总排号（键类型为 sn 时即标识符本身，
键类型为 id 时由数据库回取）；若数据库未配置 id_field 且使用了 "id" 键，
则退化为按总排号列查询（仅向后兼容，会在日志告警）。
一对一关系：同一标识符出现多条记录时取第一条并记录 WARNING。
""" """
if not zong_pai_hao_list: if not id_keys:
return {} return {}
result: dict[str, str | None] = {zph: None for zph in zong_pai_hao_list} result: dict[str, dict[str, Any | None]] = {
idt: {"param": None, "sn": None} for idt, _ in id_keys
}
schema = db_cfg["schema"] schema = db_cfg["schema"]
table = db_cfg["table"] table = db_cfg["table"]
id_col = db_cfg["id_column"]
param_col = db_cfg["param_column"] param_col = db_cfg["param_column"]
sn_col = db_cfg["id_column"] # 总排号列
# 表名必须带 schema 前缀（如 [dbo].[表名]），只写表名在 schema 不是默认dbo时 id_field = db_cfg.get("id_field") # 真实 ID 列，可缺省
# 会查到错误的表，甚至直接报"找不到对象"。schema 和表名分别加中括号转义，
# 不能写成 [schema.table]，那样会被当成一个整体标识符解析。
qualified_table = f"[{schema}].[{table}]" qualified_table = f"[{schema}].[{table}]"
# 用参数化查询防止总排号里混入特殊字符导致 SQL 注入或语法错误 sn_items = [idt for idt, k in id_keys if k == "sn"]
placeholders = ",".join("?" for _ in zong_pai_hao_list) id_items = [idt for idt, k in id_keys if k == "id"]
sql = f"SELECT [{id_col}], [{param_col}] FROM {qualified_table} WHERE [{id_col}] IN ({placeholders})"
conn_str = _build_conn_str(db_cfg) def _fill(identifiers: list[str], where_col: str, is_id_key: bool) -> None:
try: if not identifiers:
with pyodbc.connect(conn_str, timeout=db_cfg["connect_timeout"]) as conn: return
cursor = conn.cursor() # 表名/schema 加中括号转义，不能写成 [schema.table]
# 查询超时设在 Connection 上（pyodbc 的 timeout 是 Connection 属性， placeholders = ",".join("?" for _ in identifiers)
# Cursor 没有该属性，设 cursor.timeout 会报 AttributeError）。 if is_id_key:
conn.timeout = db_cfg["query_timeout"] # 查真实 ID 列，同时回取总排号列(sn_col)作为 sn
cursor.execute(sql, zong_pai_hao_list) sql = (
seen = set() f"SELECT [{where_col}], [{sn_col}], [{param_col}] "
for row in cursor.fetchall(): f"FROM {qualified_table} WHERE [{where_col}] IN ({placeholders})"
zph, param = row[0], row[1] )
if zph in seen: else:
logger.warning("总排号 %s 存在重复记录，已取第一条", zph) sql = (
continue f"SELECT [{where_col}], [{param_col}] "
seen.add(zph) f"FROM {qualified_table} WHERE [{where_col}] IN ({placeholders})"
result[zph] = param )
except pyodbc.Error as e:
raise DatabaseError(f"数据库查询失败: {e}") from e conn_str = _build_conn_str(db_cfg)
try:
with pyodbc.connect(conn_str, timeout=db_cfg["connect_timeout"]) as conn:
conn.timeout = db_cfg["query_timeout"]
cursor = conn.cursor()
cursor.execute(sql, identifiers)
seen: set[str] = set()
for row in cursor.fetchall():
if is_id_key:
ident, sn_val, param = row[0], row[1], row[2]
else:
ident, param = row[0], row[1]
sn_val = ident # 总排号即标识符本身
ident = str(ident)
if ident in seen:
logger.warning("标识符 %s 在列 [%s] 上重复，已取第一条", ident, where_col)
continue
seen.add(ident)
entry = result.get(ident)
if entry is not None:
entry["param"] = param
entry["sn"] = sn_val
except pyodbc.Error as e:
raise DatabaseError(f"数据库查询失败: {e}") from e
# 总排号键：直接查 id_column
_fill(sn_items, sn_col, is_id_key=False)
# 真实 ID 键：查 id_field；未配置时降级为总排号列并告警
if id_items:
if id_field:
_fill(id_items, id_field, is_id_key=True)
else:
logger.warning(
"未配置 database.id_field，--id 将退化为按总排号列 [%s] 查询", sn_col
)
_fill(id_items, sn_col, is_id_key=False)
return result return result
def fetch_param_by_id(db_cfg: dict[str, Any], zong_pai_hao: str) -> str | None: def fetch_param_by_id(
"""单个总排号查询的便捷封装。""" db_cfg: dict[str, Any], identifier: str, key: str = "sn"
return fetch_params_by_ids(db_cfg, [zong_pai_hao]).get(zong_pai_hao) ) -> dict[str, Any | None] | None:
"""单个查询的便捷封装，返回 {param, sn} 或 None（未查到）。"""
return fetch_params_by_ids(db_cfg, [(identifier, key)]).get(identifier)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------

63
main.py
View File

@@ -3,15 +3,23 @@
"""布莱迪压力表 - 订单附件识别 命令行入口。 """布莱迪压力表 - 订单附件识别 命令行入口。
用法: 用法:
python main.py --id 26B742 # 单个总排号(粗分类,默认) python main.py --sn 26B742 # 单个总排号(粗分类,默认)
python main.py --id 26B742,26B743,26B744 # 批量，逗号分隔 python main.py --sn 26B742,26B743,26B744 # 批量，逗号分隔
python main.py --ids-file ids.txt # 批量，文件每行一个总排号 python main.py --id 802 # 单个数据库真实 ID
python main.py --id 26B742 --mode fine # 精分类(输出"大类:细分"格式) python main.py --id 802,803,804 # 批量真实 ID，逗号分隔
python main.py --id 26B742 --config other.yaml # 指定其他配置文件 python main.py --ids-file ids.txt # 批量，文件每行一个总排号(键类型 sn)
python main.py --id 26B742 --pretty # 格式化输出JSON(默认单行紧凑) python main.py --sn 26B742 --mode fine # 精分类(输出"大类:细分"格式)
python main.py --id 26B742 --log-dir /tmp/logs # 覆盖配置文件里的日志目录 python main.py --id 802 --config other.yaml # 指定其他配置文件
python main.py --id 26B742 --enable-other # 允许模型使用"其他"兜底类目 python main.py --sn 26B742 --pretty # 格式化输出JSON(默认单行紧凑)
python main.py --id 26B742 --summary # 额外在 stderr 打印本批运行汇总统计 python main.py --sn 26B742 --log-dir /tmp/logs # 覆盖配置文件里的日志目录
python main.py --sn 26B742 --enable-other # 允许模型使用"其他"兜底类目
python main.py --sn 26B742 --summary # 额外在 stderr 打印本批运行汇总统计
指定方式（两种键类型可混用）:
--id 数据库真实 ID 列(id_field)，如 802
--sn 总排号列(id_column)，如 26B742（即原先 --id 的语义）
--ids-file 每行一个总排号(键类型 sn)
无论用哪种方式指定，输出 JSON 中的 zong_pai_hao 一律为回查到的总排号。
分类模式(--mode): 分类模式(--mode):
coarse (默认) - 只判断大类: 资料/配件/耗材(启用 --enable-other 时还有"其他") coarse (默认) - 只判断大类: 资料/配件/耗材(启用 --enable-other 时还有"其他")
@@ -62,19 +70,27 @@ from classifier import classify_batch, summarize_results # noqa: E402
from config_loader import ConfigError, load_config # noqa: E402 from config_loader import ConfigError, load_config # noqa: E402
def _parse_ids(args: argparse.Namespace) -> list[str]: def _parse_ids(args: argparse.Namespace) -> list[tuple[str, str]]:
"""从 --id 或 --ids-file 解析出总排号列表，去除空白项。""" """从 --id / --sn / --ids-file 解析出 (标识符, 键类型) 列表。
ids: list[str] = []
--id -> 键类型 "id"（数据库真实 ID 列）
--sn -> 键类型 "sn"（总排号列，即原先 --id 的语义）
--ids-file -> 每行一个总排号，键类型 "sn"
三者可同时提供、合并后返回（顺序：--id, --sn, --ids-file）。
"""
items: list[tuple[str, str]] = []
if args.id: if args.id:
ids.extend(s.strip() for s in args.id.split(",") if s.strip()) 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: if args.ids_file:
p = Path(args.ids_file) p = Path(args.ids_file)
if not p.exists(): if not p.exists():
print(f"[错误] 总排号文件不存在: {p.resolve()}", file=sys.stderr) print(f"[错误] 总排号文件不存在: {p.resolve()}", file=sys.stderr)
sys.exit(1) sys.exit(1)
with p.open("r", encoding="utf-8") as f: with p.open("r", encoding="utf-8") as f:
ids.extend(line.strip() for line in f if line.strip()) items.extend((line.strip(), "sn") for line in f if line.strip())
return ids return items
def main() -> None: def main() -> None:
@@ -85,11 +101,16 @@ def main() -> None:
) )
parser.add_argument( parser.add_argument(
"--id", type=str, default=None, "--id", type=str, default=None,
help="总排号，单个或逗号分隔的多个，例如: 26B742 或 26B742,26B743", help="数据库真实 ID（如 802），单个或逗号分隔的多个，例如: 802 或 802,803",
)
parser.add_argument(
"--sn", type=str, default=None,
help="总排号（如 26B742），单个或逗号分隔的多个，例如: 26B742 或 26B742,26B743；"
"与 --id 同义但键类型不同（--sn 查总排号列，--id 查真实 ID 列）",
) )
parser.add_argument( parser.add_argument(
"--ids-file", type=str, default=None, "--ids-file", type=str, default=None,
help="包含总排号的文本文件路径，每行一个总排号", help="包含总排号的文本文件路径，每行一个总排号（键类型 sn）",
) )
parser.add_argument( parser.add_argument(
"--config", type=str, default="config.yaml", "--config", type=str, default="config.yaml",
@@ -119,12 +140,12 @@ def main() -> None:
) )
args = parser.parse_args() args = parser.parse_args()
if not args.id and not args.ids_file: if not args.id and not args.sn and not args.ids_file:
parser.error("必须提供 --id 或 --ids-file 其中之一") parser.error("必须提供 --id / --sn / --ids-file 其中之一")
ids = _parse_ids(args) ids = _parse_ids(args)
if not ids: if not ids:
print("[错误] 未解析到任何有效的总排号", file=sys.stderr) print("[错误] 未解析到任何有效的标识符（--id / --sn / --ids-file 均为空）", file=sys.stderr)
sys.exit(1) sys.exit(1)
try: try:
@@ -148,7 +169,7 @@ def main() -> None:
results = classify_batch( results = classify_batch(
db_cfg=cfg["database"], db_cfg=cfg["database"],
llm_cfg=cfg["llm"], llm_cfg=cfg["llm"],
zong_pai_hao_list=ids, id_list=ids,
max_workers=cfg["business"]["max_workers"], max_workers=cfg["business"]["max_workers"],
mode=mode, mode=mode,
log_dir=log_dir, log_dir=log_dir,

58
write_attachments.py
View File

@@ -13,11 +13,19 @@
python write_attachments.py # 全量写入 python write_attachments.py # 全量写入
python write_attachments.py --limit 10 # 仅前 10 个总排号（测试） python write_attachments.py --limit 10 # 仅前 10 个总排号（测试）
python write_attachments.py --ids-file ids.txt # 指定总排号（文件，每行一个） python write_attachments.py --ids-file ids.txt # 指定总排号（文件，每行一个）
python write_attachments.py --id 26B742 # 单个总排号写入 python write_attachments.py --sn 26B742 # 单个总排号写入（键类型 sn）
python write_attachments.py --id 26B742,26B743 # 多个总排号（逗号分隔）写入 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 --mode coarse # 粗分类写入
python write_attachments.py --dry-run # 只打印将写入的行，不落库 python write_attachments.py --dry-run # 只打印将写入的行，不落库
python write_attachments.py --config other.yaml 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.NS。
""" """
from __future__ import annotations from __future__ import annotations
@@ -75,23 +83,28 @@ def convert_results_to_rows(results: list[dict], mode: str) -> tuple[list[tuple[
return rows, skipped return rows, skipped
def _parse_ids(args: argparse.Namespace) -> list[str]: def _parse_ids(args: argparse.Namespace) -> list[tuple[str, str]]:
"""从 --id（逗号分隔，单个或多个）或 --ids-file 解析出总排号列表，去除空白项。 """从 --id / --sn / --ids-file 解析出 (标识符, 键类型) 列表。
两个来源可同时提供，合并后返回（顺序：先 --id，后 --ids-file）。 --id -> 键类型 "id"（数据库真实 ID 列）
--sn -> 键类型 "sn"（总排号列，即原先 --id 的语义）
--ids-file -> 每行一个总排号，键类型 "sn"
三者可同时提供、合并后返回（顺序：--id, --sn, --ids-file）。
--ids-file 文件不存在时直接报错退出。 --ids-file 文件不存在时直接报错退出。
""" """
ids: list[str] = [] items: list[tuple[str, str]] = []
if args.id: if args.id:
ids.extend(s.strip() for s in args.id.split(",") if s.strip()) 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: if args.ids_file:
p = Path(args.ids_file) p = Path(args.ids_file)
if not p.exists(): if not p.exists():
print(f"[错误] 总排号文件不存在: {p.resolve()}", file=sys.stderr) print(f"[错误] 总排号文件不存在: {p.resolve()}", file=sys.stderr)
sys.exit(1) sys.exit(1)
with p.open("r", encoding="utf-8") as f: with p.open("r", encoding="utf-8") as f:
ids.extend(line.strip() for line in f if line.strip()) items.extend((line.strip(), "sn") for line in f if line.strip())
return ids return items
def main() -> None: def main() -> None:
@@ -103,16 +116,21 @@ def main() -> None:
parser.add_argument("--config", type=str, default="config.yaml") parser.add_argument("--config", type=str, default="config.yaml")
parser.add_argument( parser.add_argument(
"--id", type=str, default=None, "--id", type=str, default=None,
help="总排号，单个或逗号分隔的多个，例如: 26B742 或 26B742,26B743；" help="数据库真实 ID（如 802），单个或逗号分隔的多个；提供后按真实 ID 列查询，"
"提供后直接按指定总排号写入，不再全表扫描（--limit 此时无效）", "不再全表扫描（--limit 此时无效）",
)
parser.add_argument(
"--sn", type=str, default=None,
help="总排号（如 26B742），单个或逗号分隔的多个（即原先 --id 的语义）；"
"提供后直接按总排号写入，不再全表扫描（--limit 此时无效）",
) )
parser.add_argument( parser.add_argument(
"--limit", type=int, default=None, "--limit", type=int, default=None,
help="仅处理前 N 个总排号（按总排号排序），测试用；与 --id/--ids-file 互斥", help="仅处理前 N 个总排号（按总排号排序），测试用；与 --id/--sn/--ids-file 互斥",
) )
parser.add_argument( parser.add_argument(
"--ids-file", type=str, default=None, "--ids-file", type=str, default=None,
help="包含总排号的文本文件路径，每行一个总排号；可单独使用或与 --id 合并", help="包含总排号的文本文件路径，每行一个总排号（键类型 sn）；可单独使用或与 --id/--sn 合并",
) )
parser.add_argument( parser.add_argument(
"--mode", type=str, default=None, choices=["coarse", "fine"], "--mode", type=str, default=None, choices=["coarse", "fine"],
@@ -147,14 +165,18 @@ def main() -> None:
) )
# 1) 取总排号列表 # 1) 取总排号列表
if args.id or args.ids_file: if args.id or args.sn or args.ids_file:
ids = _parse_ids(args) ids = _parse_ids(args)
if not ids: if not ids:
print("[错误] --id / --ids-file 未解析到任何有效的总排号", file=sys.stderr) print("[错误] --id / --sn / --ids-file 未解析到任何有效的标识符", file=sys.stderr)
sys.exit(1) sys.exit(1)
logger.info("从 --id / --ids-file 读取 %d 个总排号", len(ids)) 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: else:
ids = fetch_all_ids(cfg["database"], limit=args.limit) sn_list = fetch_all_ids(cfg["database"], limit=args.limit)
# 全表扫描得到的是总排号，键类型统一为 sn
ids = [(s, "sn") for s in sn_list]
logger.info( logger.info(
"源表读取 %d 个总排号%s", len(ids), "源表读取 %d 个总排号%s", len(ids),
f"（limit={args.limit}）" if args.limit else "", f"（limit={args.limit}）" if args.limit else "",
@@ -169,7 +191,7 @@ def main() -> None:
results = classify_batch( results = classify_batch(
db_cfg=cfg["database"], db_cfg=cfg["database"],
llm_cfg=cfg["llm"], llm_cfg=cfg["llm"],
zong_pai_hao_list=ids, id_list=ids,
max_workers=cfg["business"]["max_workers"], max_workers=cfg["business"]["max_workers"],
mode=mode, mode=mode,
log_dir=log_dir, log_dir=log_dir,