"""T-8 规则引擎改造单测（开发计划 §7.1 DoD / 验收 5·6·7）。

覆盖三层：
1. `rules.amount_view` 纯函数 —— 同组只留转出端 / 无 gid 恒等 / 组内无 redeem 防御 / 顺序保持
2. `run_rules` 视图分流 —— RISK-002 与 RISK-005 走金额视图（**不翻倍**），
   RISK-001 / RISK-003 仍看**全量**（证明去重未删行，验收 6）
3. `engine.process_convert_event` —— **一张单 + `payload.events` 两条**（验收 7）、
   两条流水只跑一次、D19 `on_error_hook` 容错（hook 自身抛异常不得反噬主流程）

sqlite 内存库（`conftest.sqlite_engine`），数据自建。阈值一律**显式传入**，
不读 settings（避免与 conftest 的 autouse 隔离相互干扰，也让去重差异可精确归因）。
"""

from __future__ import annotations

import ast
from datetime import datetime, timedelta
from decimal import Decimal
from pathlib import Path

import pytest
from sqlalchemy import text

from app.repository.core_ro import CoreReadOnlyRepository
from app.repository.risk_repository import RiskRepository
from app.service.risk.engine import process_convert_event, process_trade_event
from app.service.risk.rules import RiskThresholds, amount_view, run_rules

CUST = "CUST-T8"
PROD_A = "PROD-T8A"  # 转出方
PROD_B = "PROD-T8B"  # 转入方
GID = "CNV-T8-0001"
NOW = datetime(2026, 9, 4, 10, 0, 0)


def _th(**over) -> RiskThresholds:
    """显式阈值（不让用例依赖 settings；concentration 推到不可达避免 RISK-006 干扰）。"""
    base = dict(
        large_amount=Decimal("200000"),
        daily_total=Decimal("500000"),
        freq_count=3,
        probe_window_minutes=5,
        probe_count=3,
        probe_amount=Decimal("400000"),
        small_amount=Decimal("10000"),
        small_count=3,
        concentration_threshold=1.01,
    )
    base.update(over)
    return RiskThresholds(**base)


def _trade(
    trade_id: str,
    trade_type: str,
    amount: str,
    at: datetime = NOW,
    *,
    gid: str | None = None,
    product: str = PROD_A,
) -> dict:
    """构造一条规则引擎口径的流水（等价 core_trade 行）。"""
    return {
        "trade_id": trade_id,
        "customer_id": CUST,
        "product_id": product,
        "trade_type": trade_type,
        "amount": Decimal(amount),
        "qty": Decimal("1"),
        "trade_status": "confirmed",
        "traded_at": at,
        "convert_group_id": gid,
    }


def _convert_legs(out_amount: str = "300000", in_amount: str = "299000") -> tuple[dict, dict]:
    """一次转换的两条流水（同组，转出在前）。"""
    out = _trade("TRD-T8-OUT", "redeem", out_amount, gid=GID, product=PROD_A)
    inn = _trade(
        "TRD-T8-IN", "subscribe", in_amount, NOW + timedelta(seconds=1), gid=GID, product=PROD_B
    )
    return out, inn


def _ids(hits) -> set[str]:
    return {h.rule_id for h in hits}


# ── 1. `amount_view` 纯函数 ─────────────────────────────────────────
def test_amount_view_identity_without_group():
    """无 convert_group_id 的交易**原样通过**（非 convert 场景恒等 → 既有断言零影响）。"""
    raw = [
        _trade("T-1", "redeem", "1000"),
        _trade("T-2", "subscribe", "900", NOW + timedelta(seconds=1), product=PROD_B),
    ]
    view = amount_view(raw)
    assert view == raw, "无 gid 输入必须恒等（内容与顺序都不变）"
    assert view is not raw, "返回独立列表，不得共享可变状态"


def test_amount_view_keeps_redeem_when_subscribe_comes_first():
    out, inn = _convert_legs()
    view = amount_view([inn, out])
    assert [t["trade_id"] for t in view] == ["TRD-T8-OUT"], "同组只留转出端，且位置不变"


def test_amount_view_keeps_redeem_when_redeem_comes_first():
    out, inn = _convert_legs()
    view = amount_view([out, inn])
    assert [t["trade_id"] for t in view] == ["TRD-T8-OUT"], "转入端不得覆盖已占位的转出端"


def test_amount_view_group_without_redeem_keeps_first_row():
    """防御分支：组内无 redeem 时保留首条（R-b 规定 convert 必有转出端，理论不可达）。"""
    a = _trade("T-IN-1", "subscribe", "100", gid="CNV-T8-X", product=PROD_B)
    b = _trade("T-IN-2", "subscribe", "200", NOW + timedelta(seconds=1), gid="CNV-T8-X",
               product=PROD_B)
    assert amount_view([a, b]) == [a]


def test_amount_view_mixes_groups_and_plain_trades_in_order():
    """多组 + 无 gid 混合：顺序保持，各组各留一条。"""
    n1 = _trade("T-N1", "redeem", "1000")
    n2 = _trade("T-N2", "subscribe", "2000", NOW + timedelta(seconds=3), product=PROD_B)
    out1, in1 = _convert_legs()
    out2 = _trade("TRD-T8-OUT2", "redeem", "500", NOW + timedelta(seconds=4), gid="CNV-T8-0002")
    in2 = _trade("TRD-T8-IN2", "subscribe", "480", NOW + timedelta(seconds=5),
                 gid="CNV-T8-0002", product=PROD_B)
    view = amount_view([n1, in1, out1, n2, in2, out2])
    assert [t["trade_id"] for t in view] == ["T-N1", "TRD-T8-OUT", "T-N2", "TRD-T8-OUT2"]


# ── 2. `run_rules` 视图分流（验收 5/6）───────────────────────────────
def test_daily_total_counts_convert_only_once():
    """RISK-002 走金额视图：一次转换只计转出一端，不翻倍（验收 5）。"""
    out, inn = _convert_legs()
    # 阈值 500000：全量口径 300000+299000=599000 会命中；去重后 300000 不命中
    assert "RISK-002" not in _ids(run_rules([out, inn], _th(), now=NOW))

    # 反证：同一对流水若**没有 gid**（视为两笔独立交易）→ 命中，证明差异来自去重而非阈值
    loose = [_trade("TRD-T8-OUT", "redeem", "300000"), _trade("TRD-T8-IN", "subscribe", "299000")]
    assert "RISK-002" in _ids(run_rules(loose, _th(), now=NOW))


def test_daily_total_detail_reflects_deduped_amount():
    """合计金额与笔数都只算转出端（阈值下调到 290000 让 RISK-002 命中以便读 detail）。"""
    out, inn = _convert_legs()
    hits = run_rules([out, inn], _th(daily_total=Decimal("290000")), now=NOW)
    hit = next(h for h in hits if h.rule_id == "RISK-002")
    assert "300000" in hit.detail
    assert "599000" not in hit.detail, "detail 不得出现两条流水之和"
    assert "共 1 笔" in hit.detail


def test_freq_trade_still_sees_both_legs():
    """RISK-003 看全量（去重未删行，验收 6）：转入端也参与分产品计数。"""
    out, inn = _convert_legs("1000", "900")
    extra = _trade("TRD-T8-B2", "subscribe", "800", NOW + timedelta(seconds=2), product=PROD_B)
    hits = run_rules([out, inn, extra], _th(freq_count=2), now=NOW)
    assert "RISK-003" in _ids(hits), "产品 B 应有 2 笔（convert 转入端 + 普通申购）"
    assert any(PROD_B in h.detail and "2 笔" in h.detail for h in hits if h.rule_id == "RISK-003")


def test_large_amount_still_sees_both_legs():
    """RISK-001 看全量：转入端大额也能命中（逐笔判定，不走去重视图）。"""
    out, inn = _convert_legs("100000", "250000")  # 仅转入端 ≥ 200000
    assert "RISK-001" in _ids(run_rules([out, inn], _th(), now=NOW))


def test_small_then_large_ignores_converted_leg_as_buildup():
    """RISK-005 走金额视图：转入端不得被当成「铺垫」小额（否则虚增 small_count 误报）。"""
    small1 = _trade("T-S1", "redeem", "5000", product=PROD_A)
    small2 = _trade("T-S2", "redeem", "6000", NOW + timedelta(seconds=1), product=PROD_A)
    # 转入端本身是小额（5000）且时间早于转出端 → 若全量参与，会被算作第 3 笔铺垫
    inn = _trade("TRD-T8-IN", "subscribe", "5000", NOW + timedelta(seconds=2),
                 gid=GID, product=PROD_B)
    out = _trade("TRD-T8-OUT", "redeem", "300000", NOW + timedelta(seconds=3),
                 gid=GID, product=PROD_A)
    assert "RISK-005" not in _ids(run_rules([small1, small2, inn, out], _th(), now=NOW))

    # 反证：去掉 gid（转入端成为独立小额）→ 铺垫凑够 3 笔 → 命中
    inn_loose = _trade("TRD-T8-IN", "subscribe", "5000", NOW + timedelta(seconds=2),
                       product=PROD_B)
    assert "RISK-005" in _ids(run_rules([small1, small2, inn_loose, out], _th(), now=NOW))


# ── 3. `process_convert_event` 集成（验收 7）─────────────────────────
def _exec(engine, sql: str, **params) -> None:
    with engine.begin() as conn:
        conn.execute(text(sql), params)


def _rows(engine, sql: str, **params) -> list[dict]:
    with engine.connect() as conn:
        return [dict(r) for r in conn.execute(text(sql), params).mappings()]


def _seed_trades(engine, *trades: dict) -> None:
    for i, t in enumerate(trades):
        _exec(
            engine,
            "INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type, amount, "
            "qty, convert_group_id, trade_status, traded_at) "
            "VALUES (:tid, :c, :p, :tt, :amt, 1, :gid, :st, :at)",
            tid=t["trade_id"], c=t["customer_id"], p=t["product_id"], tt=t["trade_type"],
            amt=float(t["amount"]), gid=t["convert_group_id"], st=t["trade_status"],
            at=t["traded_at"],
        )


def _services(engine) -> dict:
    return dict(
        core_ro=CoreReadOnlyRepository(engine=engine),
        risk_repo=RiskRepository(engine=engine),
    )


def test_process_convert_event_creates_one_alert_with_two_events(sqlite_engine):
    """一次转换 → 一张单、`payload.events` 两条、RISK-002 不翻倍（验收 5/7）。"""
    out, inn = _convert_legs()
    _seed_trades(sqlite_engine, out, inn)
    th = _th(daily_total=Decimal("400000"))  # 全量 599000 会命中；去重后 300000 不命中

    result = process_convert_event(out, inn, thresholds=th, **_services(sqlite_engine))

    assert result["triggered_rules"] == ["RISK-001"], "只应命中单笔大额，累计不翻倍"
    assert result["aml_hit"] is False
    assert len(result["alert_ids"]) == 1, "一次转换只出一张单"

    alerts = _rows(sqlite_engine, "SELECT * FROM risk_alert")
    assert len(alerts) == 1
    import json

    payload = json.loads(alerts[0]["payload"])
    assert len(payload["events"]) == 2, "一张单承载两条事件"
    assert [e["trade_id"] for e in payload["events"]] == ["TRD-T8-OUT", "TRD-T8-IN"]
    assert payload["events"][0]["trade_type"] == "redeem"
    assert payload["events"][1]["trade_type"] == "subscribe"
    # 主流水口径：单主体与 payload.product_id 取转出端
    assert alerts[0]["trade_id"] == "TRD-T8-OUT"
    assert payload["product_id"] == PROD_A

    # 去重不删行：两条流水仍在
    assert len(_rows(sqlite_engine, "SELECT * FROM core_trade")) == 2


def test_process_convert_event_no_hit_writes_pass_audit_only(sqlite_engine):
    """无命中 → 不建单，仅落 pass 审计（events 参数不参与该分支）。"""
    out, inn = _convert_legs("1000", "900")
    _seed_trades(sqlite_engine, out, inn)

    result = process_convert_event(out, inn, thresholds=_th(), **_services(sqlite_engine))

    assert result["alert_ids"] == []
    assert _rows(sqlite_engine, "SELECT * FROM risk_alert") == []
    rows = _rows(sqlite_engine, "SELECT * FROM audit_log WHERE decision = 'pass'")
    assert len(rows) == 1


def test_process_trade_event_still_writes_single_event(sqlite_engine):
    """回归保护：`process_trade_event` 签名与行为不变（payload.events 仍为 1 条）。"""
    t = _trade("TRD-T8-PLAIN", "redeem", "300000")
    _seed_trades(sqlite_engine, t)

    result = process_trade_event(t, thresholds=_th(), **_services(sqlite_engine))

    assert result["triggered_rules"] == ["RISK-001"]
    import json

    alerts = _rows(sqlite_engine, "SELECT * FROM risk_alert")
    payload = json.loads(alerts[0]["payload"])
    assert len(payload["events"]) == 1
    assert payload["events"][0]["trade_id"] == "TRD-T8-PLAIN"


# ── 4. D19 `on_error_hook` 容错 ──────────────────────────────────────
def _boom(*_args, **_kwargs):
    raise ZeroDivisionError("引擎内部炸了")


def test_error_hook_receives_exception_and_swallows_its_own(sqlite_engine, monkeypatch):
    """hook 被调用一次；**hook 自身抛异常必须被吞掉**，原始异常照常上抛。"""
    out, inn = _convert_legs()
    core = CoreReadOnlyRepository(engine=sqlite_engine)
    monkeypatch.setattr(core, "list_trades_range", _boom)
    calls: list[tuple[dict, Exception]] = []

    def hook(out_trade: dict, exc: Exception) -> None:
        calls.append((out_trade, exc))
        raise RuntimeError("hook 自己也炸了")

    with pytest.raises(ZeroDivisionError, match="引擎内部炸了"):
        process_convert_event(
            out, inn, core_ro=core, risk_repo=RiskRepository(engine=sqlite_engine),
            thresholds=_th(), on_error_hook=hook,
        )
    assert len(calls) == 1, "hook 必须被调用一次"
    assert calls[0][0]["trade_id"] == "TRD-T8-OUT", "hook 收到主流水（转出端）"
    assert isinstance(calls[0][1], ZeroDivisionError)


def test_error_hook_none_keeps_original_exception(sqlite_engine, monkeypatch):
    """一期默认 `on_error_hook=None`：异常原样上抛（由阶段 1.5 落 engine_error）。"""
    out, inn = _convert_legs()
    core = CoreReadOnlyRepository(engine=sqlite_engine)
    monkeypatch.setattr(core, "list_trades_range", _boom)

    with pytest.raises(ZeroDivisionError):
        process_convert_event(
            out, inn, core_ro=core, risk_repo=RiskRepository(engine=sqlite_engine),
            thresholds=_th(),
        )


# ── 4. 引擎调用点唯一性（T-8 DoD 1 / R-7）────────────────────────────
def test_process_convert_event_has_single_call_site():
    """`process_convert_event(` 在 `app/` 下**恰 1 处调用**，且位于 `engine_call.run_convert_engine`。

    T-8（FR-C28）把引擎时机从「受理后」挪到「确认事务提交后」，并要求**全仓唯一调用点** ——
    两处调用会让 RISK-002 当日累计翻倍（PRD 验收 5/7）。用 AST 统计 `Call` 节点，
    **不靠字符串匹配**：docstring / 注释里提到函数名不算调用（`engine_call.py` 的说明段
    就含该字样），定义点（`def`）与 `import` 也不是 `Call`。
    """
    root = Path(__file__).resolve().parents[1]
    hits: list[str] = []
    for path in sorted((root / "app").rglob("*.py")):
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue
            func = node.func
            name = getattr(func, "id", None) or getattr(func, "attr", None)
            if name == "process_convert_event":
                hits.append(f"{path.relative_to(root).as_posix()}:{node.lineno}")

    assert len(hits) == 1, (
        "`process_convert_event(` 应恰有 1 处调用点（engine_call.run_convert_engine）；"
        f"实际 {len(hits)} 处：{hits}"
    )
    assert hits[0].startswith("app/service/convert/engine_call.py:"), (
        f"唯一调用点必须落在 engine_call.py，实际：{hits[0]}"
    )
    # 调用所在函数体应是 `run_convert_engine`（防「文件对但函数被搬走」）
    tree = ast.parse(
        (root / "app/service/convert/engine_call.py").read_text(encoding="utf-8")
    )
    owner = None
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            for child in ast.walk(node):
                if isinstance(child, ast.Call):
                    name = getattr(child.func, "id", None) or getattr(child.func, "attr", None)
                    if name == "process_convert_event":
                        owner = node.name
    assert owner == "run_convert_engine", f"调用点宿主函数应为 run_convert_engine，实际 {owner!r}"