shaguabijia-app-server/app/models/risk.py

"""通用行为事件、风险事件与主体限制。

这三张表不是“风控监控页面专用表”：

- ``behavior_event`` 保存服务端权威的关键行为流水，后续可复用于安全审计、
  漏斗核查和客诉排查；
- ``risk_incident`` 保存规则命中后的可处置事件，负责“待处理/忽略/封禁/解除”
  生命周期；
- ``subject_restriction`` 保存当前生效的主体限制，统一承载设备、账号等主体的
  业务拦截状态。
"""
from __future__ import annotations

from datetime import datetime

from sqlalchemy import (
    JSON,
    Boolean,
    DateTime,
    Index,
    Integer,
    String,
    UniqueConstraint,
    func,
    true,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column

from app.db.base import Base

_JSON = JSON().with_variant(JSONB(), "postgresql")


class BehaviorEvent(Base):
    """服务端权威行为流水；只增不改。"""

    __tablename__ = "behavior_event"
    __table_args__ = (
        Index("ix_behavior_event_type_time", "event_type", "occurred_at"),
        Index(
            "ix_behavior_event_subject_time",
            "subject_type",
            "subject_id",
            "occurred_at",
        ),
        Index("ix_behavior_event_user_time", "user_id", "occurred_at"),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    event_type: Mapped[str] = mapped_column(String(64), nullable=False)
    subject_type: Mapped[str] = mapped_column(String(32), nullable=False)
    subject_id: Mapped[str] = mapped_column(String(128), nullable=False)

    user_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
    device_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
    device_model: Mapped[str | None] = mapped_column(String(128), nullable=True)
    phone: Mapped[str | None] = mapped_column(String(20), nullable=True)
    client_ip: Mapped[str | None] = mapped_column(String(64), nullable=True)

    # success / failed / rejected / attempted
    outcome: Mapped[str] = mapped_column(String(24), nullable=False, default="success")
    reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
    details: Mapped[dict | None] = mapped_column(_JSON, nullable=True)

    occurred_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, index=True
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )


class RiskIncident(Base):
    """规则命中后形成的可处置风险事件。"""

    __tablename__ = "risk_incident"
    __table_args__ = (
        UniqueConstraint(
            "rule_code",
            "subject_type",
            "subject_id",
            "window_key",
            name="uq_risk_incident_rule_subject_window",
        ),
        Index("ix_risk_incident_rule_status", "rule_code", "status", "triggered_at"),
        Index(
            "ix_risk_incident_subject",
            "subject_type",
            "subject_id",
            "triggered_at",
        ),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    rule_code: Mapped[str] = mapped_column(String(64), nullable=False)
    event_type: Mapped[str] = mapped_column(String(64), nullable=False)
    subject_type: Mapped[str] = mapped_column(String(32), nullable=False)
    subject_id: Mapped[str] = mapped_column(String(128), nullable=False)
    window_key: Mapped[str] = mapped_column(String(64), nullable=False)

    window_start: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    window_end: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    first_event_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    triggered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    last_event_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    event_count: Mapped[int] = mapped_column(Integer, nullable=False)

    # open / ignored / blocked / resolved
    status: Mapped[str] = mapped_column(String(24), nullable=False, default="open")
    action_reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
    handled_by: Mapped[int | None] = mapped_column(Integer, nullable=True)
    handled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    details: Mapped[dict | None] = mapped_column(_JSON, nullable=True)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),
        nullable=False,
    )


class SubjectRestriction(Base):
    """主体当前限制状态；同一主体同一作用域只有一条，可封禁后再解除/重启。"""

    __tablename__ = "subject_restriction"
    __table_args__ = (
        UniqueConstraint(
            "subject_type",
            "subject_id",
            "scope",
            name="uq_subject_restriction_subject_scope",
        ),
        Index(
            "ix_subject_restriction_lookup",
            "subject_type",
            "subject_id",
            "scope",
            "active",
        ),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    subject_type: Mapped[str] = mapped_column(String(32), nullable=False)
    subject_id: Mapped[str] = mapped_column(String(128), nullable=False)
    # auth_device / economic_account / all
    scope: Mapped[str] = mapped_column(String(32), nullable=False)
    active: Mapped[bool] = mapped_column(
        Boolean, nullable=False, default=True, server_default=true()
    )
    reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
    incident_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
    created_by: Mapped[int | None] = mapped_column(Integer, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )
    revoked_by: Mapped[int | None] = mapped_column(Integer, nullable=True)
    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    details: Mapped[dict | None] = mapped_column(_JSON, nullable=True)