live-hub-py/web/backend/services/account_events_service.py

"""账号事件流：写入与查询。

## 定位

`accounts` 存**当前状态**（积分、余额、开通标记），这张表存**发生过什么**。
账号详情页 = 状态卡（读 `accounts`）+ 时间线（读本表）。

## 唯一写入路径：直接发射

业务事实**在它发生的那一刻**由业务代码发射（`emit` 或 `business_events` 里的命名
构造函数）。曾经有过第二条路径 —— 一个作业事后扫 `douyu_tasks.result` 的 JSON、
猜字段路径反推事件。那条路已经删掉，原因是它必然漏数据：

- 靠 allowlist 认业务类型 → 新增类型没人想起维护，**静默丢**（电竞兑换、和平小店
  兑换、锁单支付、电竞礼物积分因此从来没进过时间线）；
- 「任务 success」不等于「业务真的发生了」（预约、已绑定、未确认到账都会被标 success）；
- 作业默认关闭，生产上它一直没跑。

历史**不从 `douyu_tasks` 回填**。需要历史时按斗鱼侧权威接口同步
（`query_exchange_records` / `query_xpd_purchase_records` / `query_points` 都是现成任务类型）
—— 宁可少，不可假。

## 两条硬约束

- **发射是旁路**：`emit` 任何失败都不能影响登录/兑换主流程（内部吞异常 + 记日志，
  并用 SAVEPOINT 隔离，避免污染调用方的事务）；
- **幂等**：`source_ref` 唯一键兜住「收集器重跑」「事件重复发射」。没有它，重跑一次
  时间线就多一份重复记录。
"""

from __future__ import annotations

import json
from datetime import UTC, datetime
from typing import Any

from loguru import logger
from sqlalchemy import text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

# ── 事件类型 ──────────────────────────────────────────────
# 命名保持「名词_状态」风格，便于前端归类与筛选。
EVENT_IMPORTED = "imported"  # 账号导入（读 accounts.created_at，一般不落表）
EVENT_LOGIN_OK = "login_ok"  # 登录成功（含新 CK）
#: 登录/重登**失败**（手动与批量登录都走这条）。
#:
#: 2026-10-07 补：此前只有成功会写事件，失败只改 `login_tasks` 行 ——
#: 生产实测 247 次登录失败在账号时间线上**一条痕迹都没有**（成功却有 1,456 条），
#: 「这个账号为什么一直登录不上」在账号页上无从回答。
#: ⚠️ **被停止 / 被中断不算失败**，那种情况不写（见 `services/login_failure`）。
EVENT_LOGIN_FAILED = "login_failed"
EVENT_CK_INVALID = "ck_invalid"  # CK 判定失效
EVENT_CK_CLEARED = "ck_cleared"  # CK 被清除（人工清除 / 账号删除前的清理）
EVENT_RELOGIN_OK = "relogin_ok"  # 自动重登成功
EVENT_RELOGIN_FAILED = "relogin_failed"  # 自动重登失败（转人工/退避）
EVENT_BIND_OK = "bind_ok"  # 绑定/换绑
EVENT_RECHARGE = "recharge"  # 充值到账
#: 事件来源：从斗鱼**官方兑换记录**补录（区别于任务执行时当场记的）
OFFICIAL_SYNC_SOURCE = "official_sync"

EVENT_EXCHANGE = "exchange"  # 兑换（含消耗积分）
EVENT_POINTS_CREDIT = "points_credit"  # 积分到账（含到期时间）
EVENT_POINTS_EXPIRE = "points_expire"  # 积分到期（资产减少）
EVENT_GOLD_SPENT = "gold_spent"  # 鱼翅消耗（送礼等）
EVENT_ELITE_OPENED = "elite_opened"  # 精英宝典开通
EVENT_ESPORTS_OPENED = "esports_opened"  # 电竞手册开通
EVENT_ASSIGNED = "assigned"  # 分配客服
EVENT_TAG_CHANGED = "tag_changed"  # 标签变更
EVENT_REMARK_CHANGED = "remark_changed"  # 备注变更
EVENT_STATUS_CHANGED = "status_changed"  # 生命周期状态变迁（停用/归档/失效…）

# 有符号金额的事件类型（用于页面「累计充值/累计消耗」）：
# 充值为正、兑换为负，其余不计入金额统计。
INCOME_TYPES = (EVENT_RECHARGE,)
OUTCOME_TYPES = (EVENT_EXCHANGE,)

# 粗分类：决定时间线怎么分组、以及哪些事件算「业务台账」。
CATEGORY_LIFECYCLE = "lifecycle"  # 生命周期状态变迁
CATEGORY_AUTH = "auth"  # 登录与 CK
CATEGORY_BUSINESS = "business"  # 充值 / 兑换 / 积分 / 开通
CATEGORY_OPS = "ops"  # 人工操作（分配 / 标签 / 停用）


def _is_duplicate_entry(exc: BaseException) -> bool:
    """是否 MySQL 唯一键冲突（1062）。

    只有**确认**是撞唯一键才降级成 DEBUG；其它完整性错误（例如非空约束）
    仍然是真问题，必须继续 WARNING —— 别把真错误一起静音了。
    """
    code = getattr(getattr(exc, "orig", None), "args", None)
    if code and code[0] == 1062:
        return True
    return "Duplicate entry" in str(exc)


def to_naive_utc(value: datetime) -> datetime:
    """把时间归一成 **naive UTC**（本服务所有时间列的落库形态）。

    约定与 `database.UTCDateTime` 一致，也是全仓通行写法：

    - **naive → 视为「已经是 UTC」，原样使用**；
    - aware → 换算到 UTC 后去掉 tzinfo。

    ⚠️ 绝对不能直接写 `value.astimezone(UTC)`：对 naive 值，Python 会把它当作
    **服务器本地时间**再换算。本机 UTC+8 时事件时间会被整整扣掉 8 小时，
    而 naive 值的来源很常见 —— 用**裸 SQL** 读 DATETIME 列（如收集器读
    `douyu_tasks.finished_at`）拿到的就是 naive，因为它绕过了 `UTCDateTime`
    装饰器。仓库里 `scheduler._as_utc` / `yyb_service._as_utc` / `audit.py`
    都遵守了这条约定，本函数与它们同口径。
    """
    if value.tzinfo is None:
        return value
    return value.astimezone(UTC).replace(tzinfo=None)


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


_INSERT_SQL = text(
    """
    INSERT INTO account_events (
        account_id, event_type, category, occurred_at, title,
        amount, balance_after, subject_type, subject_key,
        detail, source, actor_type, actor_id, source_ref, created_at
    ) VALUES (
        :account_id, :event_type, :category, :occurred_at, :title,
        :amount, :balance_after, :subject_type, :subject_key,
        :detail, :source, :actor_type, :actor_id, :source_ref, :created_at
    )
    """
)


def _event_params(
    *,
    account_id: int,
    event_type: str,
    title: str,
    occurred_at: datetime | None = None,
    amount: int | None = None,
    balance_after: int | None = None,
    detail: dict[str, Any] | None = None,
    source: str = "manual",
    source_ref: str | None = None,
    category: str = "",
    subject_type: str = "",
    subject_key: str = "",
    actor_type: str = "",
    actor_id: int | None = None,
) -> dict[str, Any]:
    """把一次调用归一成落库参数（`emit` 与 `emit_bulk` 共用，避免 SQL 写两遍）。"""
    return {
        "account_id": account_id,
        "event_type": event_type,
        "category": (category or "")[:16],
        "occurred_at": to_naive_utc(occurred_at or _now()),
        "title": (title or "")[:256],
        "amount": amount,
        "balance_after": balance_after,
        "subject_type": (subject_type or "")[:16],
        "subject_key": (subject_key or "")[:96],
        "detail": json.dumps(detail, ensure_ascii=False, default=str)
        if detail
        else None,
        "source": source[:24],
        "actor_type": (actor_type or "")[:16],
        "actor_id": actor_id,
        "source_ref": (source_ref or None),
        "created_at": to_naive_utc(_now()),
    }


def emit(
    db: Session,
    *,
    account_id: int,
    event_type: str,
    title: str,
    occurred_at: datetime | None = None,
    amount: int | None = None,
    balance_after: int | None = None,
    detail: dict[str, Any] | None = None,
    source: str = "manual",
    source_ref: str | None = None,
    category: str = "",
    subject_type: str = "",
    subject_key: str = "",
    actor_type: str = "",
    actor_id: int | None = None,
) -> bool:
    """记录一条账号事件。返回是否真的写入（幂等命中/异常时为 False）。

    **绝不抛异常**：事件是旁路，不能因为记不上一条时间线就让登录或兑换失败。
    用 SAVEPOINT 包住，失败只回滚这一条，不影响调用方已经做过的事。

    `occurred_at` 是**业务发生时间**，可以是 aware，也可以是 naive
    （naive 按约定视为已是 UTC，见 `to_naive_utc`）。

    `category` / `subject_*` / `actor_*` 支撑台账检索：
    例如「该账号该商品最近一次兑换」走
    `(account_id, event_type, subject_key, occurred_at)` 复合索引。
    """
    if not account_id or not event_type:
        return False

    params = _event_params(
        account_id=account_id,
        event_type=event_type,
        title=title,
        occurred_at=occurred_at,
        amount=amount,
        balance_after=balance_after,
        detail=detail,
        source=source,
        source_ref=source_ref,
        category=category,
        subject_type=subject_type,
        subject_key=subject_key,
        actor_type=actor_type,
        actor_id=actor_id,
    )

    try:
        # SAVEPOINT：即使这条插入失败（例如 source_ref 撞唯一键），
        # 也只回滚它自己，调用方事务里已完成的写入不受影响。
        with db.begin_nested():
            db.execute(_INSERT_SQL, params)
        return True
    except IntegrityError as exc:
        # 撞唯一键 = **幂等命中**，是预期内的正常情况，不是故障。
        #
        # 典型场景：一键查询会把斗鱼官方兑换记录补成事件，而同一个账号会被
        # 反复查询 —— 第二次开始每一条记录都会撞 source_ref。
        # 以前这里打 WARNING 并把**整条 SQL + 参数**dump 出来，
        # 一次查询就能刷几十屏，既淹掉真日志、又让人以为出错了（用户报过）。
        # 所以降成 DEBUG：重复是设计的一部分。
        if _is_duplicate_entry(exc):
            logger.debug(
                f"账号事件已存在，跳过（account={account_id} {event_type} "
                f"source_ref={source_ref}）"
            )
            return False
        logger.warning(f"写入账号事件失败（account={account_id} {event_type}）：{exc}")
        return False
    except Exception as exc:  # noqa: BLE001 - 旁路，任何失败都不上抛
        logger.warning(f"写入账号事件失败（account={account_id} {event_type}）：{exc}")
        return False


def emit_bulk(db: Session, events: list[dict[str, Any]]) -> int:
    """一次写多条事件，返回实际写入条数。

    用于**批量账号操作**（批量分配 / 批量改标签）：一次操作可能覆盖几千个账号，
    每个账号的时间线都该各留一条「谁在什么时候改了什么」。逐条走 `emit()` 会给
    每条开一个 SAVEPOINT，几千条时请求明显变慢 —— 这里改成一次 executemany，
    整批包在**一个** SAVEPOINT 里。

    与 `emit()` 一样**绝不抛异常**：整批失败就整批丢弃（记 warning），
    不会污染调用方的事务，也不会让「批量分配」因为记不上时间线而失败。

    每个 dict 的键与 `emit()` 的关键字参数同名（`account_id` / `event_type` /
    `title` 必填）。
    """
    rows = [
        _event_params(**{**item, "occurred_at": item.get("occurred_at")})
        for item in events
        if item.get("account_id") and item.get("event_type")
    ]
    if not rows:
        return 0

    try:
        with db.begin_nested():
            db.execute(_INSERT_SQL, rows)
        return len(rows)
    except Exception as exc:  # noqa: BLE001 - 旁路，任何失败都不上抛
        logger.warning(f"批量写入账号事件失败（{len(rows)} 条）：{exc}")
        return 0