yml2213
631e33a4f6
## 现象（用户从日志里发现的）
WARNING | 写入账号事件失败（account=60 exchange）：
(1062, "Duplicate entry 'official_exchange:elite:60:20260120QYOOB_bag360:...'")
[SQL: INSERT INTO account_events (...) VALUES (...)]
[parameters: {...}]
## 分析：**不是 bug，是幂等机制在工作，但日志级别错了**
`emit()` 刻意**不做预检查**，直接 INSERT，靠 `source_ref` 的唯一索引去重；
撞了就 catch（SAVEPOINT 只回滚这一条）。所以：
- **数据是对的** —— 重复的被挡住，`exchange_events_added` 只计新写入的
- **日志是错的** —— 撞唯一键被当成异常打 WARNING，还把整条 SQL + 参数 dump 出来
而"撞唯一键"在这里是**预期内的正常情况**：一键查询会把斗鱼官方兑换记录补成事件，
同一个账号会被反复查询 —— 第二次开始每条记录都会撞。一次查询能刷几十屏，
既淹掉真日志、又让人以为出事了（用户就是因此来问的）。
## 修法
区分「唯一键冲突」与「真错误」：
- 确认是 **1062**（并带 `Duplicate entry` 字样）→ 降为 **DEBUG**，一句话说明"已存在，跳过"
- 其它完整性错误（如非空约束 1048）→ **仍然是 WARNING**
刻意**不把 IntegrityError 整个静音** —— 那会把真问题一起藏起来。
新增 `_is_duplicate_entry()` 只在两种信号都对上时才降级。
## 测试（+3）
- 重复写入返回 False，且**不产生**任何 "写入账号事件失败" 的 WARNING
（用 loguru 的临时 sink 真的抓日志来断言，不是看代码）
- 重复写入不会写进第二条
- `_is_duplicate_entry` 只认 1062，1048（非空约束）仍然返回 False
274 lines
11 KiB
Python
274 lines
11 KiB
Python
"""账号事件流：写入与查询。
|
||
|
||
## 定位
|
||
|
||
`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）
|
||
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
|