- 豆包: response_format.json_schema(strict);DeepSeek: 切 Responses API text.format.json_schema (官方 chat/completions 通道不支持 json_schema),不支持时自动降级 json_object([output] schema) - 重试两段式:传输错误 cfg.retries 次;解析失败/空正文有独立 3 次专用重试,仍失败落 parse_fail - 移除三票复核(decide_final/_verify_clean/verify_clean),DeepSeek 单次判定即终稿 - 新增 [run] deepseek_recheck 开关(默认 no=仅豆包初筛;yes=追加 DeepSeek 复检) - GUI 不再覆盖 config 提示词;prompt 相对/前导斜杠路径按 exe 目录解析 - 报表去「票型/复核」列并同步说明;README/config.example.ini 同步 - 测试新增输出格式、重试、开关与提示词路径用例(50 passed)
48 lines
1.6 KiB
Python
48 lines
1.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""pipeline 模块测试:排序、成本计算、缓存路径。"""
|
|
from violation_detector.pipeline import (
|
|
_cache_path, _cat_num, deepseek_cost, natural_key,
|
|
)
|
|
|
|
|
|
def R(cat):
|
|
return {"category": cat, "attr": "a", "logic": "l", "file": "x.jpg",
|
|
"usage": {"prompt": 2199, "completion": 2282, "cached": 1792}}
|
|
|
|
|
|
def test_natural_key_sort():
|
|
names = ["10_a.jpg", "2_a.jpg", "1_a.jpg", "21_a.jpg", "3_a.jpg"]
|
|
assert sorted(names, key=natural_key) == ["1_a.jpg", "2_a.jpg", "3_a.jpg",
|
|
"10_a.jpg", "21_a.jpg"]
|
|
|
|
|
|
def test_cat_num():
|
|
assert _cat_num("12. 侵权 - 其他") == 12
|
|
assert _cat_num("1. 烟草") == 1
|
|
# 无编号的结论统一排到最后(无违规/违规不明不变动分类)
|
|
assert _cat_num("无违规") == 99
|
|
assert _cat_num("违规不明") == 99
|
|
|
|
|
|
def test_deepseek_cost_known_usage():
|
|
# miss 407 + cached 1792 + completion 2282
|
|
results = {"a": R("无违规")}
|
|
peak, idle, comp = deepseek_cost(results)
|
|
expect_peak = (407 * 3.0 + 1792 * 0.10 + 2282 * 9.0) / 1e6
|
|
expect_idle = (407 * 1.5 + 1792 * 0.05 + 2282 * 4.5) / 1e6
|
|
assert abs(peak - expect_peak) < 1e-9
|
|
assert abs(idle - expect_idle) < 1e-9
|
|
assert comp == 2282
|
|
|
|
|
|
def test_deepseek_cost_empty():
|
|
assert deepseek_cost({}) == (0.0, 0.0, 0.0)
|
|
|
|
|
|
def test_cache_path_sanitizes_folder(tmp_path):
|
|
p = _cache_path("ark", str(tmp_path / "we ird\\name+:x"))
|
|
assert p.parent.exists()
|
|
assert p.name.startswith("ark_")
|
|
for ch in ' :+':
|
|
assert ch not in p.name
|