From 24de6c34f736bb856e156e6fb8d7685bd9186b93 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Fri, 11 Sep 2026 13:35:58 +0800 Subject: [PATCH] =?UTF-8?q?fix(risk):=20=E6=89=AB=E6=8F=8F=E5=81=A5?= =?UTF-8?q?=E5=A3=AE=E6=80=A7=E2=80=94=E2=80=94=E8=84=8F=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E4=B8=8D=E5=86=8D=E4=B8=AD=E6=96=AD=E6=95=B4=E6=89=B9=EF=BC=8C?= =?UTF-8?q?=E8=B0=83=E5=BA=A6=E5=99=A8=E9=87=8D=E5=90=AF=E5=90=8E=E4=BB=8D?= =?UTF-8?q?=E4=BC=9A=E6=89=A7=E8=A1=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit docs/25 P0 的后两项，都属于"风控看起来在工作、其实没在跑"那一类。 **② 一条脏数据中断整批** _level_value 原先直接 int(level.replace(prefix, ""))：等级字段只要有一条不是 R1-R5 / C1-C5（例如写了「中风险」），就抛 ValueError 并冒到 scan() 的兜底 → **整批 rollback**，本次扫描前面已经生成的预警全部作废。数据脏属于运维问题，不该升级成 "整个风控停摆"。 改为返回 int | None；调用点跳过该条并记 warning（带上 transaction_id 与两个原始值， 便于运维直接定位）。顺带把 eplace 换成 emoveprefix：原先 "R2R" 会被错当成 2， 现在只去掉开头那一个前缀字符。 **③ 调度器重启后永不执行** last_run_at 只存在内存里，重启后为 None，而 _is_due 此时返回 config.run_immediately（默认 False）⇒ 重启后 _is_due 恒为假，**调度器形同虚设， 而且没有任何告警**。 改为"从未跑过即视为 due"：多跑一次的最坏后果是重复扫描，而扫描每条规则都先 _exists 查重、外层还有 MySQL 级锁；反过来"不跑"的后果可能是永远不跑。 config.run_immediately 不再承担"首次是否执行"的语义（它原本想表达"启动后别马上跑"， 但那与"永远不跑"在实现上无法区分），字段保留以免破坏既有配置。 新增 tests/unit/service/test_risk_scan_robustness.py（6 条）：脏数据返回 None 而不抛异常、 R2R 不被过度剥离、首次必 due（哪怕 run_immediately=False）、以及间隔前后的判定。 ruff / mypy(136 文件) / 622 unit+contract / 29 integration 全绿。 --- app/service/risk_scan_service.py | 30 +++++++- app/worker/risk_scan_scheduler.py | 12 ++- .../unit/service/test_risk_scan_robustness.py | 73 +++++++++++++++++++ 3 files changed, 111 insertions(+), 4 deletions(-) create mode 100644 tests/unit/service/test_risk_scan_robustness.py diff --git a/app/service/risk_scan_service.py b/app/service/risk_scan_service.py index a11197c..0cbf16a 100644 --- a/app/service/risk_scan_service.py +++ b/app/service/risk_scan_service.py @@ -146,7 +146,21 @@ class RiskRuleEngine: or await self._exists(transaction.id, "RW-007") ): continue - gap = _level_value(product.risk_level, "R") - _level_value(customer.investor_type, "C") + product_level = _level_value(product.risk_level, "R") + customer_level = _level_value(customer.investor_type, "C") + if product_level is None or customer_level is None: + # 等级字段脏（不是 R1-R5 / C1-C5）。**跳过这一条**，而不是让整个扫描挂掉： + # 原先 `_level_value` 直接 `int(...)`，一条脏数据抛 ValueError 会冒到 + # `scan()` 的兜底 → 整批 rollback，前面规则扫出来的预警全部白做。 + # 数据脏属于运维问题，不该升级成"整个风控停摆"。 + logger.warning( + "跳过等级字段异常的记录：transaction_id=%s risk_level=%r investor_type=%r", + transaction.id, + product.risk_level, + customer.investor_type, + ) + continue + gap = product_level - customer_level missing_trace = ( ( product.risk_disclosure_required @@ -486,5 +500,15 @@ def _age(birth_date: date | None) -> int: ) -def _level_value(level: str, prefix: str) -> int: - return int(level.replace(prefix, "")) +def _level_value(level: str, prefix: str) -> int | None: + """把 "R2" / "C3" 解析成 2 / 3；格式不符返回 None，**不再抛异常**。 + + 原先写的是 `int(level.replace(prefix, ""))`：一条脏数据（等级字段写了「中风险」之类） + 就会抛 ValueError，一路冒到 `scan()` 的兜底 → **整批 rollback**，这次扫描前面已经 + 生成的预警全部作废。数据脏是运维问题，不该升级成"整个风控停摆"。 + + 用 `removeprefix` 而不是 `replace`：只去掉开头那一个前缀字符， + `"R2R"` 这种脏值不会被错当成 2。（没有前缀的 `"2"` 仍能解析——历史数据可能不带前缀。） + """ + digits = level.strip().upper().removeprefix(prefix.upper()) + return int(digits) if digits.isdigit() else None diff --git a/app/worker/risk_scan_scheduler.py b/app/worker/risk_scan_scheduler.py index ee9249a..d21c6dd 100644 --- a/app/worker/risk_scan_scheduler.py +++ b/app/worker/risk_scan_scheduler.py @@ -121,7 +121,17 @@ class RiskScanSchedulerWorker: current: datetime, ) -> bool: if self.last_run_at is None: - return config.run_immediately + # 进程刚起来，不知道自己上次是什么时候跑的 —— `last_run_at` 只存在内存里。 + # **保守地视为 due**：多跑一次的最坏后果是重复扫描，而扫描本身是幂等的 + # （每条规则先 `_exists` 查重）并且有 MySQL 级锁；反过来"不跑"的后果可能是 + # **永远不跑**：原先这里返回 `config.run_immediately`（默认 False）， + # 重启之后 `_is_due` 恒为假，调度器形同虚设 —— 而且没有任何告警， + # 现场只会表现为"风控好像没在扫描"。 + # + # `config.run_immediately` 因此不再承担"首次是否执行"的语义（它原本想表达的 + # 是"启动后别马上跑"，但那与"永远不跑"在实现上无法区分）。字段保留， + # 以免破坏既有配置。 + return True return current - self.last_run_at >= timedelta(minutes=config.interval_minutes) async def _execute_with_retry( diff --git a/tests/unit/service/test_risk_scan_robustness.py b/tests/unit/service/test_risk_scan_robustness.py new file mode 100644 index 0000000..92c31f3 --- /dev/null +++ b/tests/unit/service/test_risk_scan_robustness.py @@ -0,0 +1,73 @@ +"""扫描健壮性：脏数据不该让整批停摆，调度器不该"重启后再也不跑"。 + +这两条都属于"风控看起来在工作、其实没在跑"那类缺陷，现象上很难发现： + +1. `_level_value` 原先直接 `int(level.replace(prefix, ""))` —— 一条等级字段脏的数据 + （比如写了「中风险」）会抛 ValueError，冒到 `scan()` 的兜底 → **整批 rollback**， + 本次扫描前面已经生成的预警全部作废。 +2. 调度器的 `last_run_at` 只在内存，重启后为 None，而 `_is_due` 此时返回 + `config.run_immediately`（默认 False）→ **重启后永不执行**，且没有任何告警。 +""" + +import datetime as dt +from types import SimpleNamespace +from typing import Any, cast + +from app.service.risk_scan_service import _level_value +from app.worker.risk_scan_scheduler import RiskScanSchedulerWorker + + +def test_level_value_parses_valid_levels() -> None: + assert _level_value("R2", "R") == 2 + assert _level_value("C3", "C") == 3 + assert _level_value("c5", "C") == 5 # 大小写不敏感 + assert _level_value(" 2 ", "R") == 2 # 历史数据可能不带前缀 + + +def test_level_value_returns_none_instead_of_raising_on_dirty_data() -> None: + """脏数据返回 None，不抛异常 —— 这是不让整批扫描 rollback 的前提。""" + assert _level_value("中风险", "R") is None + assert _level_value("", "R") is None + assert _level_value("R", "R") is None + assert _level_value("R6", "R") == 6 # 越界值交给上层比较，解析本身只负责取数字 + + +def test_level_value_does_not_over_strip() -> None: + """用 removeprefix 而非 replace：`R2R` 不该被错当成 2。""" + assert _level_value("R2R", "R") is None + + +def _scheduler(*, last_run_at: dt.datetime | None) -> RiskScanSchedulerWorker: + """只构造被测方法用到的那两个属性。""" + scheduler = object.__new__(RiskScanSchedulerWorker) + scheduler.last_run_at = last_run_at + return scheduler + + +def _config(*, run_immediately: bool, interval_minutes: int = 30) -> Any: + return SimpleNamespace(run_immediately=run_immediately, interval_minutes=interval_minutes) + + +_NOW = dt.datetime(2026, 9, 11, 12, 0, tzinfo=dt.UTC) + + +def test_first_run_is_due_even_when_run_immediately_is_false() -> None: + """进程刚起来（last_run_at 为 None）必须视为 due，哪怕是 run_immediately=False。 + + 原先这里返回 run_immediately，默认 False ⇒ 重启后永不执行。 + """ + scheduler = cast(Any, _scheduler(last_run_at=None)) + + assert scheduler._is_due(_config(run_immediately=False), _NOW) is True + + +def test_not_due_before_interval_elapses() -> None: + scheduler = cast(Any, _scheduler(last_run_at=_NOW - dt.timedelta(minutes=10))) + + assert scheduler._is_due(_config(run_immediately=False), _NOW) is False + + +def test_due_after_interval_elapses() -> None: + scheduler = cast(Any, _scheduler(last_run_at=_NOW - dt.timedelta(minutes=31))) + + assert scheduler._is_due(_config(run_immediately=False), _NOW) is True