"""风控监控：三类规则、明细及处置闭环。"""
from __future__ import annotations

from datetime import datetime, timedelta
from uuid import uuid4

import pytest
from fastapi.testclient import TestClient

from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.api.v1 import compare as compare_api
from app.core.config_schema import (
    RISK_COMPARE_DAILY_THRESHOLD_KEY,
    RISK_ONECLICK_DAILY_THRESHOLD_KEY,
    RISK_SMS_HOURLY_THRESHOLD_KEY,
)
from app.core.security import issue_token_pair
from app.db.session import SessionLocal
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
from app.models.device import DeviceLiveness
from app.models.risk import RiskIncident
from app.repositories import risk as risk_repo
from app.repositories import user as user_repo


@pytest.fixture(autouse=True)
def _reset_risk_rule_config():
    keys = (
        RISK_SMS_HOURLY_THRESHOLD_KEY,
        RISK_ONECLICK_DAILY_THRESHOLD_KEY,
        RISK_COMPARE_DAILY_THRESHOLD_KEY,
        risk_repo.RISK_RESET_BASELINES_KEY,
    )
    with SessionLocal() as db:
        db.query(AppConfig).filter(AppConfig.key.in_(keys)).delete(
            synchronize_session=False
        )
        db.commit()
    yield
    with SessionLocal() as db:
        db.query(AppConfig).filter(AppConfig.key.in_(keys)).delete(
            synchronize_session=False
        )
        db.commit()


def _admin_token() -> str:
    username = "risk_monitor_admin"
    with SessionLocal() as db:
        if admin_repo.get_by_username(db, username) is None:
            admin_repo.create_admin(
                db, username=username, password="risk-pass-123", role="super_admin"
            )
    with TestClient(admin_app) as client:
        response = client.post(
            "/admin/api/auth/login",
            json={"username": username, "password": "risk-pass-123"},
        )
    assert response.status_code == 200
    return response.json()["access_token"]


def _headers() -> dict[str, str]:
    return {"Authorization": f"Bearer {_admin_token()}"}


def _seed_incidents() -> tuple[int, int, int, int, str]:
    suffix = uuid4().hex[:8]
    sms_subject = f"risk_sms_{suffix}"
    oneclick_subject = f"risk_oneclick_{suffix}"
    now = (risk_repo.utcnow() - timedelta(minutes=5)).replace(microsecond=0)
    with SessionLocal() as db:
        user = user_repo.upsert_user_for_login(
            db,
            phone=f"139{int(suffix, 16) % 100_000_000:08d}",
            register_channel="sms",
        )
        second_user = user_repo.upsert_user_for_login(
            db,
            phone=f"137{int(suffix, 16) % 100_000_000:08d}",
            register_channel="sms",
        )
        db.add(
            DeviceLiveness(
                user_id=user.id,
                device_id=sms_subject,
                created_at=now - timedelta(days=10),
            )
        )
        for index in range(5):
            risk_repo.record_behavior_event(
                db,
                event_type=risk_repo.EVENT_SMS_SEND,
                subject_type="device",
                subject_id=sms_subject,
                device_id=sms_subject,
                device_model="OPPO A5",
                phone=f"13877771{index:03d}",
                outcome="success",
                occurred_at=now + timedelta(seconds=index),
                evaluate_rule=risk_repo.RULE_SMS_HOURLY,
            )
        risk_repo.record_behavior_event(
            db,
            event_type=risk_repo.EVENT_SMS_LOGIN,
            subject_type="device",
            subject_id=sms_subject,
            user_id=user.id,
            device_id=sms_subject,
            device_model="OPPO A5",
            phone="13877771000",
            outcome="success",
            occurred_at=now + timedelta(seconds=30),
        )
        risk_repo.record_behavior_event(
            db,
            event_type=risk_repo.EVENT_SMS_LOGIN,
            subject_type="device",
            subject_id=sms_subject,
            user_id=second_user.id,
            device_id=sms_subject,
            device_model="OPPO A5",
            phone="13877771001",
            outcome="success",
            occurred_at=now + timedelta(seconds=31),
        )

        for index in range(21):
            risk_repo.record_behavior_event(
                db,
                event_type=risk_repo.EVENT_ONECLICK_LOGIN,
                subject_type="device",
                subject_id=oneclick_subject,
                user_id=second_user.id if index == 20 else user.id,
                device_id=oneclick_subject,
                device_model="OPPO Reno",
                phone=user.phone,
                outcome="success" if index % 3 else "failed",
                reason="mock provider failure" if index % 3 == 0 else None,
                occurred_at=now + timedelta(seconds=index),
                evaluate_rule=risk_repo.RULE_ONECLICK_DAILY,
            )

        local_now = now.astimezone(risk_repo.CN_TZ).replace(tzinfo=None)
        for index in range(100):
            db.add(
                ComparisonRecord(
                    user_id=user.id,
                    device_id="risk_compare_device",
                    trace_id=f"risk-monitor-{suffix}-{index}",
                    store_name="测试门店",
                    product_names="测试菜品",
                    status="success",
                    items=[{"name": "测试菜品", "qty": 1}],
                    comparison_results=[
                        {"platform_id": "meituan", "platform_name": "美团", "price": 12.5},
                        {"platform_id": "taobao", "platform_name": "淘宝", "price": 11.8},
                    ],
                    skipped_dish_names=[],
                    saved_amount_cents=70,
                    created_at=local_now + timedelta(seconds=index),
                )
            )
        db.commit()
        risk_repo.sync_compare_incident(db, user_id=user.id, at=local_now)

        incidents = {
            incident.rule_code: incident.id
            for incident in db.query(RiskIncident)
            .filter(
                RiskIncident.subject_id.in_(
                    (sms_subject, oneclick_subject, str(user.id))
                )
            )
            .all()
        }
        return (
            incidents[risk_repo.RULE_SMS_HOURLY],
            incidents[risk_repo.RULE_ONECLICK_DAILY],
            incidents[risk_repo.RULE_COMPARE_DAILY],
            user.id,
            sms_subject,
        )


def test_risk_monitor_summary_lists_and_details() -> None:
    sms_id, oneclick_id, compare_id, _, _ = _seed_incidents()
    headers = _headers()
    with TestClient(admin_app) as client:
        summary = client.get("/admin/api/risk-monitor/summary", headers=headers)
        assert summary.status_code == 200
        cards = {card["kind"]: card for card in summary.json()["cards"]}
        assert cards["sms"]["alert_subject_count"] >= 1
        assert cards["sms"]["today_total"] >= 5
        assert cards["oneclick"]["alert_subject_count"] >= 1
        assert cards["oneclick"]["today_total"] >= 20
        assert cards["compare"]["alert_subject_count"] >= 1
        assert cards["compare"]["today_total"] >= 100

        sms_list = client.get("/admin/api/risk-monitor/incidents/sms", headers=headers)
        assert sms_list.status_code == 200
        sms_item = next(
            item for item in sms_list.json()["items"] if item["incident_id"] == sms_id
        )
        assert sms_item["device_model"] == "OPPO A5"
        assert sms_item["event_count"] == 5
        assert sms_item["first_used_at"].startswith(
            (risk_repo.utcnow() - timedelta(days=10)).date().isoformat()
        )

        sms_detail = client.get(
            f"/admin/api/risk-monitor/incidents/sms/{sms_id}/details",
            headers=headers,
        )
        assert sms_detail.status_code == 200
        assert sms_detail.json()["total"] == 5
        assert sms_detail.json()["distinct_accounts"] == 2
        assert any(item["verified"] for item in sms_detail.json()["items"])

        oneclick_detail = client.get(
            f"/admin/api/risk-monitor/incidents/oneclick/{oneclick_id}/details?limit=1",
            headers=headers,
        )
        assert oneclick_detail.status_code == 200
        assert oneclick_detail.json()["total"] == 21
        assert oneclick_detail.json()["distinct_accounts"] == 2

        compare_detail = client.get(
            f"/admin/api/risk-monitor/incidents/compare/{compare_id}/details",
            headers=headers,
        )
        assert compare_detail.status_code == 200
        assert compare_detail.json()["total"] == 100
        assert compare_detail.json()["items"][0]["prices"]["美团"] == 12.5


def test_risk_monitor_ignore_and_block_are_enforced(client: TestClient) -> None:
    sms_id, oneclick_id, compare_id, user_id, sms_subject = _seed_incidents()
    headers = _headers()
    with TestClient(admin_app) as admin_client:
        ignored = admin_client.post(
            f"/admin/api/risk-monitor/incidents/{oneclick_id}/ignore",
            json={"reason": "测试确认正常"},
            headers=headers,
        )
        assert ignored.status_code == 200
        assert ignored.json()["status"] == "ignored"

        blocked_device = admin_client.post(
            f"/admin/api/risk-monitor/incidents/{sms_id}/block",
            json={"reason": "测试设备异常"},
            headers=headers,
        )
        assert blocked_device.status_code == 200
        assert blocked_device.json()["restriction_id"]

        blocked_user = admin_client.post(
            f"/admin/api/risk-monitor/incidents/{compare_id}/block",
            json={"reason": "测试账号异常"},
            headers=headers,
        )
        assert blocked_user.status_code == 200
        device_restriction_id = blocked_device.json()["restriction_id"]
        user_restriction_id = blocked_user.json()["restriction_id"]

        blocked_sms = admin_client.get(
            "/admin/api/risk-monitor/incidents/sms?status=blocked",
            headers=headers,
        )
        assert blocked_sms.status_code == 200
        assert blocked_sms.json()["items"][0]["restriction_id"] == device_restriction_id
        assert blocked_sms.json()["items"][0]["restricted"] is True

        blocked_compare = admin_client.get(
            "/admin/api/risk-monitor/incidents/compare?status=blocked",
            headers=headers,
        )
        assert blocked_compare.status_code == 200
        assert blocked_compare.json()["items"][0]["restriction_id"] == user_restriction_id
        assert blocked_compare.json()["items"][0]["restricted"] is True

    denied_sms = client.post(
        "/api/v1/auth/sms/send",
        json={
            "phone": "13877779999",
            "device_id": sms_subject,
            "device_model": "OPPO A5",
        },
    )
    assert denied_sms.status_code == 403
    assert denied_sms.json()["detail"] == "当前设备环境异常，暂无法发送验证码"

    tokens = issue_token_pair(user_id)
    denied_compare = client.post(
        "/api/v1/compare/start",
        json={"trace_id": "risk-blocked-new-trace", "device_id": "risk_compare_device"},
        headers={"Authorization": f"Bearer {tokens['access_token']}"},
    )
    assert denied_compare.status_code == 403
    assert denied_compare.json()["detail"] == "账号存在异常，该功能暂不可用"

    denied_compare_step = client.post(
        "/api/v1/intent/recognize",
        json={},
        headers={"Authorization": f"Bearer {tokens['access_token']}"},
    )
    assert denied_compare_step.status_code == 403
    assert denied_compare_step.json()["detail"] == "账号存在异常，该功能暂不可用"

    with TestClient(admin_app) as admin_client:
        assert (
            admin_client.post(
                f"/admin/api/risk-monitor/restrictions/{device_restriction_id}/revoke",
                headers=headers,
            ).status_code
            == 200
        )
        assert (
            admin_client.post(
                f"/admin/api/risk-monitor/restrictions/{user_restriction_id}/revoke",
                headers=headers,
            ).status_code
            == 200
        )

    allowed_sms = client.post(
        "/api/v1/auth/sms/send",
        json={
            "phone": "13877779998",
            "device_id": sms_subject,
            "device_model": "OPPO A5",
        },
    )
    assert allowed_sms.status_code == 200

    with SessionLocal() as db:
        assert not risk_repo.is_restricted(
            db,
            subject_type="user",
            subject_id=str(user_id),
            scope=risk_repo.SCOPE_ECONOMIC_ACCOUNT,
        )
        assert db.get(RiskIncident, compare_id).status == "resolved"


def test_admin_can_reset_all_alerts_and_sms_restarts_from_zero() -> None:
    sms_id, oneclick_id, compare_id, _, sms_subject = _seed_incidents()
    headers = _headers()

    with TestClient(admin_app) as client:
        reset = client.post("/admin/api/risk-monitor/reset", headers=headers)
        assert reset.status_code == 200
        payload = reset.json()
        assert payload["reset_incident_count"] >= 3
        assert payload["reset_counts"]["sms"] >= 1
        assert payload["reset_counts"]["oneclick"] >= 1
        assert payload["reset_counts"]["compare"] >= 1

        summary = client.get("/admin/api/risk-monitor/summary", headers=headers)
        assert summary.status_code == 200
        cards = {card["kind"]: card for card in summary.json()["cards"]}
        assert all(card["alert_subject_count"] == 0 for card in cards.values())
        # 重置只清待处理报警，不删除今日事实总量。
        assert cards["sms"]["today_total"] >= 5
        assert cards["oneclick"]["today_total"] >= 20
        assert cards["compare"]["today_total"] >= 100

    reset_at = datetime.fromisoformat(payload["reset_at"])
    with SessionLocal() as db:
        for incident_id in (sms_id, oneclick_id, compare_id):
            incident = db.get(RiskIncident, incident_id)
            assert incident.status == "resolved"
            assert incident.action_reason == risk_repo.MANUAL_RESET_REASON

        for index in range(4):
            risk_repo.record_behavior_event(
                db,
                event_type=risk_repo.EVENT_SMS_SEND,
                subject_type="device",
                subject_id=sms_subject,
                device_id=sms_subject,
                phone=f"13688880{index:03d}",
                outcome="success",
                occurred_at=reset_at + timedelta(seconds=index + 1),
                evaluate_rule=risk_repo.RULE_SMS_HOURLY,
            )
        assert db.get(RiskIncident, sms_id).status == "resolved"

        risk_repo.record_behavior_event(
            db,
            event_type=risk_repo.EVENT_SMS_SEND,
            subject_type="device",
            subject_id=sms_subject,
            device_id=sms_subject,
            phone="13688880999",
            outcome="success",
            occurred_at=reset_at + timedelta(seconds=5),
            evaluate_rule=risk_repo.RULE_SMS_HOURLY,
        )
        incident = db.get(RiskIncident, sms_id)
        assert incident.status == "open"
        assert incident.event_count == 5
        assert incident.window_start == reset_at.replace(tzinfo=None)


def test_real_compare_harvest_creates_incident_at_threshold() -> None:
    suffix = uuid4().hex[:8]
    local_now = risk_repo.utcnow().astimezone(risk_repo.CN_TZ).replace(tzinfo=None)
    with SessionLocal() as db:
        user = user_repo.upsert_user_for_login(
            db,
            phone=f"136{int(suffix, 16) % 100_000_000:08d}",
            register_channel="sms",
        )
        for index in range(99):
            db.add(
                ComparisonRecord(
                    user_id=user.id,
                    device_id="real-harvest-device",
                    trace_id=f"risk-real-harvest-{suffix}-{index}",
                    status="running",
                    created_at=local_now + timedelta(milliseconds=index),
                )
            )
        db.commit()
        user_id = user.id

    compare_api._harvest_running_blocking(
        f"risk-real-harvest-{suffix}-99",
        user_id,
        "food",
        "real-harvest-device",
        {"model": "OPPO A5"},
        None,
    )

    with SessionLocal() as db:
        incident = db.query(RiskIncident).filter(
            RiskIncident.rule_code == risk_repo.RULE_COMPARE_DAILY,
            RiskIncident.subject_id == str(user_id),
            RiskIncident.window_key == local_now.strftime("%Y-%m-%d"),
        ).one()
        assert incident.event_count == 100


def test_admin_can_edit_rules_and_current_window_is_reconciled() -> None:
    suffix = uuid4().hex[:8]
    subject = f"risk_dynamic_sms_{suffix}"
    now = risk_repo.utcnow().replace(microsecond=0)
    with SessionLocal() as db:
        for index in range(3):
            risk_repo.record_behavior_event(
                db,
                event_type=risk_repo.EVENT_SMS_SEND,
                subject_type="device",
                subject_id=subject,
                device_id=subject,
                device_model="动态规则测试机",
                phone=f"13800001{index:03d}",
                outcome="success",
                occurred_at=now + timedelta(seconds=index),
            )

    headers = _headers()
    with TestClient(admin_app) as client:
        defaults = client.get("/admin/api/risk-monitor/rules", headers=headers)
        assert defaults.status_code == 200
        assert defaults.json() == {
            "sms_hourly_threshold": 5,
            "oneclick_daily_threshold": 20,
            "compare_daily_threshold": 100,
        }

        lowered = client.patch(
            "/admin/api/risk-monitor/rules",
            headers=headers,
            json={
                "sms_hourly_threshold": 3,
                "oneclick_daily_threshold": 20,
                "compare_daily_threshold": 100,
            },
        )
        assert lowered.status_code == 200
        with SessionLocal() as db:
            incident = db.query(RiskIncident).filter(
                RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
                RiskIncident.subject_id == subject,
            ).one()
            assert incident.status == "open"
            assert incident.event_count == 3

        raised = client.patch(
            "/admin/api/risk-monitor/rules",
            headers=headers,
            json={
                "sms_hourly_threshold": 4,
                "oneclick_daily_threshold": 20,
                "compare_daily_threshold": 100,
            },
        )
        assert raised.status_code == 200
        with SessionLocal() as db:
            incident = db.query(RiskIncident).filter(
                RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
                RiskIncident.subject_id == subject,
            ).one()
            assert incident.status == "resolved"
            assert incident.action_reason == risk_repo.AUTO_RESOLVED_REASON

        reopened = client.patch(
            "/admin/api/risk-monitor/rules",
            headers=headers,
            json={
                "sms_hourly_threshold": 3,
                "oneclick_daily_threshold": 20,
                "compare_daily_threshold": 100,
            },
        )
        assert reopened.status_code == 200
        with SessionLocal() as db:
            incident = db.query(RiskIncident).filter(
                RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
                RiskIncident.subject_id == subject,
            ).one()
            assert incident.status == "open"
            assert incident.triggered_at == (
                now + timedelta(seconds=2)
            ).replace(tzinfo=None)

        invalid = client.patch(
            "/admin/api/risk-monitor/rules",
            headers=headers,
            json={
                "sms_hourly_threshold": 21,
                "oneclick_daily_threshold": 20,
                "compare_daily_threshold": 100,
            },
        )
        assert invalid.status_code == 422