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

"""Admin 后台:管理员账号 + 操作审计日志。

与 App 用户(user 表)完全隔离:admin 走独立 JWT secret、独立鉴权链(见 app/admin/)。
- admin_user:账号密码(bcrypt)登录,带角色(super_admin / finance / operator)。
- admin_audit_log:每个写操作落一条,记前后值,不可删,用于追溯"谁在何时改了谁的钱/状态"。
  admin_username / target_id 冗余存字符串,即使关联对象被删/改名也能追溯。
"""
from __future__ import annotations

from datetime import datetime

from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column

from app.db.base import Base

# PG 用 JSONB,SQLite(本地/测试)退化为通用 JSON(同 comparison_record.raw_payload)。
_JSON = JSON().with_variant(JSONB(), "postgresql")


class AdminUser(Base):
    __tablename__ = "admin_user"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
    password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
    # super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘)
    role: Mapped[str] = mapped_column(String(20), nullable=False, default="operator")
    # active / disabled
    status: Mapped[str] = mapped_column(String(20), nullable=False, default="active")

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

    def __repr__(self) -> str:  # pragma: no cover
        return f"<AdminUser id={self.id} username={self.username} role={self.role}>"


class AdminAuditLog(Base):
    __tablename__ = "admin_audit_log"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    admin_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("admin_user.id"), index=True, nullable=False
    )
    # 冗余存操作者用户名:admin 改名/禁用后仍可追溯是谁干的
    admin_username: Mapped[str] = mapped_column(String(64), nullable=False)
    # 操作类型,如 user.coins.grant / user.status.set / withdraw.refresh / feedback.handle
    action: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
    # 被操作对象类型 + id(id 用字符串以兼容 out_bill_no 等非整型主键)
    target_type: Mapped[str] = mapped_column(String(32), nullable=False)
    target_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
    # 上下文 + 前后值,如 {"amount": 1000, "reason": "...", "before": {...}, "after": {...}}
    detail: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
    ip: Mapped[str | None] = mapped_column(String(64), nullable=True)

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

    def __repr__(self) -> str:  # pragma: no cover
        return (
            f"<AdminAuditLog id={self.id} admin={self.admin_username} "
            f"action={self.action} target={self.target_type}:{self.target_id}>"
        )