feat: trace 贯通与脱敏工具 + risk 阈值配置(A1)

app/config/settings.py

@@ -1,5 +1,7 @@
 """环境配置（从 .env 读取，见 .env.example）。"""
 
+from decimal import Decimal
+
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -29,5 +31,17 @@ class Settings(BaseSettings):
     deepseek_api_key: str = ""
     deepseek_base_url: str = "https://api.deepseek.com"
 
+    # ===== Risk 阈值（默认值=冻结规则 · docs/PRD/附-风控规则表.md）=====
+    risk_assessment_valid_days: int = 365
+    risk_large_amount: Decimal = Decimal("500000")
+    risk_daily_total: Decimal = Decimal("500000")
+    risk_freq_count: int = 3
+    risk_probe_window_minutes: int = 5
+    risk_probe_count: int = 3
+    risk_probe_amount: Decimal = Decimal("400000")
+    risk_small_amount: Decimal = Decimal("10000")
+    risk_small_count: int = 3
+    risk_aml_default_threshold: Decimal = Decimal("0.85")
+
 
 settings = Settings()

app/utils/desensitize.py

@@ -0,0 +1,43 @@
+"""数据脱敏纯函数（DESENS-001~004 · docs/PRD/附-风控规则表.md §4）。
+
+原则：API 出参 / 预警 payload / LLM 上下文传入前必须已脱敏；
+customer_id 为系统内部代理键，不脱敏（DESENS 规则明确）；
+DESENS-005（资产金额展示）归前端组件，后端不实现。
+"""
+
+from __future__ import annotations
+
+
+def mask_id_card(value: str | None) -> str:
+    """DESENS-001 身份证号：保留前 3 后 4，中间 ****。"""
+    return _mask_middle(value, head=3, tail=4)
+
+
+def mask_phone(value: str | None) -> str:
+    """DESENS-002 手机号：保留前 3 后 4，中间 ****。"""
+    return _mask_middle(value, head=3, tail=4)
+
+
+def mask_name(value: str | None) -> str:
+    """DESENS-003 姓名：保留姓氏，其余 **。"""
+    if not value:
+        return ""
+    return value[0] + "**"
+
+
+def mask_bank_card(value: str | None) -> str:
+    """DESENS-004 银行卡号：保留后 4 位，其余 ****。"""
+    if not value:
+        return ""
+    if len(value) <= 4:
+        return "*" * len(value)
+    return "*" * (len(value) - 4) + value[-4:]
+
+
+def _mask_middle(value: str | None, head: int, tail: int) -> str:
+    """通用中段掩码；长度不足 head+tail 时整体掩码（宁多勿漏）。"""
+    if not value:
+        return ""
+    if len(value) <= head + tail:
+        return "*" * len(value)
+    return value[:head] + "****" + value[-tail:]

app/utils/trace.py

@@ -0,0 +1,26 @@
+"""trace_id 全链路贯通（contextvars · 架构 §5.1）。
+
+约束（编码前必读）：
+- 中间件必须在 call_next 之前 set()，禁止在 endpoint/service 内重新 set()；
+- 后台任务（create_task）须显式 contextvars.copy_context()；
+- 同步 def 路由跑线程池时 context 经 anyio 传播，由 B8 trace 一致性断言实测兜底。
+"""
+
+from __future__ import annotations
+
+from contextvars import ContextVar
+from uuid import uuid4
+
+_trace_id: ContextVar[str] = ContextVar("trace_id", default="")
+
+
+def new_trace(trace_id: str | None = None) -> str:
+    """生成并绑定新 trace_id；透传外部 X-Trace-Id 时传入该值。"""
+    tid = trace_id or f"trc-{uuid4().hex[:16]}"
+    _trace_id.set(tid)
+    return tid
+
+
+def current_trace() -> str:
+    """读取当前 trace_id；未初始化时返回空串（调用方应兜底生成）。"""
+    return _trace_id.get()

requirements.txt

@@ -21,3 +21,6 @@ python-jose[cryptography]>=3.3.0
 
 # Utils
 python-multipart>=0.0.18
+
+# Testing
+pytest>=8.3.0

tests/test_desensitize.py

@@ -0,0 +1,61 @@
+"""desensitize.py 单测（A1 · DESENS-001~004，005 归前端）。"""
+
+from app.utils.desensitize import mask_bank_card, mask_id_card, mask_name, mask_phone
+
+
+class TestMaskIdCard:
+    def test_normal_18(self):
+        assert mask_id_card("110101199003071234") == "110****1234"
+
+    def test_short_overmasked(self):
+        # 长度不足 head+tail：整体掩码（宁多勿漏）
+        assert mask_id_card("123456") == "******"
+
+    def test_none_and_empty(self):
+        assert mask_id_card(None) == ""
+        assert mask_id_card("") == ""
+
+
+class TestMaskPhone:
+    def test_normal_11(self):
+        assert mask_phone("13812349527") == "138****9527"
+
+    def test_boundary_7(self):
+        # 恰好 head+tail 长度 → 整体掩码
+        assert mask_phone("1234567") == "*******"
+
+    def test_boundary_8(self):
+        # 8 = 3+4+1，可保留头尾
+        assert mask_phone("12345678") == "123****5678"
+
+
+class TestMaskName:
+    def test_keeps_surname(self):
+        assert mask_name("张三丰") == "张**"
+        assert mask_name("李四") == "李**"
+
+    def test_single_char(self):
+        # 单字名：姓氏 + **（规则如此，不特判）
+        assert mask_name("王") == "王**"
+
+    def test_none_and_empty(self):
+        assert mask_name(None) == ""
+        assert mask_name("") == ""
+
+
+class TestMaskBankCard:
+    def test_normal_16(self):
+        assert mask_bank_card("6222021234561234") == "************1234"
+
+    def test_keep_last4_only(self):
+        masked = mask_bank_card("6222021234561234")
+        assert masked[-4:] == "1234"
+        assert set(masked[:-4]) == {"*"}
+
+
+    def test_short_overmasked(self):
+        assert mask_bank_card("123") == "***"
+
+    def test_none_and_empty(self):
+        assert mask_bank_card(None) == ""
+        assert mask_bank_card("") == ""

tests/test_settings.py

@@ -0,0 +1,25 @@
+"""settings risk_* 配置单测（A1）：默认值=冻结规则。"""
+
+from decimal import Decimal
+
+from app.config.settings import Settings
+
+
+def test_risk_defaults_match_frozen_rules():
+    s = Settings(_env_file=None)  # 不读 .env，纯默认值
+    assert s.risk_assessment_valid_days == 365
+    assert s.risk_large_amount == Decimal("500000")
+    assert s.risk_daily_total == Decimal("500000")
+    assert s.risk_freq_count == 3
+    assert s.risk_probe_window_minutes == 5
+    assert s.risk_probe_count == 3
+    assert s.risk_probe_amount == Decimal("400000")
+    assert s.risk_small_amount == Decimal("10000")
+    assert s.risk_small_count == 3
+    assert s.risk_aml_default_threshold == Decimal("0.85")
+
+
+def test_amounts_are_decimal():
+    s = Settings(_env_file=None)
+    assert isinstance(s.risk_large_amount, Decimal)
+    assert isinstance(s.risk_aml_default_threshold, Decimal)

tests/test_trace.py

@@ -0,0 +1,21 @@
+"""trace.py 单测（A1）。"""
+
+from app.utils.trace import current_trace, new_trace
+
+
+def test_new_trace_generates_and_binds():
+    tid = new_trace()
+    assert tid.startswith("trc-")
+    assert len(tid) == 20  # trc- + 16 hex
+    assert current_trace() == tid
+
+
+def test_new_trace_accepts_external_id():
+    tid = new_trace("trc-external-0001")
+    assert current_trace() == "trc-external-0001"
+    assert tid == "trc-external-0001"
+
+
+def test_uniqueness():
+    ids = {new_trace() for _ in range(100)}
+    assert len(ids) == 100