yml2213
6e729e7a80
时间线从此覆盖**全部**业务事实，`douyu_tasks` 也有了可控的增长边界。
## 一、删掉收集器（它必然漏数据）
`jobs/account_events_collect.py` 连同 `_task_to_event` / `COLLECTED_TASK_TYPES` /
水位线一起删除，迁移 `20261004_0043` 清掉它的作业行。删除理由是实测出来的：
- **靠 allowlist 认业务类型，新增类型就静默丢**：原有名单只有 5 个，
而 `exchange_esports_goods`（电竞兑换）、`exchange_xpd_goods`（和平小店兑换）、
`pay_locked_order`（锁单支付）、`donate_esports_*`（电竞礼物积分）**从来没进过时间线**；
- **「任务 success」≠「业务真的发生了」**（见下）；
- **字段路径靠抽样猜**：二维码充值的 `result` 里根本没有金额，收集器恒写
`amount=None`，`recharge_total` 系统性少计；
- 作业默认 `enabled=False`，生产上一直没跑。
历史**不回填**（既有结论不变）：需要历史走斗鱼侧权威接口
（`query_exchange_records` / `query_xpd_purchase_records` / `query_points` 都是现成任务类型）。
## 二、业务事实直发：命名构造函数
新增 `services/business_events.py`：`record_recharge` / `record_exchange` /
`record_points_credit` / `record_bind`。每类事实的字段、标题、金额符号只定义一次
（与 S4 的 `record_account_action`、`on_ck_verdict` 同一套路）。
**9 处成功点接线**：`exchange_goods` 真兑换分支、`exchange_esports_goods`、
`exchange_xpd_goods`、`pay_locked_order`、`create_gold_qr`（二维码 + 供应商轮询 +
**供应商回调**三条路径）、`donate_elite_gift`、`donate_esports_gift`、
`confirm_bind`、`confirm_esports_bind`。
**只记真实事实**（按你的决定）：`exchange_goods` 里「预约到货 / 已预兑 / 预兑确认」
三条 success 分支**不发事件** —— 它们不消耗积分也没换到东西，旧收集器只按 status
收集，于是每预约一次就生成一条金额为空的 `exchange` 把 `exchange_count` 灌大。
`confirm_bind` 里「检测到账号本来就已绑定」那条也跳过。
## 三、修掉两个真 bug（都是钱账）
1. **到账判据 `!=` → `>`**（`douyu_runner_gold.py`）：余额**下降**也算「鱼翅已到账」。
批内并发（最高 10）时同账号送礼会消耗鱼翅，余额一变就被充值任务读成到账，
**往台账记一笔根本没发生的充值**。
2. **供应商回调晚于 runner 超时**：原来守卫是「任务状态没被改过才更新」，
于是 runner 先判 failed、回调随后说成功的那笔真金白银**永远不进时间线**。
改成按**回调自己的终态**判断，并补发事件。
## 四、发射顺序：一个会静默丢事件的坑
`_mark_task` 内部 `db.commit()`，而 `emit` 用 `db.begin_nested()` **只 RELEASE
SAVEPOINT、不提交外层事务**；runner 的 `finally` 只 `close()`（未提交即回滚）。
所以**在 `_mark_task` 之后发事件会被静默丢弃，且 `emit` 仍返回 True**。
所有接线一律紧贴 `_mark_task(success)` **之前**，并加了源码扫描用例钉住
（已反向验证：故意挪到之后，用例立刻报出 file:line）。
## 五、`douyu_tasks` 保留策略（新增作业，默认关闭）
`jobs/douyu_tasks_retention.py`，两档：
| 档 | 保留 |
| --- | --- |
| 噪声查询（只读看一眼就走，且前端不拿它派生账号状态） | **7 天** |
| 其余一切（业务事实 + 状态派生依赖的查询 + **以后新增的类型**） | **90 天** |
长保留档用 `task_type NOT IN (噪声名单)` 表达 → **新增类型默认长保留**，不会因为
「忘了加名单」被误删（正是收集器翻车的反面）。在途任务（planned/pending/running）
**绝不删**；分批删避免长锁；两个窗口都能在作业 `config` 里改，不必发版。
⚠️ 「前端从任务列表反推账号当前状态」那几类查询（`query_xpd_bind_info` /
`query_esports_game_name` 等）**故意留在长保留档** —— 删早了工作台账号行会变
「未查」，那是把**当前显示**搞坏。根治要靠 §3.6 的状态列拆解（后续阶段）。
作业默认 `enabled=False`，由运维决定何时打开。
## 验证
- `TZ=Asia/Shanghai`：**450 passed / 6 skipped**（S4 后 426 → 净增 24）
- `TZ=UTC`（相关文件）101 passed；ruff check/format 全绿
- 前端 build 通过、lint 0 errors
- 真实开发库：迁移已应用（head `20261004_0043`，孤儿作业行已清）；
保留作业实跑确认「作业行默认关闭、SQL 在 MySQL 上正确、近期行不误删」
- **两处反向验证**：把 `record_bind` 挪到 `_mark_task` 之后 → 顺序用例报红；
修掉 `registered_jobs()` 那个**假绿用例**（它返回 dict，原来的集合推导式
遍历空字典恒成立），并在断言里加了反向确认
未做（后续）：S6 账号页重构 + `bind_status` 三列拆解（做完才能把状态派生类查询
也纳入短保留）、S7 账号检测改账号维度、S8 丢 `login_tasks` 旧凭据列。
346 lines
13 KiB
Python
346 lines
13 KiB
Python
"""S5：业务事实直发事件。
|
||
|
||
取代了原来的「收集器事后扫 `douyu_tasks.result` 反推事件」。本文件守住三件事：
|
||
|
||
1. **命名构造函数的字段口径**（金额符号、detail、subject）；
|
||
2. **幂等**：`source_ref=task_ref(task_id)` —— 让直发与仍在跑的收集器能互相去重，
|
||
也让「runner 轮询」与「供应商回调」不会对同一笔充值重复记账；
|
||
3. **发射顺序**：事件必须在 `_mark_task` **之前**发。`_mark_task` 内部 commit，
|
||
而 `emit` 只 RELEASE SAVEPOINT、不提交外层事务，runner 的 `finally` 只 close
|
||
（未提交即回滚）—— 发在后面会被静默丢弃，且 `emit` 仍返回 True。
|
||
"""
|
||
|
||
from datetime import UTC, datetime
|
||
|
||
from tests.support.db import get_test_engine, make_test_session
|
||
from web.backend.models import Account, AccountEvent, DouyuTask, User
|
||
from web.backend.services import account_events_service as svc
|
||
from web.backend.services import business_events as biz
|
||
|
||
NOW = datetime(2026, 10, 4, 12, 0, 0, tzinfo=UTC)
|
||
|
||
|
||
class _Base:
|
||
def setup_method(self):
|
||
self.engine = get_test_engine()
|
||
self.session = make_test_session(self.engine)
|
||
self.session.query(AccountEvent).delete()
|
||
self.user = User(username="biz-admin", password_hash="h", role="super_admin")
|
||
self.session.add(self.user)
|
||
self.session.flush()
|
||
self.account = Account(
|
||
username="biz-acc", password="p", email="b@e.com", email_password="ep"
|
||
)
|
||
self.session.add(self.account)
|
||
self.session.commit()
|
||
|
||
def teardown_method(self):
|
||
self.session.close()
|
||
|
||
def events(self, event_type: str) -> list[AccountEvent]:
|
||
return (
|
||
self.session.query(AccountEvent)
|
||
.filter(
|
||
AccountEvent.account_id == self.account.id,
|
||
AccountEvent.event_type == event_type,
|
||
)
|
||
.order_by(AccountEvent.id.asc())
|
||
.all()
|
||
)
|
||
|
||
|
||
class TestConstructors(_Base):
|
||
def test_recharge_records_amount_and_channel(self):
|
||
assert (
|
||
biz.record_recharge(
|
||
self.session,
|
||
self.account,
|
||
task_id=101,
|
||
amount=100,
|
||
channel="wechat_qr",
|
||
douyu_uid="u1",
|
||
)
|
||
is True
|
||
)
|
||
self.session.commit()
|
||
events = self.events(svc.EVENT_RECHARGE)
|
||
assert len(events) == 1
|
||
event = events[0]
|
||
assert event.amount == 100
|
||
assert event.category == svc.CATEGORY_BUSINESS
|
||
assert event.source_ref == "douyu_task:101"
|
||
assert event.detail["channel"] == "wechat_qr"
|
||
assert event.detail["confirmed"] is True
|
||
|
||
def test_recharge_without_amount_still_records(self):
|
||
"""金额拿不到时宁可记一条金额为空的事件，也不要凭空编一个数字。"""
|
||
biz.record_recharge(self.session, self.account, task_id=102, amount=None)
|
||
self.session.commit()
|
||
event = self.events(svc.EVENT_RECHARGE)[0]
|
||
assert event.amount is None
|
||
|
||
def test_recharge_marks_supplier_confirmation_honestly(self):
|
||
"""供应商终态成功 ≠ 斗鱼余额已变，必须如实标注。"""
|
||
biz.record_recharge(
|
||
self.session,
|
||
self.account,
|
||
task_id=103,
|
||
amount=50,
|
||
channel="supplier_api",
|
||
confirmed=False,
|
||
)
|
||
self.session.commit()
|
||
event = self.events(svc.EVENT_RECHARGE)[0]
|
||
assert event.detail["confirmed"] is False
|
||
assert "供应商回执" in event.title
|
||
|
||
def test_exchange_amount_is_negative_and_subject_is_commodity(self):
|
||
biz.record_exchange(
|
||
self.session,
|
||
self.account,
|
||
task_id=201,
|
||
name="兔兔白日梦",
|
||
score=1200,
|
||
commodity_id="sku-9",
|
||
points_after=800,
|
||
scope="elite",
|
||
)
|
||
self.session.commit()
|
||
event = self.events(svc.EVENT_EXCHANGE)[0]
|
||
assert event.amount == -1200
|
||
assert event.balance_after == 800
|
||
assert event.subject_type == "commodity"
|
||
assert event.subject_key == "sku-9"
|
||
assert event.detail["scope"] == "elite"
|
||
|
||
def test_exchange_without_score_keeps_amount_null(self):
|
||
biz.record_exchange(
|
||
self.session, self.account, task_id=202, name="x", score=None
|
||
)
|
||
self.session.commit()
|
||
assert self.events(svc.EVENT_EXCHANGE)[0].amount is None
|
||
|
||
def test_points_credit_marks_unconfirmed(self):
|
||
"""未确认到账也记，但必须标出来 —— 不能把「预期」写成「已到账」。"""
|
||
biz.record_points_credit(
|
||
self.session,
|
||
self.account,
|
||
task_id=301,
|
||
gain=980,
|
||
points_after=100,
|
||
confirmed=False,
|
||
)
|
||
self.session.commit()
|
||
event = self.events(svc.EVENT_POINTS_CREDIT)[0]
|
||
assert event.amount == 980
|
||
assert event.detail["confirmed"] is False
|
||
assert "未确认" in event.title
|
||
|
||
def test_points_credit_confirmed(self):
|
||
biz.record_points_credit(
|
||
self.session, self.account, task_id=302, gain=10, confirmed=True
|
||
)
|
||
self.session.commit()
|
||
event = self.events(svc.EVENT_POINTS_CREDIT)[0]
|
||
assert event.detail["confirmed"] is True
|
||
assert "未确认" not in event.title
|
||
|
||
def test_bind_records_scope_and_rebind(self):
|
||
biz.record_bind(
|
||
self.session,
|
||
self.account,
|
||
task_id=401,
|
||
scope="esports",
|
||
role_name="小明",
|
||
rebind=True,
|
||
)
|
||
self.session.commit()
|
||
event = self.events(svc.EVENT_BIND_OK)[0]
|
||
assert event.subject_type == "esports_role"
|
||
assert event.subject_key == "小明"
|
||
assert event.detail["rebind"] is True
|
||
assert "换绑" in event.title
|
||
|
||
def test_bind_weak_is_flagged(self):
|
||
biz.record_bind(
|
||
self.session,
|
||
self.account,
|
||
task_id=402,
|
||
scope="elite",
|
||
role_name="小华",
|
||
weak=True,
|
||
)
|
||
self.session.commit()
|
||
event = self.events(svc.EVENT_BIND_OK)[0]
|
||
assert event.detail["confirmed"] is False
|
||
assert "未回查确认" in event.title
|
||
|
||
|
||
class TestIdempotency(_Base):
|
||
"""同一个 task 只能对应一条事实 —— 这是直发与收集器/回调共存的基石。"""
|
||
|
||
def test_same_task_id_does_not_duplicate(self):
|
||
biz.record_recharge(self.session, self.account, task_id=500, amount=10)
|
||
self.session.commit()
|
||
# 模拟「另一个路径也观测到同一笔」（收集器重跑 / 供应商回调）
|
||
biz.record_recharge(self.session, self.account, task_id=500, amount=10)
|
||
self.session.commit()
|
||
assert len(self.events(svc.EVENT_RECHARGE)) == 1
|
||
|
||
def test_different_task_id_is_a_new_fact(self):
|
||
biz.record_recharge(self.session, self.account, task_id=501, amount=10)
|
||
self.session.commit()
|
||
biz.record_recharge(self.session, self.account, task_id=502, amount=10)
|
||
self.session.commit()
|
||
assert len(self.events(svc.EVENT_RECHARGE)) == 2
|
||
|
||
def test_task_ref_matches_old_collector_key(self):
|
||
"""必须与旧收集器的 key 完全一致，否则迁移窗口内会重复记账。"""
|
||
assert biz.task_ref(123) == "douyu_task:123"
|
||
|
||
|
||
class TestEmitSurvivesWorkerSessionClose(_Base):
|
||
"""发射顺序的**行为验证**：事件要能挺过 runner 的 `finally: worker_db.close()`。
|
||
|
||
runner 的每个任务在 `_execute_one` 的 finally 里只 close 不 commit。
|
||
`_mark_task` 内部有 commit，所以「紧贴 _mark_task 之前 emit」会被那次 commit
|
||
一起提交；若把 emit 放在 _mark_task 之后，它就落在一个没人提交的事务里，
|
||
close 时被回滚 —— 而且 `emit` 照样返回 True，没有任何日志。
|
||
"""
|
||
|
||
def test_event_committed_together_with_mark_task(self):
|
||
# 模拟 runner：emit 后由 _mark_task 的 commit 收尾
|
||
biz.record_recharge(self.session, self.account, task_id=600, amount=7)
|
||
self.session.commit() # 等价于 _mark_task 内部的 commit
|
||
self.session.close() # 等价于 runner 的 finally
|
||
|
||
fresh = make_test_session(self.engine)
|
||
try:
|
||
rows = (
|
||
fresh.query(AccountEvent)
|
||
.filter(AccountEvent.source_ref == "douyu_task:600")
|
||
.all()
|
||
)
|
||
assert len(rows) == 1, "事件没有挺过 session close —— 发射顺序错了"
|
||
finally:
|
||
fresh.close()
|
||
|
||
def test_emit_without_commit_is_lost_on_close(self):
|
||
"""反证：不 commit 就 close，事件确实会丢（这就是顺序规则存在的理由）。"""
|
||
biz.record_recharge(self.session, self.account, task_id=601, amount=7)
|
||
self.session.close() # 故意不 commit
|
||
|
||
fresh = make_test_session(self.engine)
|
||
try:
|
||
rows = (
|
||
fresh.query(AccountEvent)
|
||
.filter(AccountEvent.source_ref == "douyu_task:601")
|
||
.all()
|
||
)
|
||
assert rows == [], "若不 commit 也能落库，说明事务边界与预期不符"
|
||
finally:
|
||
fresh.close()
|
||
|
||
|
||
class TestCollectorIsGone:
|
||
"""收集器必须被彻底删掉，不留半条命。"""
|
||
|
||
def test_module_removed(self):
|
||
import importlib
|
||
|
||
try:
|
||
importlib.import_module(
|
||
"web.backend.services.automation.jobs.account_events_collect"
|
||
)
|
||
except ModuleNotFoundError:
|
||
return
|
||
raise AssertionError("收集器模块还在")
|
||
|
||
def test_no_collector_symbols_left(self):
|
||
assert not hasattr(svc, "COLLECTED_TASK_TYPES")
|
||
assert not hasattr(svc, "_task_to_event")
|
||
|
||
def test_job_key_not_registered(self):
|
||
"""⚠️ `registered_jobs()` 返回的是 **dict**（key → JobSpec）。
|
||
|
||
写成 `{spec.key for spec in registered_jobs()}` 会遍历出字符串键、
|
||
`spec.key` 直接报错；而如果它恰好是空字典，推导式又会得到空集让断言**假绿**。
|
||
所以这里显式 register 后再按键判断。
|
||
"""
|
||
from web.backend.services.automation import (
|
||
register_default_jobs,
|
||
registered_jobs,
|
||
)
|
||
|
||
register_default_jobs()
|
||
keys = set(registered_jobs())
|
||
assert "account_events_collect" not in keys
|
||
# 反向确认这个断言不是假绿：新作业确实在里面
|
||
assert "douyu_tasks_retention" in keys
|
||
|
||
|
||
class TestEmitOrderingInRunners:
|
||
"""源码扫描：runner 里的 `record_*` 必须在 `_mark_task(..., "success")` **之前**。
|
||
|
||
这条规则很隐蔽（违反了不会报错、只是事件静默消失），所以用用例钉住。
|
||
做法：对每个 `record_*` 调用，找它之后出现的第一个 `_mark_task(`，该调用里
|
||
必须带 `"success"`；如果事件被写在 mark 之后，那么它后面第一个 `_mark_task`
|
||
要么是另一个分支的、要么根本不存在 —— 两种情况都会让用例变红。
|
||
"""
|
||
|
||
RUNNER_FILES = (
|
||
"web/backend/services/douyu_runner_goods.py",
|
||
"web/backend/services/douyu_runner_gold.py",
|
||
"web/backend/services/douyu_runner_bind.py",
|
||
"web/backend/services/douyu_runner_donate.py",
|
||
"web/backend/services/douyu_runner_xpd.py",
|
||
)
|
||
|
||
def test_record_calls_precede_success_mark(self):
|
||
import re
|
||
from pathlib import Path
|
||
|
||
root = Path(__file__).resolve().parents[1]
|
||
record_re = re.compile(r"\brecord_(recharge|exchange|points_credit|bind)\(")
|
||
offenders: list[str] = []
|
||
checked = 0
|
||
|
||
for rel in self.RUNNER_FILES:
|
||
source = (root / rel).read_text(encoding="utf-8")
|
||
for match in record_re.finditer(source):
|
||
checked += 1
|
||
tail = source[match.end() :]
|
||
mark = tail.find("_mark_task(")
|
||
line_no = source[: match.start()].count("\n") + 1
|
||
if mark == -1:
|
||
offenders.append(f"{rel}:{line_no} 之后没有 _mark_task")
|
||
continue
|
||
# 取这次 _mark_task 调用的前若干字符，判断是不是 success 分支
|
||
snippet = tail[mark : mark + 400]
|
||
if '"success"' not in snippet:
|
||
offenders.append(
|
||
f"{rel}:{line_no} 之后的第一个 _mark_task 不是 success"
|
||
)
|
||
|
||
assert checked >= 10, f"只扫到 {checked} 个 record_ 调用，扫描规则可能失效了"
|
||
assert not offenders, (
|
||
"record_* 必须在 _mark_task(success) **之前**调用"
|
||
"（否则事件会被 runner 的 close 静默回滚）：\n" + "\n".join(offenders)
|
||
)
|
||
|
||
|
||
class TestTaskReadyForApi(_Base):
|
||
def test_douyu_task_still_works(self):
|
||
"""兜底：删掉收集器不应影响 douyu_tasks 本身的读写。"""
|
||
task = DouyuTask(
|
||
batch_id="b1",
|
||
account_id=self.account.id,
|
||
task_type="exchange_goods",
|
||
status="success",
|
||
result={"commodity_name": "x"},
|
||
created_by=self.user.id,
|
||
)
|
||
self.session.add(task)
|
||
self.session.commit()
|
||
assert task.id is not None
|