浅念
6b99202e9b
- scheduler: yunke_analytics_tick job(与消息增量同节奏 IntervalTrigger), _analytics_tick 纯 DB——stat_daily_staff 滚动3天刷新 + 敏感词增量扫描, 不抢 _page_lock 无子进程,异常只记日志不断链;today 用 time.time() UTC epoch 喂 day_key,规避服务器非上海时区的 naive now 误判 - cli: staff-daily-stats(全量重建 rollup,幂等)+ sensitive-scan (默认增量水位扫描,--full 首跑建水位/新词回填历史;description 明示 加词不回填历史)+ 幂等补种起步词库 - full_sync_all.sh: daily-stats 后追加 staff-daily-stats(rollup 随 msg_daily_stats 重建同步自愈) - 测试: tick 行为测试(今日 seed→stat_daily_staff 出行)+ 异常吞噬 (monkeypatch 抛错不外抛);同步更新注册断言(3→4 jobs) Co-Authored-By: Claude Code <noreply@anthropic.com>
77 lines
2.9 KiB
Python
77 lines
2.9 KiB
Python
"""_analytics_tick:滚动刷新 + 增量扫描 + 异常吞掉(数据分析板块 v1, T11)。
|
|
|
|
行为测试是底线——start() 的 job 注册断言在 test_sync_scheduler.py 的
|
|
注册测试里同步更新(那里 spy 断言 add_job 全集)。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
|
|
import pytest
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.db.base import Base
|
|
from app.db.models import MsgDailyStats, SensitiveWord, StatDailyStaff
|
|
from app.storage.summary import day_key
|
|
from app.sync.scheduler import SyncScheduler
|
|
|
|
|
|
@pytest.fixture
|
|
async def sched_engine():
|
|
# StaticPool:内存库单连接共享,同 engine 的多 session 看到同一份数据
|
|
engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:", poolclass=StaticPool,
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
maker = async_sessionmaker(engine, expire_on_commit=False)
|
|
yield engine, maker
|
|
await engine.dispose()
|
|
|
|
|
|
async def test_analytics_tick_runs(sched_engine):
|
|
"""滚动 3 天窗覆盖今日 seed → stat_daily_staff 出行;tick 不抛。
|
|
|
|
_analytics_tick 用真实「今天」算滚动窗(time.time() 喂 day_key,上海
|
|
自然日)——seed 的 day 必须相对当日,否则窗口错过。
|
|
"""
|
|
engine, maker = sched_engine
|
|
today = day_key(int(time.time() * 1000))
|
|
async with maker() as s:
|
|
s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1",
|
|
peer_kind="friend", peer_id="t1", day=today,
|
|
send_cnt=1, recv_cnt=0, last_ts=1756950000000,
|
|
last_is_send=True))
|
|
s.add(SensitiveWord(word="私单"))
|
|
await s.commit()
|
|
sch = SyncScheduler(engine=engine, session_maker=maker)
|
|
await sch._analytics_tick() # 不抛即过
|
|
async with maker() as s:
|
|
n = (await s.execute(
|
|
select(func.count()).select_from(StatDailyStaff)
|
|
)).scalar_one()
|
|
assert n >= 1
|
|
|
|
|
|
async def test_analytics_tick_swallows_errors(sched_engine, monkeypatch, caplog):
|
|
"""内部异常只记日志不抛(保住下一个调度周期)。
|
|
|
|
直接让 recompute_days 抛错——_analytics_tick 内部的
|
|
``from app.storage.staff_summary import recompute_days`` 在调用时求值,
|
|
monkeypatch 模块属性即可命中。
|
|
"""
|
|
engine, maker = sched_engine
|
|
|
|
async def _boom(*args, **kwargs):
|
|
raise RuntimeError("boom")
|
|
|
|
monkeypatch.setattr("app.storage.staff_summary.recompute_days", _boom)
|
|
sch = SyncScheduler(engine=engine, session_maker=maker)
|
|
with caplog.at_level(logging.ERROR, logger="app.sync.scheduler"):
|
|
await sch._analytics_tick() # 不抛
|
|
assert any("analytics tick failed" in r.getMessage() for r in caplog.records)
|