From 8499a74a724c362cb0e68228e0652a85dbc815f7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Fri, 4 Sep 2026 23:27:41 +0800 Subject: [PATCH] =?UTF-8?q?fix(analytics):=20run=5Ffull=E6=B0=B4=E4=BD=8D?= =?UTF-8?q?=E5=89=8D=E7=BD=AE=E9=98=B2=E5=9B=9E=E5=A1=AB=E7=AA=97=E5=8F=A3?= =?UTF-8?q?=E6=BC=8F=E6=89=AB+README/DEPLOY/=E8=84=9A=E6=9C=AC=E5=8F=B7?= =?UTF-8?q?=E6=94=B6=E5=B0=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Code --- README.md | 10 ++++- docs/DEPLOY.md | 6 ++- .../app/analytics/sensitive/scanner.py | 16 +++++--- yunke-server/scripts/full_sync_all.sh | 2 +- yunke-server/tests/test_sensitive.py | 39 +++++++++++++++++++ 5 files changed, 63 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 7d96ecf..280e4b9 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,14 @@ python -m app.sync.cli sync --entity friends # one entity python -m app.sync.cli --help # full usage ``` +One-shot analytics backfills / self-heal (all idempotent): + +```bash +python -m app.sync.cli daily-stats # msg_daily_stats 日汇总全量重建 +python -m app.sync.cli staff-daily-stats # 员工日汇总(stat_daily_staff)全量重建 +python -m app.sync.cli sensitive-scan --full # 敏感词全量回填+建扫描水位(首跑必须 --full) +``` + ### Docker mode (full stack) ```bash @@ -176,7 +184,7 @@ Entity names accepted by `sync --entity` and `run_once`: ## Verification ```bash -cd yunke-server && .venv/Scripts/python.exe -m pytest -q # 323 tests +cd yunke-server && .venv/Scripts/python.exe -m pytest -q # 574 tests cd yunkes && npm run build # clean production build cd yunke-server && python -m app.sync.cli --help # CLI usage docker compose config # validates the stack diff --git a/docs/DEPLOY.md b/docs/DEPLOY.md index bc4a6f1..96c0242 100644 --- a/docs/DEPLOY.md +++ b/docs/DEPLOY.md @@ -215,8 +215,10 @@ alembic 链自当前线上版本起新增 3 支迁移:`a7d3e9f1c2b4` msg_daily_s (高频词分钟级)。**首跑必须带 `--full`**:不建水位,增量扫描会拒绝扫库。 - **顺序**:sensitive-scan --full 宜赶在调度器首次 4h analytics tick 之前; tick 首次在进程启动 4h 后才触发,部署当天跑回填通常天然满足。万一 tick - 先到,也只是每 4h 记一条 ERROR 日志(analytics tick failed,无扫描水位), - 属预期提示非故障,补跑 --full 后即消失。 + 先到,每 4h 记一条 ERROR 日志(analytics tick failed,无扫描水位),且该 + tick 整笔回滚——同笔的 staff rollup 增量刷新一并停摆,由夜间 + full_sync_all.sh step 12(staff-daily-stats 全量重建)自愈兜底;补跑 + --full 后即恢复,属预期提示非故障。 - 升级收尾 `sudo docker compose -f docker-compose.server.yml restart frontend` (让 nginx 重新解析 backend 容器 IP,防漂移后 502,同既有升级流程)。 - 浏览器验收:登录后依次打开 `/analytics/dashboard`、`/analytics/churn`、 diff --git a/yunke-server/app/analytics/sensitive/scanner.py b/yunke-server/app/analytics/sensitive/scanner.py index 190340d..653fe23 100644 --- a/yunke-server/app/analytics/sensitive/scanner.py +++ b/yunke-server/app/analytics/sensitive/scanner.py @@ -165,7 +165,7 @@ async def run_incremental(session: AsyncSession) -> dict: async def run_full(session: AsyncSession) -> dict: - """全量回填(CLI 一次性):每词单独扫全库;水位推进同增量。 + """全量回填(CLI 一次性):每词单独扫全库;水位在扫描前取(见内注)。 每词结果整批取回(MATCH/LIKE 已在服务端预筛成命中行)再分片复用 _scan_batch——不在未取尽的结果上发新语句/flush(MySQL 驱动会 @@ -178,6 +178,15 @@ async def run_full(session: AsyncSession) -> dict: ) ).scalars().all() st = await _get_state(session) + # 水位在扫描前取——回填期间新入库的行 fetched_at > 水位,下轮增量自然 + # 扫到;若在词循环后才取,回填窗口内 commit 的行 fetched_at ≤ 水位, + # >= 增量也扫不到(首部署夜间回填与 02:00 全量批写重叠时漏扫)。 + # >= 边界重扫由 UNIQUE(word_id,msg_id) 去重。 + new_wm = ( + await session.execute(sa.select(sa.func.max(Message.fetched_at))) + ).scalar_one_or_none() + if new_wm is not None: + st.last_fetched_at = new_wm total_hits = 0 scanned_rows = 0 for w in words: @@ -192,11 +201,6 @@ async def run_full(session: AsyncSession) -> dict: for i in range(0, len(rows), 500): total_hits += await _scan_batch(session, w, rows[i:i + 500]) await session.flush() # 行已 .all() 物化,分片 flush 控挂起对象数 - new_wm = ( - await session.execute(sa.select(sa.func.max(Message.fetched_at))) - ).scalar_one_or_none() - if new_wm is not None: - st.last_fetched_at = new_wm st.scanned += scanned_rows st.hits += total_hits st.last_run_at = dt.datetime.now(dt.UTC) diff --git a/yunke-server/scripts/full_sync_all.sh b/yunke-server/scripts/full_sync_all.sh index b8ceaa2..66599d7 100644 --- a/yunke-server/scripts/full_sync_all.sh +++ b/yunke-server/scripts/full_sync_all.sh @@ -47,7 +47,7 @@ echo "$(date) --- 10/10 Media download (audio+images) ---" $PY scripts/scrape_via_page.py --entity media # 11:会话日汇总全量重建(夜间自愈:绕过 upsert 钩子的写入由它纠正; # sessions/staff-chat-stats 读路径依赖此表,2026-09-03) -echo "$(date) --- 11/11 Daily-stats rebuild ---" +echo "$(date) --- 11/12 Daily-stats rebuild ---" $PY -m app.sync.cli daily-stats # 12:员工×日汇总全量重建(rollup 层随 msg_daily_stats 重建同步自愈; # A1 员工指标读路径依赖此表,2026-09-04) diff --git a/yunke-server/tests/test_sensitive.py b/yunke-server/tests/test_sensitive.py index 72f1117..0fa9fa2 100644 --- a/yunke-server/tests/test_sensitive.py +++ b/yunke-server/tests/test_sensitive.py @@ -190,6 +190,45 @@ async def test_disposition_preserved(make_client): await make_client(seed) +async def test_full_backfill_window_rows_scanned_by_incremental( + make_client, monkeypatch +): + """回填窗口内新入库的行不漏:水位必须在词循环前取。 + + 若 run_full 在词循环后才读 MAX(fetched_at),窗口内 commit 的行 + fetched_at ≤ 水位,>= 增量也扫不到——首部署夜间回填与 02:00 全量 + 批写重叠时漏扫。m_max(无命中)垫高窗口内 MAX,锁死判别力。 + """ + async def seed(session): + session.add(_msg("m1", "私单", dt.datetime(2026, 9, 4, 10, 0))) + await words_svc.create_word(session, word="私单") + await session.commit() + + orig = scanner._scan_batch + fired = False + + async def _hook(sess, word, rows): + nonlocal fired + if not fired: # 首个分片时模拟回填窗口内新消息入库 + fired = True + sess.add(_msg("m_win", "窗口内私单", dt.datetime(2026, 9, 4, 11, 0))) + sess.add(_msg("m_max", "正常", dt.datetime(2026, 9, 4, 12, 0))) + await sess.commit() + return await orig(sess, word, rows) + + monkeypatch.setattr(scanner, "_scan_batch", _hook) + await scanner.run_full(session) # 词行集已物化,窗口行不在其中 + await session.commit() + st = await session.get(SensitiveScanState, "message_text") + # 水位是扫描前取的(10:00),未被窗口行推后 + assert st.last_fetched_at == dt.datetime(2026, 9, 4, 10, 0) + r = await scanner.run_incremental(session) + await session.commit() + assert r["hits"] == 1 # 窗口内命中行(m_win)由增量补上 + + await make_client(seed) + + # ------------------------------------------------------------- C1 命中列表/处置 async def test_hits_list_default_filter_and_dispose(make_client):