csctalon/docs/standards/数据同步与加工说明.md

数据同步与加工说明

性质：参考（REF） —— 记录采集链路实现现状，以代码为准

版本：v0.8 · 日期：2026-09-01 状态：实现补充文档（开发者参考） 关联：规划设计.md §十四/§十九 数据同步与加工实现补充 范围：本文件记录数据同步链路实现细节——BMS Playwright 采集、ai-acc HTTP 接入、SCRM REST API 同步、OCR 自动登录、时区加工、BMS 浏览器时区与时间窗口设置、7 天分片、去重策略、凭证管理、任务互斥/冷却/增量/异步进度/复用登录/抓取导入并行、SyncDashboard 统一入口。 v0.4 新增：§八 SCRM 同步体系、§九 OCR 自动登录、§十 SyncDashboard 统一入口 v0.8 新增：§四.5 BMS 浏览器时区与时间窗口设置（2026-09-01 生产审计修复：容器浏览器时区 UTC、timezone_id=Asia/Shanghai、fill_utc=False、time_labels 标签匹配、:visible 非法选择器、负数新增告警、op_time 窗口二次校验）

---

一、数据同步链路总览

1.1 数据源与抓取方式

系统当前支持 5 个数据源，按 fetch_mode 分流到两条抓取链路：

data_type	名称	base_url	fetch_mode	抓取器	备注
refund	退费单管理	https://wifi.cloud-cat.cn/bms	playwright	BmsFetcher	浏览器自动化
ticket	服务请求管理	https://support.cloud-cat.cn/bms	playwright	BmsFetcher	浏览器自动化
wallet	提现操作记录	https://support.cloud-cat.cn/bms	playwright	BmsFetcher	浏览器自动化
device_combo	设备套餐	https://support.cloud-cat.cn/bms	playwright	BmsFetcher	default_sync=False
cs_record	智能助手处理记录	http://ai-acc.cloud-cat.cn	http	ai_acc_fetcher	REST API，非浏览器

fetch_mode 不在 sync_source_configs DB 表里，由 BMS_DATA_SOURCES 种子在配置读取时（_row_to_config）合并进来，用于 bms_sync.sync_all 分流。

1.2 抓取链路分流

bms_sync.sync_all 把任务按 fetch_mode 分两组：

• fetch_mode == "http" → ai_acc_fetcher.fetch_one（httpx，无浏览器）
• 其他（默认 playwright）→ BmsFetcher.fetch_all（共享一个 Playwright 浏览器实例）

两条链路最终都产出一个本地 xlsx 文件，统一走 _auto_import（parse_file → load_dataframe → run_upsert）入库。因此所有数据源共用一套解析与加工口径（列映射、类型转换、时区加工、去重策略），抓取方式的差异不渗透到导入层。

1.3 时间范围计算（v0.3.2 重构后）

触发方式	时间范围	实现
定时同步	当天 00:00 ~ now（东八区）	bms_scheduler._do_scheduled_sync 硬性限定 today_start ~ now_cn
手动同步（未指定 range, 默认增量）	上次成功同步 end_time ~ 当前时间；若从未成功过或比今日 0 点早，则从今日 0 点起算	SyncTaskManager.compute_incremental_ranges (查 bms_sync_logs 最近 SUCCESS 记录的 time_end)
手动同步（用户指定自定义区间）	调用方传入的精确区间	前端按 today/yesterday/本周/自定义等 UI 传 time_ranges + force=true 跳过冷却
/task/preflight 返回值	三种组合字段：busy (有运行中任务)/cooldown_remaining(距上次成功剩余秒数)/incremental_ranges(data_type→[start,end])	供前端展示给用户，用户可确认后再调 /task/submit

硬性约束：

1. 定时同步必须显式限定到当天数据，不允许同步过多历史数据。每次定时覆盖当天 0 点至今，upsert 幂等不产生重复，反而能补回上次失败漏掉的数据。
2. 手动同步有 5 分钟冷却期 (DEFAULT_COOLDOWN_SECONDS = 300) — 上次成功结束到本次开始不足 5 分钟的，非 force=true 一律拒绝，避免无意义地频繁拉取服务器。
3. 任何时刻最多一个同步任务在执行 — 由 SyncTaskManager._lock（全局 asyncio 互斥锁）保证：手动/定时无论谁先拿到锁，另一路都会抛 RuntimeError("当前有同步任务正在执行…请等待其完成")。

---

二、ai-acc 智能助手处理记录接入（cs_record）

2.1 为何走 HTTP 而非 Playwright

ai-acc 平台提供标准 REST API（登录 + 导出），无需浏览器自动化即可完成抓取，相比 Playwright：

• 无浏览器进程开销，启动快、资源占用低
• 无 1 分钟操作间隔限制（BMS 系统有此限制）
• 无页面 DOM 变更脆弱性
• 适合服务器无人值守定时同步

2.2 接口契约

登录：POST {base_url}/ai-acc/portal/v1/auth/login

• body: {"username":"...","password":"..."}
• response: {"token":"...","userId":"...","username":"..."}
• ⚠️ token 在 JSON 顶层，不在 data 字段下——与 BMS 风格 {data:{token}} 不同。_login 同时检查 data.token 与顶层 token 兼容两种风格。
• 登录同时下发 session cookie，httpx 共享 client 自动保持。

导出：GET {base_url}/ai-acc/cs/v1/assistant/handleRecord/export

• params: createDateStart=YYYY-MM-DD HH:mm:ss & createDateEnd=YYYY-MM-DD HH:mm:ss
• auth: Authorization: Bearer {token} + session cookie
• response: xlsx 文件流（application/vnd.openxmlformats-officedocument.spreadsheetml.sheet）
• ⚠️ 用户提供的 URL /admin/assistant/handle-record-export 是前端路由，不是 API 端点。

2.3 非文件流识别

导出接口在业务异常时返回 JSON 而非文件流，典型场景：

{"code":"500","message":"操作时间范围不能超过7天","timestamp":1785459901135}

_export_chunk 检查 Content-Type，含 json 则抛异常（带响应体前 300 字符），避免把错误 JSON 当 xlsx 解析。

---

三、7 天时间范围分片

3.1 限制规则

ai-acc 导出接口限制单次时间范围不得超过 7 天：

• 正好 7 天：允许（HTTP 200）
• 超过 7 天：拒绝（返回上述 JSON 错误）

3.2 分片实现

ai_acc_fetcher._chunk_time_range 把 [start, end] 切成若干 ≤7 天的连续子区间：

def _chunk_time_range(start, end, max_days=7):
    chunks = []
    cur = start
    step = timedelta(days=max_days)
    while cur < end:
        chunk_end = min(cur + step, end)
        chunks.append((cur, chunk_end))
        cur = chunk_end
    return chunks

• 子区间边界相接（chunk_i.end == chunk_{i+1}.start）
• API 端点 inclusive，边界秒的记录可能在两片中都出现 → 合并时 drop_duplicates() 去重

3.3 合并实现

逐片 GET → pd.read_excel(BytesIO(content), dtype=str, engine="openpyxl") 读为 DataFrame → pd.concat 合并 → drop_duplicates()：

frames = []
for cs, ce in chunks:
    content = await _export_chunk(...)
    df = pd.read_excel(BytesIO(content), dtype=str, engine="openpyxl")
    frames.append(df)
merged = pd.concat(frames, ignore_index=True).drop_duplicates()

• dtype=str：与 data_parser 解析口径一致（全文本读入，由 upsert 层做类型转换），避免数值/日期列在合并时被 pandas 推断错类型。
• 合并后写入单个 xlsx（openpyxl，单行表头 + 全文本），文件名 ai_acc_{data_type}_{ts}.xlsx，落 uploads/{YYYY-MM}/。

3.4 验证

14 天范围（2 片）E2E 测试：合并文件 69015 行，内部 0 重复，边界 0 重复。已清理测试数据。

---

四、时区加工（关键 bug 修复）

4.1 问题

运营平台（BMS CSV / ai-acc xlsx）导出的时间字段为 naive 北京时间字符串（无时区标识）。原 upsert._to_datetime 用 pd.to_datetime 返回 naive datetime，存入 DateTime(timezone=True) 列时，因 DB 时区为 Etc/UTC，naive 值被当 UTC 解释，导致按北京显示时偏后 8 小时。

复现：CSV 原始 2026-07-30 22:32:00 存为 22:32+00 UTC，前端按北京显示为 2026-07-31 06:32:00（次日）。

4.2 修复

upsert._to_datetime 把 naive 视为北京时间（UTC+8）→ tz_localize(_TZ_CN).tz_convert(utc) 后再存；aware datetime 原样返回：

_TZ_CN = timezone(timedelta(hours=8))

def _to_datetime(val):
    if isinstance(val, datetime):
        if val.tzinfo is None:
            return val.replace(tzinfo=_TZ_CN).astimezone(timezone.utc)
        return val
    result = pd.to_datetime(val)
    if result.tzinfo is None:
        result = result.tz_localize(_TZ_CN).tz_convert(timezone.utc)
    return result.to_pydatetime()

• 影响所有 BMS/ai-acc 导入路径（手动上传、一键同步、定时同步、Playwright/HTTP 两条链路），因为统一走 upsert 层。
• 纯日期字段（如 complaint_date）迁移后按北京显示 00:00（日期不变）。
• 系统内部表（incidents/measures/reports 等服务器生成 UTC 时间）未动。

4.3 存量数据迁移

修复代码只影响新写入数据，存量 naive-当-UTC 的数据需一次性 SQL 迁移。对 7 张有数据的业务表执行 col - INTERVAL '8 hours'：

表	迁移行数
wallet_withdrawals	9653
service_tickets	18706
refunds	42944
cs_records	70227
device_combos	44672
external_complaints	1312
card_complaints	720
merchant_refunds	0（空表，跳过）

验证：wallet 记录迁移后显示 2026-07-30 22:32:00 北京 = CSV 原始值；wallet 小时分布回归业务时段（不再凌晨聚集）。

4.4 后续保障

代码已编译部署，未来所有同步路径（手动上传 / 一键同步 / 定时同步 8-23/0 每小时 + 8:00 近 7 天全量 + 5:00 前一天补跑 / BMS / ai-acc）自动按北京时间 → UTC 转换，无需额外操作。

4.5 BMS 浏览器时区与时间窗口设置（2026-09-01 生产审计修复）

背景：2026-08-28 ~ 09-01 生产日志审计发现「当日退款/投诉数据无法按小时实时同步」。排查结论：不是数据删除、也不是时区换算 8 小时偏移，而是 BMS 页面日期时间框（Element UI DatePicker）被自动化填入的"时间窗口"不可靠——窗口没设置成功时，BMS 页面保留默认/上次时间范围，导致 ① 当前窗口增量数据漏掉（当天看不到）② 拉到窗口外的旧数据与库内已存在记录冲突（统计负数）。真正根因是浏览器自动化填入日期框这一步，而非数据本身或时间换算。

4.5.1 根因一：容器浏览器时区 = UTC（已修复）

Playwright 无头浏览器进程默认时区是 UTC，而 BMS 前端按浏览器时区取"当前时刻"并解析输入串。此前把北京时间转 UTC 填入（fill_utc=True），BMS 又按字面值解析 → 窗口错位。

修复（bms_fetcher._do_fetch_one / _do_fetch_one_after_login）：

• new_context(timezone_id='Asia/Shanghai')（两处）→ BMS 前端当前时刻 = 北京时间
• 填入的时间字符串保持北京时间（fill_utc=False，即 _bms_time_str 默认按 dt 自身时区格式化，与用户手动导出一致）
• 每源可用 sync_source_configs.time_fill_utc 覆盖

_bms_time_str(dt, time_format="%Y-%m-%d %H:%M:%S", fill_utc=False)

4.5.2 根因二：ticket 时间标签匹配失败（已修复）

ticket 源配置 time_labels=["起始登记时间","结束登记时间"]，但 BMS 服务请求管理页：

• 这些字段是 placeholder 而非 .el-form-item__label 元素 → _fill_input_by_label 找 label 永远失败
• 时间框在高级筛选里默认折叠不可见（visible=False）→ 需先点 .show-more 展开

修复（bms_fetcher._set_time_by_labels + 新增 _fill_daterange_inputs）：

1. 先点 .show-more 展开高级筛选（仅当时间框不可见时）
2. 按 placeholder 精确匹配（input[placeholder="起始登记时间"]）定位正确的登记/完成时间框
3. _fill_daterange_inputs：每个 input click 聚焦 → fill(4s 超时) → evaluate setter 兜底，用 blur + ESC 关面板（不用 page.mouse.click 避免滚动不稳定）——Element UI DatePicker 的 input 需先 click 聚焦才可 fill，否则 fill 超时且 setter 不触发 Vue v-model → 查询用默认近一周。

4.5.3 根因三：_set_time_range 兜底 evaluate 内 :visible 非法选择器（已修复）

_set_time_range 最后兜底在 page.evaluate 里执行 document.querySelectorAll('input.el-range-input:visible, .el-date-editor input:visible')——:visible 是 jQuery / Playwright 引擎伪类，原生 JS querySelectorAll 不识别，直接抛 SyntaxError（生产 wallet 09-01 12:05/15:05 轮次 'input.el-range-input:visible' 不是合法选择器，抓取中断、窗口回退默认范围）。

修复：先按类名选中（input.el-range-input, .el-date-editor input），再用 JS getBoundingClientRect + computedStyle 过滤可见性：

const isVisible = (el) => {
  const rect = el.getBoundingClientRect();
  const style = window.getComputedStyle(el);
  return rect.width > 0 && rect.height > 0
      && style.visibility !== 'hidden' && style.display !== 'none';
};

⚠️ 注意区分：Playwright 引擎侧（page.query_selector(':visible') / page.locator('button:visible')）的 :visible 是合法的引擎伪类；只有 page.evaluate 内原生 JS 里不能用。

4.5.4 时间窗口设置的完整策略链（_set_time_range）

按优先级从高到低，命中即 return：

1. 策略0：轮询（最多 6s）找 Element UI datetimerange/daterange 容器（.el-date-editor--datetimerange/.el-date-editor--daterange/.el-range-editor），取内部 2 个 .el-range-input → click→ctrl+a→backspace→type 键盘输入 → 回填 value 二次校验（DOM value + Vue2 实例 value，失败则 fallback _fill_el_input+setter，再失败抛错中止避免按错误窗口导出脏数据）
2. 策略1：.el-date-editor--datetimerange 容器的 .el-range-input
3. 策略2：可见日期相关 input 取前两个
4. 策略3：placeholder 含「开始/结束」兜底 fill
5. 策略4（终极兜底）：所有可见 input 按 y 坐标分组，优先命中含 el-range-input 的组，否则取最上一行 ≥2 个非 readonly text input，用 JS setter + input/change 事件触发
6. 最后兜底：evaluate 找 range（见 4.5.3 修复）

_set_time_by_labels（有 time_labels 的源，如 ticket）额外前置 _fill_input_by_label 的 label 查找与 placeholder 策略0。

4.5.5 负数"新增"与告警（统计口径，不删数据）

upsert_wallet 用 on_conflict_do_nothing + 批次去重。当窗口设置失败拉到大量已存在旧流水时，existing_count > unique_rows → 旧代码 result.inserted += len(unique_rows) - existing_count 产生负数新增（生产 -4885/-587/-515/-153/-15）。不删除任何数据，负数只是统计口径失真。

修复（upsert.py upsert_wallet）：

• inserted += max(net_inserted, 0) 保证统计不为负（口径 = 本次实际新增条数）
• net_inserted < 0 时打 WARNING：[wallet] 疑似时间窗口未设置成功: 本轮拉回 N 行中 M 行已存在于库内 (净新增 X<0) → 便于早发现窗口设置失败

4.5.6 op_time 窗口二次校验（审计建议 3）

wallet 导入时输出实际拉回数据的 op_time 分布（UTC），用于对照同步日志里的「设置时间范围」判断窗口是否真的生效：

[wallet] 实际入库 34 行 op_time 范围: 2026-08-31 16:04:41 ~ 2026-09-01 02:49:45

生产 09-01 14:05 轮：同步窗口应为 09-01 00:00 ~ 14:05，实际拉回却大多落在前一天深夜~凌晨 → 直接坐实"页面保留了默认/上次时间范围"。

4.5.7 排障速查

现象	看什么	处理
当天数据看不到 / 增量缺失	同步日志 设置时间范围: X ~ Y 是否 = 期望窗口	检查 4.5.1~4.5.3 修复是否已部署（timezone_id + fill_utc=False + placeholder/:visible 修复）
[wallet] 导入完成: 新增 为负数	是否命中 4.5.5 WARNING	窗口未设置成功，回退默认范围拉到旧流水；确认修复已部署后重同步
[wallet] 实际入库 N 行 op_time 范围 与窗口不符	同上	同上
未找到 '起始登记时间' 对应的输入框 / fill 失败,退化为 evaluate 直接设值	ticket 标签匹配	确认 4.5.2 修复（placeholder 策略 + .show-more 展开）已部署
'input.el-range-input:visible' 不是合法选择器（SyntaxError）	_set_time_range 兜底 evaluate	确认 4.5.3 修复已部署

---

五、去重策略（pk_strategy）

5.1 三种策略

各业务表按是否有天然唯一键采用不同入库策略，由 upsert_*.py 内部分流：

策略	实现	适用表
upsert（主键去重更新）	pg_insert.on_conflict_do_update，已存在则更新非主键字段	refunds(refund_no)、tickets(ticket_no)、merchant_refunds(wx_refund_no)、merchants(sub_name)、device_combos(recharge_order_no)、external_complaints(有 complaint_no 的部分)、manual_refunds(复合键,见 5.3)
none（无主键插入）	pg_insert.on_conflict_do_nothing 兜底，不去重	wallet、cs_record、card_complaint、external_complaints(无 complaint_no 的部分)
fallback 主键	无天然主键时构造稳定合成主键，再走 upsert	device_combos 初始套餐

5.2 cs_record 为何不去重

cs_record（智能助手处理记录）无天然唯一键：

• 同设备同时间可多次接待（业务上合法）
• 导出表无单号/ID 列
• 即使用 (device_sn, operation_time) 组合也无法保证唯一（同秒多次接待）

因此 upsert_cs_records 走 on_conflict_do_nothing，不去重。带来的影响：

• ✅ 正常增量同步（定时用 last_end_time 非重叠区间）：无重复
• ⚠️ 手动重同步重叠范围：会产生重复行（需人工避免重叠，或后续提供去重工具）

5.3 device_combo fallback 主键

初始套餐（免费赠送/体验套餐）在 BMS 导出时无充值订单号，此前被主键非空校验跳过导致缺失。现生成稳定 fallback 主键：

INIT::{device_sn}::{package_name}::{start_date}

• INIT:: 前缀与真实订单号（BR/WX/SP 开头）隔离，绝不冲突
• (device_sn, package_name, start_date) 组合唯一，保证多次增量同步幂等（命中 on_conflict_do_update）
• device_sn 缺失时无法构造稳定唯一键，才报错跳过（安全网，正常初始套餐不触发）

5.4 external_complaints 混合策略

投诉编号有则走 upsert（96%），无编号的行（~3%）走 on_conflict_do_nothing 插入（与 cs_record 一致，重复导入会累计）。投诉编号单元格含 \n 前缀，导入时剥离。

5.5 manual_refunds 复合去重键（2026-08-24 调整）

手动退款登记（商户号手机退款处理登记.xlsx / workbuddy 推送）无单号主键，去重键为 (北京登记日期, order_no, COALESCE(device_sn,''), COALESCE(registrant,''))（唯一索引 uq_manual_refunds_dedup，部分索引 WHERE order_no IS NOT NULL）：设备号+订单号+登记日期+登记人 确定一笔退款单，退款金额/状态修改走 UPDATE 不新增（旧键含退款金额导致改金额即新增→残留累积使 D 路虚高，2026-08-24 清理 3929→1976 行）；workbuddy 推送支持 full_sync=true 全量对账删除残留行，详见 workbuddy手动退款数据同步接入规范.md

5.6 钱包提现"失败/冻结"备注口径(2026-08-28)

BMS 提现操作记录中,备注含 提现申请-冻结 等失败/冻结关键词的记录表示提现申请被冻结/拒绝,钱并未实际出账给用户,并非成功退款。但这类记录 op_type 仍是 用户提现,若不排除会导致"成功退款/成功提现"统计虚高。同时,同一笔提现会出现"冻结记录 + 成功记录"成对(同一 WD 提现单号,冻结记录 op_time=申请时刻、成功记录 op_time=实际出款时刻)。

统一口径(用户确认,2026-08-28):

1. 冻结/失败记录全部排除:备注含 冻结/退款失败/提现失败/超期订单不允许退款/驳回/拒绝/未到账 等失败关键词 → 不计入成功退款统计。
2. 同单成对只算成功:同一 WD 提现单号同时有冻结+成功记录时,只统计成功记录(排除冻结记录后自动满足)。
3. 历史数据同样生效:过滤基于备注字段,历史已入库的失败/冻结记录(如 微信退款失败：超期订单不允许退款)同样被排除,无需数据清洗。

实现位置:

• analytics.py 新增 _WITHDRAW_FAIL_REMARK_KEYWORDS + _withdraw_fail_remark_sql_cond(),_route_b_withdraw_filter() 统一加 COALESCE(remark,'') NOT LIKE '%关键词%' 条件;refund_decision_overview/refund_export_all/wallet_vs_withdraw/cash_out_trend 等内联 SQL 同步加同一条件。
• preset_views_seed.py 各"用户提现"segment 的 filters 追加 remark not_contains 排除。
• quality_check.py 的 check_wallet_row 对失败/冻结备注标记 pending(数据质量页可见,便于审计)。

5.7 通用 upsert 执行器

_execute_upsert 先查已存在主键数量再执行 on_conflict_do_update，以精确区分 insert/update 计数。同批次内重复主键会被去重（保留最后一条，符合"后到覆盖"语义），避免 ON CONFLICT affect row a second time 错误。

---

六、凭证管理

6.1 加密存储

各数据源账号密码独立配置，存于 sync_source_configs 表，用 Fernet 加密（app/core/crypto.py）：

• 启动时 ensure_sync_sources_registered 幂等 seed：缺失的源用 .env 的 BMS_USERNAME/BMS_PASSWORD 加密灌入；已存在的源不覆盖（保留用户在 UI 的编辑）
• 运行时 get_sync_source_config 解密返回给 fetcher
• API list_sync_source_configs 返回掩码（has_credentials 布尔），不泄露明文

6.2 ⚠️ 加密密钥变更影响

修改 .env 的 CONFIG_ENCRYPTION_KEY 或 JWT_SECRET_KEY 会使现有 sync_source_configs 凭证解密失败：

• 现象：解密失败后系统把密文当作用户名发送给 BMS，BMS 返回"用户密码错误"
• 修复：在「系统管理 - 同步设置」页重新填写各源账号密码（PUT /api/v1/system/sync-sources/{data_type}），重新加密入库
• 预防：变更加密密钥后，所有数据源凭证必须重新写一次

6.3 凭证来源优先级

1. DB sync_source_configs 表（解密后）—— 运行时主用
2. .env BMS_USERNAME/BMS_PASSWORD —— 仅 seed 初始值 / DB 缺失时回退

---

七、定时同步调度

7.1 调度规则（东八区）

• 白天每小时：08:00, 09:00, ..., 23:00, 00:00（共 17 次）当天窗口增量同步（带专项级熔断过滤）
• 8:00 全量兜底：近 7 天窗口全量同步（覆盖 BMS 数据按登记时间分批滞后 1~2 天的回填，忽略专项熔断）
• 5:00 前一天补跑：对前一天整天做整体同步（忽略专项熔断，强制跑所有源；2026-08-31 由 3:00 调整，避开 BMS 平台凌晨维护时段）
• 触发窗口：每分钟检查，只在定时小时的前 5 分钟内触发（防止错过）
• 同一小时去重：_last_run_hour 字典防止同一小时重复执行
• 仅同步 default_sync=True 的源（DEFAULT_SYNC_TYPES）

调度常量见 backend/app/services/bms_scheduler.py：DAYTIME_HOURS（8~23+0）、CATCHUP_HOUR（5）、FULL_WINDOW_DAYS（7）。

7.2 启动前置检查

_default_sources_have_credentials 检查默认同步的源是否都已配置凭证，未配置则跳过本次定时并告警（无人值守必须预设）。

7.3 时间范围约束

• 白天增量 / 8 点全量：time_ranges = {dt: (今天/近 N 天 0 点, now_cn)}
• 5 点补跑：锁定「前一天整天」（yesterday_full_sync=True）
• 理由：upsert 幂等不会产生重复；覆盖当天 0 点至今能补回上次失败漏掉的数据；防止无人值守时误同步过多历史数据拖垮系统。

7.4 与 SyncTaskManager 共享全局互斥锁 (v0.3.2)

调度器现在不直接跑 sync_all_data，而是把任务提交给 SyncTaskManager.submit(trigger_type='scheduled'…)：

• 效果：若此时用户正手动同步，调度器拿不到锁会立即抛错跳过本小时（日志记录 scheduler task skipped: 有 xxx 触发的任务在执行），不会阻塞。
• 反方向也保护：若定时正在执行，用户点「开始同步」也会被后端拒绝并提示"当前有定时任务在执行"。
• 冷却对 scheduled 触发无效（定时本就按 4h+ 间隔运行，无需再叠冷却）。

---

八、同步任务管理：SyncTaskManager（v0.3.2 新增）

8.1 设计背景与目标

单后端单实例部署 + 长耗时抓取（BMS playwright 登录+导出 45 张表通常 38 分钟，下载到本地与导入到 DB 合计又数分钟）下，原实现存在几类典型问题：

问题	用户表现	解决方式
重复发起	用户点击「同步」后 UI 无进度反馈，超时后再点，会在后台同时开两个浏览器，BMS 同账号多 session 被踢下线	全局 asyncio.Lock 互斥锁 + 前端提交后不再阻塞（taskSubmit 返回 task_id → 前端轮询进度）
冷却不足	用户 30 秒内点两次，其实服务器数据一分钟内并无差异 → 白白浪费抓取配额	上次成功结束后 5 分钟内，非 force=true 一律拒绝
无增量语义	每次「一键同步」都默认拉今日 0 点，重复抓取白天相同记录 → 浪费服务器流量与配额	compute_incremental_ranges 查 bms_sync_logs 最近 SUCCESS 记录的 time_end，下次就从那个时间起
无进度可见	点按钮后前端长时间转圈 / 超时，后端日志只能看 shell，无法在 UI 内反馈进度	每个任务一个 SyncTaskProgress，实时记录阶段 + 子步骤进度，taskStatus 接口可轮询

8.2 核心数据结构

SyncTask (一个任务对象，内存常驻直到被覆盖) + SyncTaskProgress：

@dataclass
class SyncTaskProgress:
    stage: str                          # preflight / fetching / importing / done
    message: str = ""                   # 中文文案, 直接展示
    current_step: int = 0
    total_steps: int = 0
    sub_progress: float = 0.0           # 0.0 ~ 1.0, 用于当前步骤内细分进度
    sub_message: str = ""
    results: list = field(default_factory=list)  # 每个 data_type 完成后立即写入
    started_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    finished_at: Optional[datetime] = None
    status: str = "running"             # running / success / failed
    error: Optional[str] = None

进度回调签名：on_progress(stage, message, *, cur, total, sub=0.0, sub_msg="") — 所有 fetcher/sync 代码都已接入这个回调，下载一个文件 / 导入一张表后立刻推进前端可见进度。

8.3 新增异步 API（/api/v1/fetch/*）

路由	方法	用途
/task/preflight	POST	提交前检查：返回 { busy, current_task, cooldown_remaining, cooldown_total, incremental_ranges }。前端可据此显示"当前有任务执行/距离上次仅过 41 秒/本次增量范围 14:23 起"等提示。
/task/submit	POST	异步提交，校验锁+冷却后立即返回 {task_id, message}，然后在后台 asyncio.create_task 里执行真正的同步（不阻塞 HTTP 响应, 避免超时）。
/task/status	GET ?task_id=sync_xxx	轮询进度（每秒一次即可），返回完整 SyncTaskProgress。前端据此渲染进度条 + 各步骤卡片。
/sync-all	POST	兼容旧前端，内部调用 submit。若 submit 失败则回落到 legacy 阻塞式同步，保证行为一致。

前端调用推荐流程：先点「开始同步」→ 前端调用 /task/preflight → UI 提示用户本次将做什么 → 确认后调用 /task/submit → 拿到 task_id 后按 1s 间隔轮询 /task/status → finished 后 toast 成功 + 刷新当前页图表；中途可任意关闭页面不影响后台执行（除非重启后端进程）。

---

九、BMS 同 base_url 复用浏览器 Context（v0.3.2 优化）

9.1 问题来源

5 张 BMS 表里，refund 的 base_url 是 wifi.cloud-cat.cn，其余 ticket/wallet/device_combo/cs_record_bms 都在 support.cloud-cat.cn。若每抓一张表都新 Context + 新登录，就会发生：

• BMS 登录滑块偶尔触发，人工 1 分钟限制也在"每次登录"上叠加 → 总耗时成倍增加
• 同一 IP 反复登录触发风控 → 导出接口变慢

9.2 优化：按 base_url 分组共享登录后的 Context

BmsFetcher.fetch_all() 内部先按 base_url 对 tasks 分组，组内共用一个 Playwright BrowserContext：

1. 组首任务第一次去该 URL 时走完整登录（含滑块 → 等待登录成功 Cookie）
2. 登录成功后 already_logged_in = True
3. 组内后续任务访问时，带上已有 Cookie / 会话 → BMS 后台直接放行，不再重登
4. 组内两个任务之间仍遵守 BMS 官方 1 分钟「两次下载间隔」限制（同一 web 间隔超过 1 分钟即可，登录复用后下载等待语义不变）
5. 整组结束后 finally: context.close() 保证会话释放

实测效果（4 张 support 域 + 1 张 wifi 域 = 两组）：登录次数从 5 次降为 2 次，端到端耗时平均缩短 40%~60%。

---

十、抓取与导入并行化（v0.3.2 优化）

10.1 关键点：下载等待 1 分钟 vs DB 导入可并发

BMS/ai-acc 对「两次导出请求的间隔」有 1 分钟下限（同一 Web），但这个 1 分钟等待发生在抓取器（BmsFetcher）内部，跟本地的 DB 写入是无关的。以前链路是串行：

抓 refund → 等待间隔 60s → 抓 ticket → 等待 60s → 抓 wallet → 等待 → …
全部抓完后 → 导入 refund → 导入 ticket → 导入 wallet → …

等待间隔期间 CPU/DB 都是空转。

10.2 优化后：每个 FetchResult 就绪立刻启动独立导入线程

# bms_sync._sync_all_async_runner 的 progress 包装回调
async def on_progress_with_import(stage, message, **kw):
    if stage == "fetch_done":
        # 单个 data_type 刚下载到本地 xlsx, 立刻扔线程池开始导入, 不阻塞下一次抓取
        asyncio.create_task(asyncio.to_thread(_do_import, fr))
    await orig_progress(...)  # 继续调用上层回调推进整体进度条

# 所有抓取跑完后, gather 等待所有导入线程结束
await asyncio.gather(*import_pending_tasks, return_exceptions=True)

如此抓取与导入重叠，总体端到端耗时可以进一步压缩 30%~40%（尤其表数多、单表体积大时效果更显著）。

不破坏原有的 1 分钟访问约束：互斥只在"抓取请求之间"，导入是本地写库，跟抓取互不阻塞。

---

十一、前端看板组件化（v0.3.2 UI 侧同步）

11.1 背景

今日看板 (Today.vue)、接待看板 (Reception.vue)、工作台首页 (Dashboard.vue)、退款与提现看板 (RefundWithdrawal.vue)、现金流分析 (Cash.vue) 存在大量相同结构的代码：

• KPI 卡（图标 + 标签 + 数值 + 单位 + 环比）在 5+ 份文件里各写了一遍 HTML+CSS
• 现金流 4 路线拆分 + 对账偏差 (A/B/C/D + 经营估算 vs 实际) 在 Today、Dashboard、Cash 三处重写
• FilterBar、区块标题、图表卡、待办队列面板样式在每个看板都各写一份 (共 3~5 个拷贝)

这会导致 UI 调整需要改多份文件，极易出现"改了看板 A 忘记看板 B"的不一致。

11.2 抽出的复用组件 (全部在 components/dashboard/)

组件	用途	已覆盖的看板
KpiCard.vue	统一 KPI 卡片（含金额/数值、图标、环比标签、去重设备人次追加行、简洁 hideYoy 模式、tip tooltip）	Today / Reception / Dashboard首页
RouteSplit.vue	四路现金拆分+对账偏差（堆叠进度条可视化 + op/fc/diff 三合计 + compact/hideHeader 模式）	Today / Dashboard首页
ChartCard.vue	图表外层卡片（标题 + loading + extra 插槽），统一所有图表卡片的 padding/间距	Today / Reception
SectionHeader.vue	区块标题（左边彩色竖条+主副标题），替代 4 处 .section-title 内联样式	Today / Reception
FilterBar.vue	快捷日期选择 + 自定义区间 + 刷新按钮（支持 v-model）	Today / Reception
QueuePanel.vue	三列事务面板（待办/专项/投诉，优先级 Tag + 截止时间格式化 + 点击跳转）	Today

11.3 代码量缩减（构建后 chunk 对比）

Dashboard.js   19.36 kB → 12.87 kB  ↓ 34%
Today.js       31.38 kB → 26.49 kB  ↓ 16%
Reception.js   17.83 kB → 17.86 kB  ≈ 持平 (拆分后公共代码抽到独立 chunk)
- 公共组件 chunk:
  KpiCard + ChartCard + RouteSplit + FilterBar + SectionHeader + QueuePanel
  合计 < 20 KB (gzip < 7 KB)

新增看板 / 新增分析卡片时可直接 <KpiCard ... /> + <SectionHeader ... /> + <RouteSplit ... /> 快速搭出风格一致的页面，不再重复写内联 CSS。

---

十二、文件清单（v0.3.2 刷新）

文件	职责	v0.3.2 是否新增/大改
backend/app/services/sync_task_manager.py	核心新增：SyncTaskManager 单例（全局互斥锁/冷却/增量/进度/异步任务提交）	✅ 新增
backend/app/services/bms_fetcher.py	BMS_DATA_SOURCES 种子 + BmsFetcher（Playwright，已按 base_url 分组复用 Context）	✅ 大改
backend/app/services/ai_acc_fetcher.py	ai-acc HTTP 抓取（登录 + 导出 + 7 天分片合并）
backend/app/services/bms_sync.py	同步编排（fetch 完成即启动导入线程并行化 + SyncTaskManager 的 runner）	✅ 大改
backend/app/services/bms_scheduler.py	定时调度（8-23/0 每小时 + 5:00 补跑 + 8:00 全量兜底；现走 SyncTaskManager 提交通道共享锁）	✅ 小改
backend/app/api/v1/fetch.py	新增 /task/preflight /task/submit /task/status 三个 API；/sync-all 兼容旧前端	✅ 大改
backend/app/services/sync_source_config_service.py	凭证加密存取 + fetch_mode 合并
backend/app/services/upsert.py	类型转换（含时区加工 _to_datetime）+ 三种去重策略
backend/app/services/data_parser.py	CSV/XLSX 解析 + 列映射建议
backend/app/core/crypto.py	Fernet 加解密
web/src/components/dashboard/KpiCard.vue	可复用 KPI 卡片	✅ 新增
web/src/components/dashboard/RouteSplit.vue	四路现金拆分 + 对账偏差可视化	✅ 新增
web/src/components/dashboard/ChartCard.vue	图表外层卡片	✅ 新增
web/src/components/dashboard/SectionHeader.vue	区块标题	✅ 新增
web/src/components/dashboard/FilterBar.vue	日期快捷筛选 + 自定义区间	✅ 新增
web/src/components/dashboard/QueuePanel.vue	三列事务面板	✅ 新增
web/src/views/dashboards/Today.vue	今日看板 (已组件化，清理死代码)	✅ 大改
web/src/views/dashboards/Reception.vue	接待看板 (已组件化)	✅ 大改
web/src/views/Dashboard.vue	工作台首页 (组件化 KpiCard + RouteSplit)	✅ 大改

---

十三、变更记录

日期	版本	变更	关联
2026-08-02	v0.3.2	同步任务管理体系（SyncTaskManager 全局互斥/冷却/增量/进度/异步提交）	sync_task_manager.py + fetch.py /task/preflight·submit·status
2026-08-02	v0.3.2	抓取优化 1：BMS 同 base_url 分组复用浏览器 Context 登录	BmsFetcher.fetch_all → group by base_url
2026-08-02	v0.3.2	抓取优化 2：单文件抓完立刻启动导入线程，抓取与导入并行	bms_sync._sync_all_async_runner (asyncio.to_thread + gather)
2026-08-02	v0.3.2	前端看板 6 类卡片抽公共组件，Today/Reception/Dashboard 首页替换；代码量↓ 16%~34%	components/dashboard/*.vue（6 个新文件）
2026-07-31	v0.3.1	新增 cs_record 数据源（ai-acc HTTP 接入）	ai_acc_fetcher.py / bms_fetcher.py / Dockerfile（httpx）
2026-07-31	v0.3.1	修复时区 bug（naive 北京时间 → UTC）	upsert.py _to_datetime + 7 张表 SQL 迁移
2026-07-31	v0.3.1	ai-acc 7 天范围分片 + 合并去重	ai_acc_fetcher.py _chunk_time_range
2026-07-31	v0.3.1	凭证密钥变更后重新写入 4 个数据源	sync_source_config_service.py
2026-08-09	v0.4	SCRM 同步体系（operator/session/transfer 三表 + 增量 + Token 自动刷新）	scrm_sync_service.py / scrm_client.py
2026-08-10	v0.4	OCR 自动登录置信度预检（ddddocr + 概率检查 + 内层预检循环）	captcha_session.py
2026-08-12	v0.4	SyncDashboard 统一入口（BMS+SCRM 合并表单 + 实时日志面板）	SyncDashboard.vue / Ingest.vue
2026-08-14	v0.4	BMS 查询按钮点击策略升级（5 层兜底 + SPA 渲染就绪守卫）	bms_fetcher.py _click_query_button
2026-08-17	v0.4	alembic migration 补齐（merge head + Group/Specialty 6 表）	alembic/versions/m0a1b2c3d4e5 + n1b2c3d4e5f6
2026-08-19	v0.6	采集并发优化：submit_multi 一次任务内 asyncio.gather 并发执行 BMS/SCRM runner，BMS 下载冷却期穿插 SCRM；同步完成统一失效 today:*/analytics:* 缓存	sync_task_manager.py + bms_scheduler.py + miniapp.py
2026-08-25	v0.7	缓存加速体系：路由层 cached_call(60/90/120s) + 缓存预热(同步后/启动后/空闲期, WARM_TTL=180s) + 管理员强制重建(cache_admin.py) + 写入高峰期 TTL 放大防雪崩。新增数据源/看板接口的缓存接入清单与命名规范见 数据源接入与缓存加速规范.md（含已知缺口 G1~G4）	cache_warmer.py / cache_admin.py / redis_client.py
2026-09-01	v0.8	BMS 时间窗口可靠性修复（生产审计）：① 容器浏览器时区 UTC → new_context(timezone_id='Asia/Shanghai') + fill_utc=False（北京时间直填）；② ticket 时间标签从 label 改 placeholder 匹配 + 先点 .show-more 展开高级筛选 + click 聚焦后再 fill；③ _set_time_range 兜底 evaluate 内 :visible 非法选择器 → 改为类名选中 + JS rect/computedStyle 过滤；④ wallet 负数"新增"改为 max(..,0) + WARNING 告警；⑤ wallet 导入输出实际 op_time 范围二次校验；⑥ 下载 response listener 后台 _read_body 任务追踪清理（防 asyncio 泄漏）	bms_fetcher.py / upsert.py

---

八、SCRM 同步体系（v0.4 新增）

8.1 数据源

SCRM（企业微信客服）通过 REST API 同步，3 张数据表：

表	用途	同步方式
scrm_operators	客服操作员	全量 upsert（覆盖更新）
scrm_sessions	客服会话（含质量指标）	增量 upsert
scrm_session_transfers	会话转接记录	先删后插（幂等）

8.2 同步策略

• 增量同步：从上次成功结束时间减 2 小时重叠开始，首次运行回溯 30 天
• 定时调度：与 BMS 共享调度（8-23/0 每小时 + 5:00 补跑），BMS 与 SCRM 经 submit_multi 并发执行（BMS 冷却期间穿插 SCRM）
• 全局互斥锁 + 冷却期：与 BMS 共用 SyncTaskManager，防止手动/定时冲突
• 独立运行：BMS 或 SCRM 任一方无凭证时，另一方可独立运行

8.3 Token 管理

• Token 自动刷新：提前 5 分钟刷新
• asyncio.Lock 双重检查锁：并发安全
• 401/403 自动重试：一次重试后仍失败才报错
• 凭证存储：ClientId/ClientSecret 存于 sync_source_configs 表，Fernet 对称加密

8.4 数据处理规则

• 操作员全量 upsert（覆盖更新，不保留历史）
• 会话质量字段（first_response_sec / avg_response_sec 等）非空时不覆盖，防止回滚
• 会话转接先删后插，保证幂等
• SCRM 聊天记录不同步到本地（数据量过大）
• 同步日志跟踪 API 调用次数、reconcile delta（API 总数 - inserted/updated）、inserted/updated/failed 计数

8.5 API 端点

端点	方法	说明
/fetch/scrm/status	GET	SCRM 同步状态摘要
/fetch/scrm/sync	POST	提交 SCRM 同步任务
/fetch/scrm/logs	GET	SCRM 同步日志
/fetch/schedule/status	GET	调度状态（含 has_credentials: {bms, scrm}）
/fetch/schedule/start	POST	启动调度器（仅需一个数据源配置即可）

8.6 会话列表 API 自动分页

3 个会话列表 API 实现自动分页和合并：

• getUserChatListByPage：按页获取
• getUserChatListByGroup：按组获取
• getVirtualChatSessionPage：虚拟会话分页

8.7 错误处理

SCRM API 错误统一封装为 ScrmApiError，包含 code、message 和 raw_response。

---

九、OCR 自动登录（v0.4 新增）

9.1 应用场景

BMS 平台登录需输入图形验证码，使用 ddddocr 进行 OCR 自动识别。

9.2 置信度预检机制

常量	默认值	说明	位置
_OCR_MIN_CHAR_PROB	0.72	单字符最低概率	captcha_session.py
_OCR_AVG_PROB	0.86	平均概率阈值	captcha_session.py
_OCR_PRECHECK_MAX_REFRESH	20	最大刷新次数	captcha_session.py
_OCR_FALLBACK_CONFIDENCE	0.70	兜底置信度（ddddocr 不支持概率输出时）	captcha_session.py

9.3 登录流程

内层预检循环 (最多20次):
  1. 刷新验证码
  2. OCR 识别 (ddddocr)
  3. 提取 CTC 输出的字符概率 (30个时间步 × 非空字符 max 概率)
  4. 检查: 单字符概率 ≥ 0.72 且 平均概率 ≥ 0.86?
     是 → 进入外层登录
     否 → 回到步骤1 (不触发登录,不产生"验证码错误")

外层登录:
  5. 输入用户名/密码/验证码
  6. 提交登录
  7. 检查响应:
     "用户名或密码错误" → 立即终止 (不重试)
     "验证码错误" → 回到内层预检循环
     成功 → 完成

9.4 性能要求

• 整体流程约 3 秒完成（典型场景）
• 验证码刷新不触发服务端"验证码错误"（仅前端刷新图片）

9.5 Playwright 上下文复用

• 同域名数据源复用浏览器上下文，避免重复登录
• BmsFetcher 对同 base_url 强制 65 秒等待
• 跨域名可立即切换

9.6 BMS 查询按钮点击策略（5 层兜底）

BMS 平台页面改版导致查询按钮 DOM 变化，采用 5 层兜底：

层级	策略	选择器
0	快速直击	button:has-text('搜索') force-click
1	标准 button	button:has-text('查询')
2	div/span role	div[role='button'], span[role='button']
3	class 匹配	a[class*=btn], div[class*=el-button]
4	aria-label	aria-label*='搜索/查询'
5	终极兜底	JS form.submit() / class*=submit-btn

SPA 渲染就绪守卫：页面加载 3 秒后再轮询 20 秒等待可点击元素出现。

---

十、SyncDashboard 统一入口（v0.4 新增）

10.1 设计目标

将分散的数据采集功能合并为统一入口，Tab 切换：

Tab	功能	权限
Data View	数据浏览	所有用户
BMS Sync	BMS 同步	所有用户
SCRM Sync	SCRM 同步	所有用户
File Upload	文件上传	所有用户
Collection Records	采集记录	所有用户
Sync Settings	同步设置	仅管理员
Data Source Management	数据源管理	仅管理员

默认打开 BMS Sync Tab。

10.2 界面布局

• 左右两栏：左侧主内容区，右侧 420px 固定实时日志面板
• 响应式：屏幕 <1200px 切换为上下布局
• 日志面板高度：calc(100vh - 88px) 独立垂直滚动

10.3 实时日志面板

• 自动滚动：上滑暂停，回底部恢复
• 清空按钮：清空当前视图
• 底部返回：跳转到最新日志
• 彩色来源标签：BMS=蓝色 / SCRM=绿色
• 进度条 + 阶段标签：顶部显示当前执行阶段
• 结果摘要标签：底部显示彩色结果统计
• 新任务自动重置：新同步任务自动重置日志偏移并清空上一任务日志

10.4 合并同步表单

• BMS + SCRM 数据源复选框统一选择
• 统一时间范围选择
• 同步任务并发执行（BMS 与 SCRM 各自 runner 经 asyncio.gather 并发）
• 按钮动态显示：开始同步(共 7 个源 · BMS 3 + SCRM 4 · 增量接续)
• 运行中任务进度条置顶通知
• Steps 进度组件使用后端 runningTask.sources 显示准确执行顺序

10.5 路由兼容

旧路由自动重定向到统一入口对应 Tab：

• /data/fetch → /data/ingest?tab=fetch
• /data/upload → /data/ingest?tab=upload
• /data/records → /data/ingest?tab=records
• /data/browser → /data/ingest?tab=browser
• ?ingest=fetch → /data/ingest?tab=fetch

旧独立页面保留在隐藏路由中，向后兼容。