"""函数复杂度棘轮：函数变复杂必须有人判断过，不许「顺手加一个 if」悄悄堆上去。 2026-09-13 全仓盘点，`src/peach/` 与 `scripts/` 里分支数达到 30 的函数有 42 个， 最大的两个（`library_processing.process_library`、`web_batch.q_ads`）分别是 91 与 85。 它们不是一次写成的，是每次「顺手加一个 if」堆出来的；这个门槛拦的就是那一个 if。 计数是纯 AST 分支数：每个 `if`／`elif`／`for`／`while`／`except`／`with`／`assert`、 每个推导式子句、每个 `and`／`or`、每个 `case` 与条件表达式各计 1，嵌套函数算进 外层。数值只和它自己比，不和别的工具比。 两张表，数都必须与代码一致： - `BASELINE` 是盘点留下的存量，只能降。函数变简单了就把数改小，降到限值以下就删掉。 - `ACCEPTED` 是判过「就该这么长」的函数，每条带一句理由：顺序流水线、按类型分派、 逐字段解析这类，拆开只会把一路传下去的状态散进几个只调用一次的函数里。它长了就在 同一次改动里改数，理由跟着再看一遍还成不成立。 函数超过自己的数时先判断，再二选一。一是拆：拆出去的那块要能独立命名，有自己的前提 和结果，比如 `jav_cover_fetch._download_best`「按档次完整下载，挑第一张静态图」。二是 挪进 `ACCEPTED` 写理由。下面几种不算拆，只是把分支数藏起来，评审一律退回：只转发一个 表达式的包装；把模块函数当参数传给 helper；为了少一个 if 建只有一项的分派表；在注释里 拿这道门槛当设计理由。 """ import ast import pathlib import unittest from types import MappingProxyType import peach REPO = pathlib.Path(peach.__file__).resolve().parents[2] SOURCE_ROOTS = (REPO / "src" / "peach", REPO / "scripts") #: 达到这个分支数的函数进入基线；基线之外的函数不许长到这里。 LIMIT = 30 #: 键是 `<仓库相对路径>:<限定名>`，值是当前的分支数，只能往下改。 BASELINE: dict[str, int] = { "src/peach/junk.py:_scored_candidates": 53, "scripts/localize_performer_names.py:collect": 63, "src/peach/web_resource_sync.py:_resource_orphan_plan": 50, "src/peach/web_review.py:_review_rows": 38, "src/peach/fanbox.py:normalize_fanbox_post": 59, "scripts/merge_duplicate_identities.py:collect": 50, "src/peach/web_review.py:_attach_review_asset_context": 46, "src/peach/web_review.py:w_review_decision": 41, "scripts/harvest_social_avatars.py:run": 45, "scripts/fetch_studio_avatar_candidates.py:main": 44, "src/peach/web_catalog.py:q_item": 44, "src/peach/web_follow.py:_follow_author_rows": 35, "src/peach/taste_history.py:_taste_analysis": 40, "src/peach/web_catalog.py:catalog_filter": 40, "scripts/audit_domain_codes.py:collect": 39, "scripts/harvest_social_avatars.py:harvest_entity": 37, "scripts/merge_duplicate_identities.py:collect_repeated_projections": 34, "src/peach/fc2_similarity.py:media_evidence": 34, "scripts/clean_names.py:run": 33, "src/peach/studio_icons.py:icon_row": 32, "scripts/find_ads.py:find_candidates": 31, "scripts/rehome_unknown_jav.py:build_plan": 31, "src/peach/resource_identification.py:ingest_results": 30, } #: 判过就该这么长的函数：`(分支数, 为什么拆开更差)`。键与 `BASELINE` 同形、不重叠。 ACCEPTED: dict[str, tuple[int, str]] = { "src/peach/api.py:create_app": ( 40, "应用工厂：按启动顺序构造数据库、各项后台服务（含可选分享）与路由并接成一个 app，每个可注入的" "依赖各带一个缺省；挂载探测按配置状态启动。拆开会把同一条构造顺序分散到几处"), "src/peach/library_processing.py:process_library": ( 48, "顺序流水线：垃圾检查、媒体属性、本地资料、本地预览、联网采集、写候选、收尾共用一份进度状态" "与问题记录；来源扫描、资产查询与外部会话分别归 `_scan_sources`、" "`_collection_rows`、`_RemoteSession`，剩下的是任务编排、逐项目资料处理与指定范围的存量重新核查"), "src/peach/jav_cover_fetch.py:run": ( 50, "批处理的主循环：每个番号依次量本机尺寸、择优、比较、落盘、记日志，统计、日志行" "和出错后换新的连接池在整轮里共用，逐段拆开只是把这几样来回传"), "scripts/probe.py:run": ( 41, "媒体批处理：选择允许读取的资产、检查垃圾、并行探测、串行提交与重判共用日志、" "停止标志和磁盘闸门；提交后的资产列表必须跨缓冲批次保留，拆散会割裂中止时的收尾"), "scripts/sheets.py:run": ( 43, "媒体批处理：按来源选择资产并检查垃圾、并行抽帧、串行登记共用日志、计数和停止标志；" "已有产物复用与磁盘闸门属于同一工作循环，拆开会把退出与失败统计分散到多处"), } #: 一句理由至少要说清「是哪种形状、拆开差在哪」，短于这个长度的多半只是占位。 MIN_REASON = 20 BRANCHES = (ast.If, ast.For, ast.AsyncFor, ast.While, ast.ExceptHandler, ast.With, ast.AsyncWith, ast.Assert, ast.IfExp, ast.match_case) def complexity(node: ast.AST) -> int: """`node` 的分支数加一：一条直线的函数是 1。""" total = 1 for child in ast.walk(node): if isinstance(child, BRANCHES): total += 1 elif isinstance(child, ast.comprehension): total += 1 + len(child.ifs) elif isinstance(child, ast.BoolOp): total += len(child.values) - 1 return total def functions(path: pathlib.Path): """`path` 里每个函数与方法的 `(限定名, 分支数)`；嵌套函数并进外层，不单列。""" tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) def walk(body, prefix): for node in body: if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): yield prefix + node.name, complexity(node) elif isinstance(node, ast.ClassDef): yield from walk(node.body, prefix + node.name + ".") yield from walk(tree.body, "") def survey() -> dict[str, int]: """当前树里每个函数的分支数，键与 `BASELINE` 同形。""" found = {} for root in SOURCE_ROOTS: for path in sorted(root.rglob("*.py")): relative = path.relative_to(REPO).as_posix() for name, value in functions(path): found[f"{relative}:{name}"] = value return found def recorded() -> dict[str, int]: """两张表里记下的数合在一起。""" return {**BASELINE, **{key: value for key, (value, _reason) in ACCEPTED.items()}} class ComplexityRatchetTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.current = MappingProxyType(survey()) def test_the_counter_counts_the_shapes_it_names(self): source = ( "def f(items):\n" " for item in items:\n" " if item and item.ok or item.forced:\n" " try:\n" " pass\n" " except ValueError:\n" " pass\n" " return [x for x in items if x]\n" ) node = ast.parse(source).body[0] # 1 + for + if + and + or + except + 推导式 + 推导式的 if self.assertEqual(complexity(node), 8) def test_no_function_grew_past_its_recorded_count(self): current, limits = self.current, recorded() grown = sorted( f"{key}: {value}（记录 {limits.get(key, LIMIT - 1)}）" for key, value in current.items() if value > limits.get(key, LIMIT - 1)) self.assertEqual(grown, [], "这些函数的分支数超过了记录。拆出能独立命名的一块，或判过它就该" "这么长后挪进 ACCEPTED 写理由；转发包装、把函数当参数传、单项分派表" "都不算拆：\n " + "\n ".join(grown)) def test_the_recorded_counts_match_the_code(self): current = self.current stale = sorted( f"{key}: 现在 {current.get(key, '不存在')}，记录 {value}" for key, value in recorded().items() if key not in current or current[key] < value) self.assertEqual(stale, [], "记录落后于代码：函数变简单或删掉了就把这里改小或删掉，" "记录不能替将来的增长留余量：\n " + "\n ".join(stale)) def test_the_tables_hold_only_functions_at_or_over_the_limit(self): below = sorted(key for key, value in recorded().items() if value < LIMIT) self.assertEqual(below, [], f"低于 {LIMIT} 的函数不该进表：{below}") def test_an_accepted_function_says_why_splitting_is_worse(self): self.assertEqual(sorted(set(BASELINE) & set(ACCEPTED)), [], "一个函数只能在一张表里") thin = sorted(key for key, (_value, reason) in ACCEPTED.items() if len(reason.strip()) < MIN_REASON) self.assertEqual(thin, [], f"这些条目的理由不足 {MIN_REASON} 字：{thin}") if __name__ == "__main__": unittest.main()