diff --git a/tests/test_account_overview.py b/tests/test_account_overview.py index bef8fe1..dfcb93f 100644 --- a/tests/test_account_overview.py +++ b/tests/test_account_overview.py @@ -1371,3 +1371,93 @@ class TestOverviewWritesBindStatus: assert self.account.role_bind_status == "bound" assert self.account.esports_bind_status == "unknown", "推不出来应保持原值" assert self.account.xpd_bind_status == "unknown" + + +class TestDuplicateEventIsQuiet: + """撞唯一键 = 幂等命中,是**预期内的正常情况**,不该刷 WARNING。 + + 踩过的坑:`emit()` 不做预检查、直接 INSERT,靠唯一索引去重;撞了就 + 打 WARNING 并把**整条 SQL + 参数**dump 出来。而一键查询会把官方兑换记录 + 补成事件、同一个账号会被反复查询 —— 于是每查一次就刷一片 WARNING: + + WARNING | 写入账号事件失败(account=60 exchange): + (1062, "Duplicate entry 'official_exchange:elite:60:...' ...") + [SQL: INSERT INTO account_events (...) VALUES (...)] + [parameters: {...}] + + 数据没错(重复的被挡住了),但日志被淹、运维以为出事了(用户就是因此来问的)。 + """ + + def setup_method(self): + self.engine = get_test_engine() + self.session = make_test_session(self.engine) + self.user = User(username="dup-admin", password_hash="h", role="super_admin") + self.session.add(self.user) + self.session.flush() + self.account = Account( + username="dup-acc", password="p", email="d@e.com", email_password="ep" + ) + self.session.add(self.account) + self.session.commit() + + def teardown_method(self): + self.session.close() + + def _emit_once(self): + from web.backend.services.account_events_service import emit + + return emit( + self.session, + account_id=self.account.id, + event_type="exchange", + title="兑换 某商品", + category="business", + source="official_sync", + source_ref="official_exchange:elite:dup:commodity:20261001", + ) + + def test_duplicate_returns_false_without_warning(self): + from loguru import logger + + assert self._emit_once() is True + self.session.commit() + + warnings: list[str] = [] + sink_id = logger.add(lambda msg: warnings.append(str(msg)), level="WARNING") + try: + assert self._emit_once() is False, "重复写入应返回 False" + finally: + logger.remove(sink_id) + + assert not [w for w in warnings if "写入账号事件失败" in w], ( + "撞唯一键是正常的幂等命中,不该打 WARNING(日志会被淹,且误导运维)" + ) + + def test_duplicate_does_not_write_twice(self): + from web.backend.services import account_events_service as svc + + self._emit_once() + self.session.commit() + self._emit_once() + self.session.commit() + assert ( + self.session.query(AccountEvent) + .filter(AccountEvent.event_type == svc.EVENT_EXCHANGE) + .count() + == 1 + ) + + def test_duplicate_helper_only_matches_1062(self): + """只有**确认**是唯一键冲突才降级;其它完整性错误仍要告警。""" + from web.backend.services.account_events_service import _is_duplicate_entry + + class _Orig(Exception): + pass + + dup = Exception("x") + dup.orig = _Orig(1062, "Duplicate entry 'a' for key 'k'") # type: ignore[attr-defined] + assert _is_duplicate_entry(dup) is True + + other = Exception("x") + other.orig = _Orig(1048, "Column 'a' cannot be null") # type: ignore[attr-defined] + assert _is_duplicate_entry(other) is False diff --git a/web/backend/services/account_events_service.py b/web/backend/services/account_events_service.py index 5da64b7..9b74b80 100644 --- a/web/backend/services/account_events_service.py +++ b/web/backend/services/account_events_service.py @@ -36,6 +36,7 @@ from typing import Any from loguru import logger from sqlalchemy import text +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session # ── 事件类型 ────────────────────────────────────────────── @@ -74,6 +75,18 @@ 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**(本服务所有时间列的落库形态)。 @@ -208,6 +221,22 @@ def emit( 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