"""通用行为事件、风险规则与主体限制仓库。"""
from __future__ import annotations

from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from zoneinfo import ZoneInfo

from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.core.config_schema import (
    RISK_COMPARE_DAILY_THRESHOLD_KEY,
    RISK_ONECLICK_DAILY_THRESHOLD_KEY,
    RISK_SMS_HOURLY_THRESHOLD_KEY,
)
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
from app.models.risk import BehaviorEvent, RiskIncident, SubjectRestriction
from app.repositories import app_config

CN_TZ = ZoneInfo("Asia/Shanghai")

RULE_SMS_HOURLY = "sms_device_hourly_5"
RULE_ONECLICK_DAILY = "oneclick_device_daily_20"
RULE_COMPARE_DAILY = "compare_account_daily_100"

EVENT_SMS_SEND = "sms_send"
EVENT_SMS_LOGIN = "sms_login"
EVENT_ONECLICK_LOGIN = "oneclick_login"

SCOPE_AUTH_DEVICE = "auth_device"
SCOPE_ECONOMIC_ACCOUNT = "economic_account"
AUTO_RESOLVED_REASON = "规则阈值调整后不再命中"
MANUAL_RESET_REASON = "管理员重置报警计数"
RISK_RESET_BASELINES_KEY = "risk_monitor_reset_baselines"


@dataclass(frozen=True)
class RuleSpec:
    code: str
    event_type: str
    subject_type: str
    threshold_key: str
    window: str
    count_outcomes: tuple[str, ...]


RULES: dict[str, RuleSpec] = {
    RULE_SMS_HOURLY: RuleSpec(
        code=RULE_SMS_HOURLY,
        event_type=EVENT_SMS_SEND,
        subject_type="device",
        threshold_key=RISK_SMS_HOURLY_THRESHOLD_KEY,
        window="hour",
        count_outcomes=("success",),
    ),
    RULE_ONECLICK_DAILY: RuleSpec(
        code=RULE_ONECLICK_DAILY,
        event_type=EVENT_ONECLICK_LOGIN,
        subject_type="device",
        threshold_key=RISK_ONECLICK_DAILY_THRESHOLD_KEY,
        window="day",
        count_outcomes=("success", "failed"),
    ),
}

RULE_THRESHOLD_KEYS: dict[str, str] = {
    RULE_SMS_HOURLY: RISK_SMS_HOURLY_THRESHOLD_KEY,
    RULE_ONECLICK_DAILY: RISK_ONECLICK_DAILY_THRESHOLD_KEY,
    RULE_COMPARE_DAILY: RISK_COMPARE_DAILY_THRESHOLD_KEY,
}


def get_rule_threshold(db: Session, rule_code: str) -> int:
    """读取规则当前阈值；配置表为空时回退上线前的 5/20/100 默认值。"""
    return int(app_config.get_value(db, RULE_THRESHOLD_KEYS[rule_code]))


def utcnow() -> datetime:
    return datetime.now(UTC)


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


def get_rule_reset_at(db: Session, rule_code: str) -> datetime | None:
    """返回规则最近一次全局重置时间；异常旧值按未重置处理。"""
    row = db.get(AppConfig, RISK_RESET_BASELINES_KEY)
    if row is None or not isinstance(row.value, dict):
        return None
    raw = row.value.get(rule_code)
    if not isinstance(raw, str):
        return None
    try:
        parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
    except ValueError:
        return None
    return _ensure_aware(parsed).astimezone(UTC)


def reset_rule_baselines(
    db: Session,
    *,
    rule_codes: tuple[str, ...],
    reset_at: datetime,
    admin_id: int,
) -> None:
    """持久化全局重置基线；不删除任何行为或比价流水。"""
    row = db.get(AppConfig, RISK_RESET_BASELINES_KEY)
    value = dict(row.value) if row and isinstance(row.value, dict) else {}
    timestamp = _ensure_aware(reset_at).astimezone(UTC).isoformat()
    value.update({rule_code: timestamp for rule_code in rule_codes})
    if row is None:
        db.add(
            AppConfig(
                key=RISK_RESET_BASELINES_KEY,
                value=value,
                updated_by_admin_id=admin_id,
            )
        )
    else:
        row.value = value
        row.updated_by_admin_id = admin_id
    db.flush()


def _window_bounds(at: datetime, window: str) -> tuple[str, datetime, datetime]:
    local = _ensure_aware(at).astimezone(CN_TZ)
    if window == "hour":
        start_local = local.replace(minute=0, second=0, microsecond=0)
        key = start_local.strftime("%Y-%m-%dT%H")
        end_local = start_local + timedelta(hours=1)
    elif window == "day":
        start_local = local.replace(hour=0, minute=0, second=0, microsecond=0)
        key = start_local.strftime("%Y-%m-%d")
        end_local = start_local + timedelta(days=1)
    else:  # pragma: no cover - 规则声明错误应尽早暴露
        raise ValueError(f"unsupported risk window: {window}")
    return key, start_local.astimezone(UTC), end_local.astimezone(UTC)


def _event_stats(
    db: Session,
    spec: RuleSpec,
    *,
    subject_id: str,
    start: datetime,
    end: datetime,
    threshold: int,
) -> tuple[int, datetime | None, datetime | None, datetime | None]:
    filters = (
        BehaviorEvent.event_type == spec.event_type,
        BehaviorEvent.subject_type == spec.subject_type,
        BehaviorEvent.subject_id == subject_id,
        BehaviorEvent.outcome.in_(spec.count_outcomes),
        BehaviorEvent.occurred_at >= start,
        BehaviorEvent.occurred_at < end,
    )
    count, first_at, last_at = db.execute(
        select(
            func.count(BehaviorEvent.id),
            func.min(BehaviorEvent.occurred_at),
            func.max(BehaviorEvent.occurred_at),
        ).where(*filters)
    ).one()
    triggered_at = None
    if int(count or 0) >= threshold:
        triggered_at = db.execute(
            select(BehaviorEvent.occurred_at)
            .where(*filters)
            .order_by(BehaviorEvent.occurred_at.asc(), BehaviorEvent.id.asc())
            .offset(threshold - 1)
            .limit(1)
        ).scalar_one()
    return int(count or 0), first_at, last_at, triggered_at


def _upsert_incident(
    db: Session,
    *,
    rule_code: str,
    event_type: str,
    subject_type: str,
    subject_id: str,
    window_key: str,
    window_start: datetime,
    window_end: datetime,
    first_event_at: datetime,
    triggered_at: datetime,
    last_event_at: datetime,
    event_count: int,
    details: dict | None = None,
) -> RiskIncident:
    incident = db.execute(
        select(RiskIncident).where(
            RiskIncident.rule_code == rule_code,
            RiskIncident.subject_type == subject_type,
            RiskIncident.subject_id == subject_id,
            RiskIncident.window_key == window_key,
        )
    ).scalar_one_or_none()
    if incident is None:
        incident = RiskIncident(
            rule_code=rule_code,
            event_type=event_type,
            subject_type=subject_type,
            subject_id=subject_id,
            window_key=window_key,
            window_start=window_start,
            window_end=window_end,
            first_event_at=first_event_at,
            triggered_at=triggered_at,
            last_event_at=last_event_at,
            event_count=event_count,
            status="open",
            details=details,
        )
        # 并发首次命中可能同时插入；保存点只回滚重复 incident，不丢行为流水。
        try:
            with db.begin_nested():
                db.add(incident)
                db.flush()
        except IntegrityError:
            incident = db.execute(
                select(RiskIncident).where(
                    RiskIncident.rule_code == rule_code,
                    RiskIncident.subject_type == subject_type,
                    RiskIncident.subject_id == subject_id,
                    RiskIncident.window_key == window_key,
                )
            ).scalar_one()
    else:
        reopened = incident.status == "resolved" and incident.action_reason in (
            AUTO_RESOLVED_REASON,
            MANUAL_RESET_REASON,
        )
        if reopened:
            incident.status = "open"
            incident.action_reason = None
            incident.handled_by = None
            incident.handled_at = None
            # 同一自然窗口内重置后会复用唯一 incident；明细窗口必须同步切到新基线，
            # 否则展开时会把重置前的旧流水也混进来。
            incident.window_start = window_start
            incident.window_end = window_end
        incident.first_event_at = first_event_at
        if incident.status == "open":
            incident.triggered_at = triggered_at
        incident.last_event_at = last_event_at
        incident.event_count = event_count
        # 已忽略/封禁的事件只更新事实数据，不重新打开。
        if details:
            incident.details = {**(incident.details or {}), **details}
    return incident


def evaluate_behavior_rule(
    db: Session, *, rule_code: str, subject_id: str, at: datetime
) -> RiskIncident | None:
    spec = RULES[rule_code]
    threshold = get_rule_threshold(db, rule_code)
    window_key, window_start, end = _window_bounds(at, spec.window)
    reset_at = get_rule_reset_at(db, rule_code)
    start = max(window_start, reset_at) if reset_at else window_start
    count, first_at, last_at, triggered_at = _event_stats(
        db,
        spec,
        subject_id=subject_id,
        start=start,
        end=end,
        threshold=threshold,
    )
    if count < threshold or first_at is None or last_at is None or triggered_at is None:
        return None
    return _upsert_incident(
        db,
        rule_code=spec.code,
        event_type=spec.event_type,
        subject_type=spec.subject_type,
        subject_id=subject_id,
        window_key=window_key,
        window_start=start,
        window_end=end,
        first_event_at=first_at,
        triggered_at=triggered_at,
        last_event_at=last_at,
        event_count=count,
    )


def reconcile_behavior_rule(
    db: Session,
    *,
    rule_code: str,
    at: datetime | None = None,
    commit: bool = True,
) -> int:
    """按当前阈值重算短信/一键登录当前窗口，并收起已不再命中的待处理告警。"""
    spec = RULES[rule_code]
    current = at or utcnow()
    threshold = get_rule_threshold(db, rule_code)
    window_key, window_start, end = _window_bounds(current, spec.window)
    reset_at = get_rule_reset_at(db, rule_code)
    start = max(window_start, reset_at) if reset_at else window_start
    filters = (
        BehaviorEvent.event_type == spec.event_type,
        BehaviorEvent.subject_type == spec.subject_type,
        BehaviorEvent.outcome.in_(spec.count_outcomes),
        BehaviorEvent.occurred_at >= start,
        BehaviorEvent.occurred_at < end,
    )
    qualifying_rows = db.execute(
        select(
            BehaviorEvent.subject_id,
            func.max(BehaviorEvent.occurred_at),
        )
        .where(*filters)
        .group_by(BehaviorEvent.subject_id)
        .having(func.count(BehaviorEvent.id) >= threshold)
    ).all()
    qualifying = {str(subject_id) for subject_id, _ in qualifying_rows}
    for subject_id, last_at in qualifying_rows:
        evaluate_behavior_rule(
            db,
            rule_code=rule_code,
            subject_id=str(subject_id),
            at=last_at or current,
        )

    open_incidents = db.scalars(
        select(RiskIncident).where(
            RiskIncident.rule_code == rule_code,
            RiskIncident.window_key == window_key,
            RiskIncident.status == "open",
        )
    ).all()
    for incident in open_incidents:
        if incident.subject_id not in qualifying:
            incident.status = "resolved"
            incident.action_reason = AUTO_RESOLVED_REASON
            incident.handled_by = None
            incident.handled_at = utcnow()
    if commit:
        db.commit()
    return len(qualifying)


def record_behavior_event(
    db: Session,
    *,
    event_type: str,
    subject_type: str,
    subject_id: str,
    user_id: int | None = None,
    device_id: str | None = None,
    device_model: str | None = None,
    phone: str | None = None,
    client_ip: str | None = None,
    outcome: str = "success",
    reason: str | None = None,
    details: dict | None = None,
    occurred_at: datetime | None = None,
    evaluate_rule: str | None = None,
    commit: bool = True,
) -> BehaviorEvent:
    at = occurred_at or utcnow()
    event = BehaviorEvent(
        event_type=event_type,
        subject_type=subject_type,
        subject_id=subject_id,
        user_id=user_id,
        device_id=device_id,
        device_model=device_model,
        phone=phone,
        client_ip=client_ip,
        outcome=outcome,
        reason=reason,
        details=details,
        occurred_at=at,
    )
    db.add(event)
    db.flush()
    if evaluate_rule:
        evaluate_behavior_rule(db, rule_code=evaluate_rule, subject_id=subject_id, at=at)
    if commit:
        db.commit()
        db.refresh(event)
    return event


def sync_compare_incident(
    db: Session,
    *,
    user_id: int,
    at: datetime,
    threshold: int | None = None,
    commit: bool = True,
) -> RiskIncident | None:
    effective_threshold = threshold or get_rule_threshold(db, RULE_COMPARE_DAILY)
    # comparison_record 的既有写入口统一落“北京时间 naive”时间；这里必须沿用同一
    # 口径，否则 SQLite/PG session timezone 不同时会把凌晨记录算到前一天。
    local = at.astimezone(CN_TZ).replace(tzinfo=None) if at.tzinfo else at
    window_start = local.replace(hour=0, minute=0, second=0, microsecond=0)
    end = window_start + timedelta(days=1)
    window_key = window_start.strftime("%Y-%m-%d")
    reset_at = get_rule_reset_at(db, RULE_COMPARE_DAILY)
    reset_local = (
        reset_at.astimezone(CN_TZ).replace(tzinfo=None) if reset_at else None
    )
    start = max(window_start, reset_local) if reset_local else window_start
    filters = (
        ComparisonRecord.user_id == user_id,
        ComparisonRecord.created_at >= start,
        ComparisonRecord.created_at < end,
    )
    count, first_at, last_at = db.execute(
        select(
            func.count(ComparisonRecord.id),
            func.min(ComparisonRecord.created_at),
            func.max(ComparisonRecord.created_at),
        ).where(*filters)
    ).one()
    if int(count or 0) < effective_threshold or first_at is None or last_at is None:
        return None
    triggered_at = db.execute(
        select(ComparisonRecord.created_at)
        .where(*filters)
        .order_by(ComparisonRecord.created_at.asc(), ComparisonRecord.id.asc())
        .offset(effective_threshold - 1)
        .limit(1)
    ).scalar_one()
    incident = _upsert_incident(
        db,
        rule_code=RULE_COMPARE_DAILY,
        event_type="compare_start",
        subject_type="user",
        subject_id=str(user_id),
        window_key=window_key,
        window_start=start,
        window_end=end,
        first_event_at=first_at,
        triggered_at=triggered_at,
        last_event_at=last_at,
        event_count=int(count),
    )
    if commit:
        db.commit()
        db.refresh(incident)
    return incident


def reconcile_compare_rule(
    db: Session, *, at: datetime | None = None, commit: bool = True
) -> int:
    """按当前阈值重算北京时间当日比价告警，并收起不再命中的待处理告警。"""
    current = (at or utcnow()).astimezone(CN_TZ).replace(tzinfo=None)
    window_start = current.replace(hour=0, minute=0, second=0, microsecond=0)
    end = window_start + timedelta(days=1)
    window_key = window_start.strftime("%Y-%m-%d")
    reset_at = get_rule_reset_at(db, RULE_COMPARE_DAILY)
    reset_local = (
        reset_at.astimezone(CN_TZ).replace(tzinfo=None) if reset_at else None
    )
    start = max(window_start, reset_local) if reset_local else window_start
    threshold = get_rule_threshold(db, RULE_COMPARE_DAILY)
    rows = db.execute(
        select(
            ComparisonRecord.user_id,
            func.max(ComparisonRecord.created_at),
        )
        .where(
            ComparisonRecord.user_id.is_not(None),
            ComparisonRecord.created_at >= start,
            ComparisonRecord.created_at < end,
        )
        .group_by(ComparisonRecord.user_id)
        .having(func.count(ComparisonRecord.id) >= threshold)
    ).all()
    qualifying = {str(user_id) for user_id, _ in rows}
    for user_id, last_at in rows:
        sync_compare_incident(
            db,
            user_id=int(user_id),
            at=last_at or current,
            threshold=threshold,
            commit=False,
        )

    open_incidents = db.scalars(
        select(RiskIncident).where(
            RiskIncident.rule_code == RULE_COMPARE_DAILY,
            RiskIncident.window_key == window_key,
            RiskIncident.status == "open",
        )
    ).all()
    for incident in open_incidents:
        if incident.subject_id not in qualifying:
            incident.status = "resolved"
            incident.action_reason = AUTO_RESOLVED_REASON
            incident.handled_by = None
            incident.handled_at = utcnow()
    if commit:
        db.commit()
    return len(qualifying)


def get_active_restriction(
    db: Session, *, subject_type: str, subject_id: str, scope: str
) -> SubjectRestriction | None:
    return db.execute(
        select(SubjectRestriction).where(
            SubjectRestriction.subject_type == subject_type,
            SubjectRestriction.subject_id == subject_id,
            SubjectRestriction.scope.in_((scope, "all")),
            SubjectRestriction.active.is_(True),
        )
    ).scalar_one_or_none()


def is_restricted(db: Session, *, subject_type: str, subject_id: str, scope: str) -> bool:
    return get_active_restriction(
        db, subject_type=subject_type, subject_id=subject_id, scope=scope
    ) is not None


def activate_restriction(
    db: Session,
    *,
    subject_type: str,
    subject_id: str,
    scope: str,
    reason: str,
    incident_id: int | None,
    admin_id: int,
) -> SubjectRestriction:
    restriction = db.execute(
        select(SubjectRestriction).where(
            SubjectRestriction.subject_type == subject_type,
            SubjectRestriction.subject_id == subject_id,
            SubjectRestriction.scope == scope,
        )
    ).scalar_one_or_none()
    if restriction is None:
        restriction = SubjectRestriction(
            subject_type=subject_type,
            subject_id=subject_id,
            scope=scope,
        )
        db.add(restriction)
    restriction.active = True
    restriction.reason = reason
    restriction.incident_id = incident_id
    restriction.created_by = admin_id
    restriction.created_at = utcnow()
    restriction.revoked_by = None
    restriction.revoked_at = None
    db.flush()
    return restriction


def revoke_restriction(
    db: Session, *, restriction: SubjectRestriction, admin_id: int
) -> None:
    restriction.active = False
    restriction.revoked_by = admin_id
    restriction.revoked_at = utcnow()
    db.flush()
