shaguabijia-app-server/app/admin/repositories/risk_monitor.py

"""风控监控读模型：聚合通用行为流水、风险事件与比价记录。"""
from __future__ import annotations

from datetime import UTC, datetime, timedelta

from sqlalchemy import desc, func, select
from sqlalchemy.orm import Session

from app.core.config import settings
from app.models.comparison import ComparisonRecord
from app.models.device import DeviceLiveness
from app.models.risk import BehaviorEvent, RiskIncident
from app.models.user import User
from app.repositories import risk as risk_repo

KIND_TO_RULE = {
    "sms": risk_repo.RULE_SMS_HOURLY,
    "oneclick": risk_repo.RULE_ONECLICK_DAILY,
    "compare": risk_repo.RULE_COMPARE_DAILY,
}
RULE_TO_KIND = {rule: kind for kind, rule in KIND_TO_RULE.items()}


def _day_bounds(now: datetime | None = None) -> tuple[str, datetime, datetime]:
    current = (now or datetime.now(UTC)).astimezone(risk_repo.CN_TZ)
    start = current.replace(hour=0, minute=0, second=0, microsecond=0)
    end = start + timedelta(days=1)
    return (
        start.strftime("%Y-%m-%d"),
        start.astimezone(UTC),
        end.astimezone(UTC),
    )


def _compare_day_bounds(now: datetime | None = None) -> tuple[datetime, datetime]:
    current = (now or datetime.now(UTC)).astimezone(risk_repo.CN_TZ).replace(
        tzinfo=None
    )
    start = current.replace(hour=0, minute=0, second=0, microsecond=0)
    return start, start + timedelta(days=1)


def sync_today_compare_incidents(db: Session, now: datetime | None = None) -> None:
    risk_repo.reconcile_compare_rule(db, at=now)


def reset_open_alerts(
    db: Session, *, admin_id: int, reset_at: datetime
) -> dict[str, int]:
    """将三类待处理报警归零，并从 reset_at 开始重新累计。"""
    rule_codes = tuple(KIND_TO_RULE.values())
    incidents = list(
        db.scalars(
            select(RiskIncident).where(
                RiskIncident.rule_code.in_(rule_codes),
                RiskIncident.status == "open",
            )
        ).all()
    )
    counts = {kind: 0 for kind in KIND_TO_RULE}
    for incident in incidents:
        incident.status = "resolved"
        incident.action_reason = risk_repo.MANUAL_RESET_REASON
        incident.handled_by = admin_id
        incident.handled_at = reset_at
        counts[RULE_TO_KIND[incident.rule_code]] += 1
    risk_repo.reset_rule_baselines(
        db,
        rule_codes=rule_codes,
        reset_at=reset_at,
        admin_id=admin_id,
    )
    return counts


def summary(db: Session, now: datetime | None = None) -> dict:
    date_key, start, end = _day_bounds(now)
    compare_start, compare_end = _compare_day_bounds(now)
    current = now or datetime.now(UTC)
    risk_repo.reconcile_behavior_rule(
        db, rule_code=risk_repo.RULE_SMS_HOURLY, at=current, commit=False
    )
    risk_repo.reconcile_behavior_rule(
        db, rule_code=risk_repo.RULE_ONECLICK_DAILY, at=current, commit=False
    )
    risk_repo.reconcile_compare_rule(db, at=current, commit=False)
    db.commit()
    sms_threshold = risk_repo.get_rule_threshold(db, risk_repo.RULE_SMS_HOURLY)
    oneclick_threshold = risk_repo.get_rule_threshold(
        db, risk_repo.RULE_ONECLICK_DAILY
    )
    compare_threshold = risk_repo.get_rule_threshold(db, risk_repo.RULE_COMPARE_DAILY)

    def _behavior_total(event_type: str, outcomes: tuple[str, ...]) -> int:
        return int(
            db.scalar(
                select(func.count(BehaviorEvent.id)).where(
                    BehaviorEvent.event_type == event_type,
                    BehaviorEvent.outcome.in_(outcomes),
                    BehaviorEvent.occurred_at >= start,
                    BehaviorEvent.occurred_at < end,
                )
            )
            or 0
        )

    compare_total = int(
        db.scalar(
            select(func.count(ComparisonRecord.id)).where(
                ComparisonRecord.created_at >= compare_start,
                ComparisonRecord.created_at < compare_end,
            )
        )
        or 0
    )

    def _open_count(rule_code: str) -> int:
        return int(
            db.scalar(
                select(func.count(RiskIncident.id)).where(
                    RiskIncident.rule_code == rule_code,
                    RiskIncident.status == "open",
                    RiskIncident.window_key.like(f"{date_key}%"),
                )
            )
            or 0
        )

    return {
        "date": date_key,
        "updated_at": now or datetime.now(UTC),
        "cards": [
            {
                "kind": "sms",
                "alert_subject_count": _open_count(risk_repo.RULE_SMS_HOURLY),
                "today_total": _behavior_total(risk_repo.EVENT_SMS_SEND, ("success",)),
                "threshold": sms_threshold,
                "rule_text": f"单设备 1 小时 ≥ {sms_threshold} 次",
            },
            {
                "kind": "oneclick",
                "alert_subject_count": _open_count(risk_repo.RULE_ONECLICK_DAILY),
                "today_total": _behavior_total(
                    risk_repo.EVENT_ONECLICK_LOGIN, ("success", "failed")
                ),
                "threshold": oneclick_threshold,
                "rule_text": f"单设备当日 ≥ {oneclick_threshold} 次",
            },
            {
                "kind": "compare",
                "alert_subject_count": _open_count(risk_repo.RULE_COMPARE_DAILY),
                "today_total": compare_total,
                "threshold": compare_threshold,
                "rule_text": f"单账户当日 ≥ {compare_threshold} 次",
            },
        ],
    }


def _latest_device_model(db: Session, subject_id: str) -> str | None:
    return db.execute(
        select(BehaviorEvent.device_model)
        .where(
            BehaviorEvent.subject_type == "device",
            BehaviorEvent.subject_id == subject_id,
            BehaviorEvent.device_model.is_not(None),
            BehaviorEvent.device_model != "",
        )
        .order_by(BehaviorEvent.occurred_at.desc(), BehaviorEvent.id.desc())
        .limit(1)
    ).scalar_one_or_none()


def _first_used_at(db: Session, subject_id: str) -> datetime | None:
    behavior_first = db.scalar(
        select(func.min(BehaviorEvent.occurred_at)).where(
            BehaviorEvent.subject_type == "device",
            BehaviorEvent.subject_id == subject_id,
        )
    )
    registered_first = db.scalar(
        select(func.min(DeviceLiveness.created_at)).where(
            DeviceLiveness.device_id == subject_id
        )
    )
    observed = [value for value in (behavior_first, registered_first) if value is not None]
    if not observed:
        return None

    def _utc(value: datetime) -> datetime:
        return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)

    return min(observed, key=_utc)


def _common_device(db: Session, user_id: int, start: datetime, end: datetime) -> str | None:
    return db.execute(
        select(ComparisonRecord.device_id)
        .where(
            ComparisonRecord.user_id == user_id,
            ComparisonRecord.device_id.is_not(None),
            ComparisonRecord.created_at >= start,
            ComparisonRecord.created_at < end,
        )
        .group_by(ComparisonRecord.device_id)
        .order_by(func.count(ComparisonRecord.id).desc(), ComparisonRecord.device_id.asc())
        .limit(1)
    ).scalar_one_or_none()


def list_incidents(
    db: Session,
    *,
    kind: str,
    status: str = "open",
    limit: int = 20,
    cursor: int = 0,
) -> tuple[list[dict], int | None, int]:
    rule_code = KIND_TO_RULE[kind]
    stmt = select(RiskIncident).where(RiskIncident.rule_code == rule_code)
    count_stmt = select(func.count(RiskIncident.id)).where(
        RiskIncident.rule_code == rule_code
    )
    if status != "all":
        stmt = stmt.where(RiskIncident.status == status)
        count_stmt = count_stmt.where(RiskIncident.status == status)
    total = int(db.scalar(count_stmt) or 0)
    incidents = list(
        db.scalars(
            stmt.order_by(desc(RiskIncident.triggered_at), desc(RiskIncident.id))
            .offset(cursor)
            .limit(limit)
        ).all()
    )

    user_ids = [int(i.subject_id) for i in incidents if i.subject_type == "user"]
    users = {
        u.id: u
        for u in db.scalars(select(User).where(User.id.in_(user_ids))).all()
    } if user_ids else {}

    items: list[dict] = []
    for incident in incidents:
        item = {
            "incident_id": incident.id,
            "kind": kind,
            "subject_type": incident.subject_type,
            "subject_id": incident.subject_id,
            "window_start": incident.window_start,
            "window_end": incident.window_end,
            "first_event_at": incident.first_event_at,
            "triggered_at": incident.triggered_at,
            "last_event_at": incident.last_event_at,
            "event_count": incident.event_count,
            "status": incident.status,
        }
        if incident.subject_type == "device":
            item["device_model"] = _latest_device_model(db, incident.subject_id)
            item["first_used_at"] = _first_used_at(db, incident.subject_id)
            item["restricted"] = risk_repo.is_restricted(
                db,
                subject_type="device",
                subject_id=incident.subject_id,
                scope=risk_repo.SCOPE_AUTH_DEVICE,
            )
        else:
            uid = int(incident.subject_id)
            user = users.get(uid)
            item.update(
                {
                    "user_id": uid,
                    "phone": user.phone if user else None,
                    "registered_at": user.created_at if user else None,
                    "common_device_id": _common_device(
                        db, uid, incident.window_start, incident.window_end
                    ),
                    "restricted": risk_repo.is_restricted(
                        db,
                        subject_type="user",
                        subject_id=incident.subject_id,
                        scope=risk_repo.SCOPE_ECONOMIC_ACCOUNT,
                    ),
                }
            )
        items.append(item)
    next_cursor = cursor + len(items) if cursor + len(items) < total else None
    return items, next_cursor, total


def _page_events(
    db: Session,
    incident: RiskIncident,
    *,
    event_type: str,
    outcomes: tuple[str, ...] | None,
    limit: int,
    cursor: int,
) -> tuple[list[BehaviorEvent], int]:
    filters = [
        BehaviorEvent.event_type == event_type,
        BehaviorEvent.subject_type == incident.subject_type,
        BehaviorEvent.subject_id == incident.subject_id,
        BehaviorEvent.occurred_at >= incident.window_start,
        BehaviorEvent.occurred_at < incident.window_end,
    ]
    if outcomes:
        filters.append(BehaviorEvent.outcome.in_(outcomes))
    total = int(db.scalar(select(func.count(BehaviorEvent.id)).where(*filters)) or 0)
    rows = list(
        db.scalars(
            select(BehaviorEvent)
            .where(*filters)
            .order_by(BehaviorEvent.occurred_at.desc(), BehaviorEvent.id.desc())
            .offset(cursor)
            .limit(limit)
        ).all()
    )
    return rows, total


def _user_map(db: Session, user_ids: set[int]) -> dict[int, User]:
    if not user_ids:
        return {}
    return {u.id: u for u in db.scalars(select(User).where(User.id.in_(user_ids))).all()}


def _distinct_event_users(
    db: Session,
    incident: RiskIncident,
    *,
    event_type: str,
    outcomes: tuple[str, ...],
) -> int:
    return int(
        db.scalar(
            select(func.count(func.distinct(BehaviorEvent.user_id))).where(
                BehaviorEvent.event_type == event_type,
                BehaviorEvent.subject_type == incident.subject_type,
                BehaviorEvent.subject_id == incident.subject_id,
                BehaviorEvent.outcome.in_(outcomes),
                BehaviorEvent.user_id.is_not(None),
                BehaviorEvent.occurred_at >= incident.window_start,
                BehaviorEvent.occurred_at < incident.window_end,
            )
        )
        or 0
    )


def _sms_distinct_accounts(db: Session, incident: RiskIncident) -> int:
    """统计整个告警窗口内，确实完成本设备短信验证的不同账号数。"""
    send_rows = db.execute(
        select(BehaviorEvent.phone, BehaviorEvent.occurred_at).where(
            BehaviorEvent.event_type == risk_repo.EVENT_SMS_SEND,
            BehaviorEvent.subject_type == incident.subject_type,
            BehaviorEvent.subject_id == incident.subject_id,
            BehaviorEvent.outcome == "success",
            BehaviorEvent.phone.is_not(None),
            BehaviorEvent.occurred_at >= incident.window_start,
            BehaviorEvent.occurred_at < incident.window_end,
        )
    ).all()
    phones = {phone for phone, _ in send_rows}
    if not phones:
        return 0
    login_rows = db.execute(
        select(
            BehaviorEvent.user_id,
            BehaviorEvent.phone,
            BehaviorEvent.occurred_at,
        )
        .where(
            BehaviorEvent.event_type == risk_repo.EVENT_SMS_LOGIN,
            BehaviorEvent.subject_id == incident.subject_id,
            BehaviorEvent.phone.in_(phones),
            BehaviorEvent.user_id.is_not(None),
            BehaviorEvent.outcome == "success",
            BehaviorEvent.occurred_at >= incident.window_start,
            BehaviorEvent.occurred_at
            < incident.window_end + timedelta(seconds=settings.SMS_CODE_TTL_SEC),
        )
        .order_by(BehaviorEvent.occurred_at.asc(), BehaviorEvent.id.asc())
    ).all()
    matched_users = {
        user_id
        for user_id, phone, login_at in login_rows
        if any(
            sent_phone == phone
            and sent_at <= login_at
            and login_at
            <= sent_at + timedelta(seconds=settings.SMS_CODE_TTL_SEC)
            for sent_phone, sent_at in send_rows
        )
    }
    return len(matched_users)


def incident_details(
    db: Session,
    *,
    incident: RiskIncident,
    kind: str,
    limit: int,
    cursor: int,
) -> dict:
    if kind == "sms":
        rows, total = _page_events(
            db,
            incident,
            event_type=risk_repo.EVENT_SMS_SEND,
            outcomes=("success",),
            limit=limit,
            cursor=cursor,
        )
        phone_values = {r.phone for r in rows if r.phone}
        login_rows = list(
            db.scalars(
                select(BehaviorEvent)
                .where(
                    BehaviorEvent.event_type == risk_repo.EVENT_SMS_LOGIN,
                    BehaviorEvent.subject_id == incident.subject_id,
                    BehaviorEvent.phone.in_(phone_values),
                    BehaviorEvent.outcome == "success",
                    BehaviorEvent.occurred_at >= incident.window_start,
                    BehaviorEvent.occurred_at
                    < incident.window_end + timedelta(seconds=settings.SMS_CODE_TTL_SEC),
                )
                .order_by(BehaviorEvent.occurred_at.asc(), BehaviorEvent.id.asc())
            ).all()
        ) if phone_values else []
        users = _user_map(db, {r.user_id for r in login_rows if r.user_id is not None})
        items = []
        for row in rows:
            match = next(
                (
                    login
                    for login in login_rows
                    if login.phone == row.phone
                    and login.occurred_at >= row.occurred_at
                    and login.occurred_at
                    <= row.occurred_at + timedelta(seconds=settings.SMS_CODE_TTL_SEC)
                ),
                None,
            )
            user = users.get(match.user_id) if match and match.user_id else None
            items.append(
                {
                    "id": row.id,
                    "occurred_at": row.occurred_at,
                    "outcome": row.outcome,
                    "phone": row.phone,
                    "user_id": user.id if user else None,
                    "username": user.username if user else None,
                    "account_registered_at": user.created_at if user else None,
                    "verified": match is not None,
                }
            )
        distinct_accounts = _sms_distinct_accounts(db, incident)
    elif kind == "oneclick":
        rows, total = _page_events(
            db,
            incident,
            event_type=risk_repo.EVENT_ONECLICK_LOGIN,
            outcomes=("success", "failed"),
            limit=limit,
            cursor=cursor,
        )
        users = _user_map(db, {r.user_id for r in rows if r.user_id is not None})
        items = []
        for row in rows:
            user = users.get(row.user_id) if row.user_id else None
            items.append(
                {
                    "id": row.id,
                    "occurred_at": row.occurred_at,
                    "outcome": row.outcome,
                    "phone": row.phone,
                    "user_id": row.user_id,
                    "username": user.username if user else None,
                    "account_registered_at": user.created_at if user else None,
                    "reason": row.reason,
                }
            )
        distinct_accounts = _distinct_event_users(
            db,
            incident,
            event_type=risk_repo.EVENT_ONECLICK_LOGIN,
            outcomes=("success", "failed"),
        )
    else:
        user_id = int(incident.subject_id)
        filters = (
            ComparisonRecord.user_id == user_id,
            ComparisonRecord.created_at >= incident.window_start,
            ComparisonRecord.created_at < incident.window_end,
        )
        total = int(db.scalar(select(func.count(ComparisonRecord.id)).where(*filters)) or 0)
        rows = list(
            db.scalars(
                select(ComparisonRecord)
                .where(*filters)
                .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())
                .offset(cursor)
                .limit(limit)
            ).all()
        )
        items = []
        for row in rows:
            prices = {
                str(result.get("platform_name") or result.get("platform_id") or "未知平台"): (
                    float(result["price"]) if result.get("price") is not None else None
                )
                for result in (row.comparison_results or [])
            }
            items.append(
                {
                    "id": row.id,
                    "occurred_at": row.created_at,
                    "outcome": row.status,
                    "store_name": row.store_name,
                    "product_names": row.product_names,
                    "prices": prices,
                    "saved_amount_yuan": (
                        row.saved_amount_cents / 100
                        if row.saved_amount_cents is not None
                        else None
                    ),
                    "status": row.status,
                }
            )
        distinct_accounts = None
    next_cursor = cursor + len(items) if cursor + len(items) < total else None
    return {
        "kind": kind,
        "incident_id": incident.id,
        "event_count": incident.event_count,
        "distinct_accounts": distinct_accounts,
        "items": items,
        "next_cursor": next_cursor,
        "total": total,
    }