xinyouhe/yunke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
73fd9d0975c2140afd940be690ae3c17238b07d5
yunke/docs/superpowers/plans/2026-08-01-yunke-data-mirror.md
T
浅念 cd8d36fd00 docs: initial design spec, data dictionary, implementation plan 
Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-01 10:53:28 +08:00

16 KiB
Raw Blame History

云客数据本地化系统 实现计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 持续抓取云客 CRM（微信+WhatsApp+电话+短信）数据到本地 MySQL + 对象存储，提供自有查看后台与增量同步。

Architecture: yunke-server/（Python/FastAPI）含会话层、抓取层（每渠道一个 scraper，混合 HTTP 重放 + 浏览器兜底）、存储层（MySQL 分表 + 对象存储哈希去重）、调度层（水位增量同步）、API 层；yunkes/（Vue3+TS）查看后台。先做 WeChat 端到端骨架，再横向扩展渠道与纵向补全实体。

Tech Stack: Python 3.11, FastAPI, SQLAlchemy 2.0 + Alembic, MySQL(asyncmy)/Redis, httpx, Playwright, APScheduler, Pydantic v2, pytest(+asyncio/respx) | Vue3 + Vite + TS + Pinia + Vue Router + Element Plus + axios

配套文档： 设计文档、数据字典

说明： 项目非 git 仓库；每个任务结尾的"commit"步骤在初始化 git 后执行（Phase 0 之后）。

文件结构

yunke-server/
  pyproject.toml  .env.example  alembic.ini  README.md
  app/
    config.py                 # 环境配置(pydantic-settings)
    db/base.py                # engine/session/Base
    db/models.py              # 全部 ORM 模型
    storage/object_store.py   # 对象存储抽象(本地磁盘/可换 MinIO)
    session/login.py          # getSecureKey -> pcLogin
    session/yunke_client.py   # 带鉴权头的 HTTP 客户端 + 自动续期
    scraper/base.py           # 分页/游标/限速 基类
    scraper/parsers.py        # 原始 type 码 -> 统一枚举
    scraper/wechat.py         # 微信 scraper
    scraper/whatsapp.py       # WhatsApp scraper
    scraper/call.py           # 电话+短信 scraper
    scraper/media.py          # 媒体下载器(每渠道策略)
    sync/watermarks.py        # sync_state 读写
    sync/orchestrator.py      # 全量+增量编排
    sync/scheduler.py         # 定时任务
    api/main.py               # FastAPI app
    api/deps.py
    api/routers/{customer,group,message,call,sms,media,search,sync}.py
    api/schemas.py            # 响应模型
  migrations/                 # alembic
  tests/                      # 镜像结构, +fixtures/cassettes(录制的真实响应)
yunkes/                       # Vue3 前端(见 M5)

Phase 0：侦察收尾（实现期首批，前置依赖）

用浏览器抓包方法（见数据字典 Phase 0）确认以下，结果写入 docs/superpowers/specs/phase0-findings.md：

  • 聊天媒体：微信 file URL 形态；WhatsApp mediaName 下载接口（OSS 桶已确认 yunke-pcfile）
  • 媒体类型码：微信/WhatsApp image/voice/video/file/transfer/redpacket/system 的 type 枚举
  • 语音通话字段：callRecord/queryWeChatVoiceList、WhatsApp wavoicecalllist
  • WhatsApp 联系人 businessFriendList 字段
  • 登录加密：getSecureKey/pcLogin 请求响应
  • 群成员昵称：memberIds 的 wxid 是否需另调接口取群昵称
  • 限速阈值观察
  • 初始化 git：cd d:\refactor\yunke && git init（此后所有 commit 步骤生效）

Phase 0 产出 phase0-findings.md 后，scraper 任务据此填实精确字段；架构不依赖它。

Milestone 1：地基 + 微信端到端骨架（Walking Skeleton）

目标： 登录 → 抓微信客户列表 + 单聊消息 → 存 MySQL → FastAPI 暴露 → 跑通。证明整条链路。

Task 1.1：项目脚手架与依赖

Files: Create yunke-server/pyproject.toml, yunke-server/.env.example, yunke-server/app/__init__.py, yunke-server/app/config.py

  • Step 1: 写 pyproject.toml（依赖：fastapi, uvicorn, sqlalchemy[asyncio], asyncmy, alembic, redis, httpx, playwright, apscheduler, pydantic, pydantic-settings, cryptography; dev: pytest, pytest-asyncio, respx, ruff）
  • Step 2: 写 .env.example（YUNKE_BASE_URL, YUNKE_USERNAME, YUNKE_PASSWORD, DB_URL=mysql+asyncmy://..., REDIS_URL, MEDIA_DIR, ENCRYPTION_KEY）
  • Step 3: 写 app/config.py（pydantic-settings Settings，从环境读取，单例 get_settings()）
  • Step 4: pip install -e ".[dev]" 并 playwright install chromium
  • Step 5: commit chore: scaffold yunke-server

Task 1.2：数据库与配置（TDD）

Files: Create app/db/base.py, tests/conftest.py, tests/test_config.py

  • Step 1: 写失败测试 tests/test_config.py：断言 get_settings().db_url 能从环境读取（用 monkeypatch.setenv）
  • Step 2: 跑测试确认失败 pytest tests/test_config.py -v → FAIL
  • Step 3: 实现 app/config.py 的 Settings；app/db/base.py：async_engine、async_session_maker、Base = DeclarativeBase
  • Step 4: 跑测试确认通过
  • Step 5: commit

Task 1.3：核心 ORM 模型

Files: Create app/db/models.py

实现（按数据字典）：Department、Staff、WechatAccount、WhatsappAccount、Customer、ChatGroup、GroupMember、Message(按月分表命名规范预留)、Media、PhoneCall、Sms、SyncState、YkAccount、SyncRun。每表带 channel、version、raw_json(JSON)、fetched_at、updated_at。Message 主键 msg_id，联合索引 (channel, owner_account_id, talker_id, timestamp)。

  • Step 1: 写测试 tests/test_models.py：建内存 SQLite，插入/查询一条 Customer，断言字段映射与 raw_json 往返（测试用 SQLite，生产 MySQL）
  • Step 2: 跑确认失败
  • Step 3: 实现 models.py
  • Step 4: 跑确认通过
  • Step 5: alembic init migrations，生成首版迁移并 alembic upgrade head（对 MySQL）；commit feat: core ORM models

Task 1.4：会话层——登录与鉴权客户端（TDD + 录制响应）

Files: Create app/session/login.py, app/session/yunke_client.py, tests/cassettes/login.json, tests/test_yunke_client.py

  • Step 1: 把 Phase 0 抓到的登录请求/响应存为 tests/cassettes/login.json
  • Step 2: 写失败测试 test_yunke_client.py：用 respx mock /usercenter/login/pcLogin 返回 cassette，断言 YunkeClient 登录后 userid/company 头被注入；再 mock 一个数据接口，断言请求头含鉴权
  • Step 3: 跑确认失败
  • Step 4: 实现 login.py（getSecureKey→pcLogin，按 Phase 0 加密密码；返回 cookie+userid+company）；yunke_client.py（httpx.AsyncClient 封装，自动注入鉴权头，post()/get()，侦测 401→重新登录续期）
  • Step 5: 跑确认通过
  • Step 6: commit feat: session login + authed client

Task 1.5：微信客户列表 scraper（TDD + 录制响应）

Files: Create app/scraper/base.py, app/scraper/wechat.py, tests/cassettes/wxFriends_page1.json, tests/test_scrape_wechat_friends.py

  • Step 1: 存录制响应 wxFriends_page1.json（来自 friend_list/es）
  • Step 2: 写失败测试：respx mock 分页（2 页），断言 scrape_wechat_friends(client) 返回 N 条归一化 dict（含 friend_wx_id/real_name/phone/version），分页正确
  • Step 3: 跑确认失败
  • Step 4: 实现 base.py（页码分页 helper、时间游标 helper、限速）；wechat.py 的 scrape_wechat_friends（调 /wechatweb/wxFriends/friend_list/es，翻页，字段映射）
  • Step 5: 跑确认通过
  • Step 6: commit feat: wechat friend scraper

Task 1.6：微信消息 scraper（TDD + 录制响应）

Files: Modify app/scraper/wechat.py; Create tests/cassettes/msg_chats_*.json, tests/test_scrape_wechat_messages.py

  • Step 1: 存录制响应（含文字 + 一条带媒体占位 + 日期符 -2）
  • Step 2: 写失败测试：断言 scrape_wechat_messages(client, wechat_id, friend_id, start, end) 返回消息，type=-2 被过滤，字段映射（msgSvrId→msg_id、type 经 parsers 归一化、roomid 保留）
  • Step 3: 跑确认失败
  • Step 4: 实现 parsers.py（normalize_wechat_type(raw)->Enum）；wechat.py 的消息抓取（/wechatweb/msg/chats，时间游标 end/hasLast 向过去翻页）
  • Step 5: 跑确认通过
  • Step 6: commit feat: wechat message scraper

Task 1.7：存储层——幂等 upsert（TDD）

Files: Create app/storage/repos.py, tests/test_repos.py

  • Step 1: 写失败测试：插入同一 customer 两次（不同 version），断言只 1 行且 version 更新；插入重复 media(同 hash) 两次，断言 1 行
  • Step 2: 跑确认失败
  • Step 3: 实现 repos.py：upsert_customer、upsert_messages（按主键 ON DUPLICATE KEY/merge）、ObjectStore.put_by_hash（本地磁盘按 hash 存，exists 判重）；实现消息分表路由器 message_table(channel, timestamp) 选表（message_<channel>_YYYYMM），并预留跨分表查询 helper（按时间/会话扇出多表合并）
  • Step 4: 跑确认通过
  • Step 5: commit feat: idempotent upsert repos + object store

Task 1.8：FastAPI 最小 API（TDD）

Files: Create app/api/main.py, app/api/routers/customer.py, app/api/schemas.py, tests/test_api_customer.py

  • Step 1: 写失败测试：GET /api/customers?channel=wechat&page=1 返回分页列表（seed 测试数据）
  • Step 2: 跑确认失败
  • Step 3: 实现 FastAPI app、customer router（分页、筛选 channel/部门）、schemas
  • Step 4: 跑确认通过；uvicorn app.api.main:app 手测
  • Step 5: commit feat: customer list API

Task 1.9：端到端冒烟脚本

Files: Create scripts/smoke_wechat.py

  • Step 1: 写脚本：登录 → 抓 1 个微信号的好友(限 1 页) + 该号 1 个会话最近消息 → upsert → 打印统计。用真实账号小范围跑（只读）
  • Step 2: 跑通，确认数据落库
  • Step 3: commit chore: wechat end-to-end smoke；Milestone 1 完成可演示

Milestone 2：微信全量补全

每个实体同 M1 模式（录制响应→TDD→实现→commit）：

  • Task 2.0 部门树：scrape_departments（POST /usercenter/dept/tree）→ upsert Department（含 parent_id/path 层级）。这是组织树的唯一来源，必须最先抓（Task 5.2 依赖它）
  • Task 2.1 微信员工+账号：scrape_wechat_accounts（wxUser/list），顺带 upsert Staff。Staff 的 dept/phone 来源：wxUser/list 仅给 userId/userName，无 dept；dept 关系从该账号下客户的 deptRoute/userId 反推（无客户账号的员工暂无 dept）。若需员工完整字段（手机/角色/离职）→ Phase 0 确认 /pc/crm_t_portaluser/... 后补一个 staff 详情 scraper
  • Task 2.2 微信群+成员：scrape_wechat_groups（chatroom/getChatroomList），解析 memberIds → 批量 upsert ChatGroup + GroupMember
  • Task 2.3 群消息：扩展消息 scraper 支持 friendId=chatRoomId（群聊），talker 为群内成员
  • Task 2.4 微信语音通话：scrape_wechat_voice_calls（callRecord/queryWeChatVoiceList，按 Phase 0 字段）
  • Task 2.5 会话列表：scrape_wechat_talkers（talkerList）用于发现哪些会话有消息
  • Task 2.6 媒体下载（微信）：app/scraper/media.py —— 抓到带 file 的消息即下载存对象存储（按 md5 去重）；失败的入重试队列不阻塞。TDD：mock OSS 直链下载 + 去重
  • Task 2.7 水位增量：sync/watermarks.py + sync/orchestrator.py。消息按时间水位（每会话 last_timestamp，查 [水位, now]）；非消息实体（客户/群/员工/电话/短信）按 version 水位推进（查 version > 水位）。推进水位、断点续抓。TDD：模拟两轮增量，断言只取新增
  • Task 2.8 媒体读取 API：api/routers/media.py —— 从对象存储流式返回媒体（Range 请求、Content-Type、inline/attachment）。与 JSON 列表模式不同，需独立实现。TDD：seed 一个 media 记录 + 本地文件，GET /api/media/{hash} 断言 200 + Content-Type + Range 支持
  • Task 2.9 后端鉴权：加一个简单共享 token 中间件（与 Task 5.10 前端配对）；本期定位本地/内网工具，可默认信任网络边界，但显式写明信任模型，避免实现者猜测

Milestone 3：WhatsApp 渠道

（与微信同模式；复用 base/storage/sync/api，加 channel='whatsapp'）

  • Task 3.1 WhatsApp 登录：复用 YunkeClient（同会话），确认 /qqweb/WhatsAppUser/ 也用同样鉴权头
  • Task 3.2 员工+账号：businessSalesList（42）
  • Task 3.3 客户：businessFriendList（按 Phase 0 字段）
  • Task 3.4 消息：chatRecords（keyId 主键、时间游标、type 归一化）
  • Task 3.5 媒体下载（WhatsApp）：按 Phase 0 找到的 mediaName 下载接口实现专属 downloader
  • Task 3.6 语音通话：wavoicecalllist（Phase 0）
  • Task 3.7：扩展 customer/message API 支持跨渠道 channel 过滤与聚合

Milestone 4：电话 + 短信

  • Task 4.1 电话记录 scraper：/pc/callLog/list（6047，含 recordFile OSS 录音 URL），归一化到 PhoneCall
  • Task 4.2 录音下载：复用对象存储，下载 recordFile mp3（直链 OSS），按内容哈希去重
  • Task 4.3 短信 scraper：/pc/customer/smsLogsList（53），归一化到 Sms
  • Task 4.4 现场录音：Phase 0 确认后补
  • Task 4.5：call/sms API + 时间范围增量同步

Milestone 5：前端查看后台（yunkes）

Files: yunkes/（Vite 初始化 + Element Plus + Pinia + Router + axios）

  • Task 5.1 脚手架：npm create vite@latest yunkes -- --template vue-ts；装 element-plus/pinia/vue-router/axios；配 axios 指向后端
  • Task 5.2 布局：左侧组织树（部门→人员→账号）+ 顶栏 + 主区（Element Plus 后台布局）
  • Task 5.3 客户列表页：分页表格 + 筛选 → 点昵称进时间轴
  • Task 5.4 消息时间轴：按时间正序渲染文字/语音(播放+转文字)/图片/视频/文件/转账红包；单聊+群聊（群显示发送成员）
  • Task 5.5 群列表 + 群成员
  • Task 5.6 通话/短信页
  • Task 5.7 全文搜索：跨渠道/日期/人员/客户/群（MVP 应用层扇出，见设计 §8）
  • Task 5.8 媒体查看：图片预览、语音/视频播放、文件下载
  • Task 5.9 同步状态面板：各实体上次同步时间/进度/错误
  • Task 5.10 简单鉴权：单一账号密码（本地工具）

Milestone 6：调度与稳健化

  • Task 6.1 定时调度：sync/scheduler.py（APScheduler + Redis）：每日全实体增量 + 手动触发接口
  • Task 6.2 重试与限速：失败指数退避重试队列；请求限速器（避免风控）
  • Task 6.3 登录态自动续期：401 自动重新登录
  • Task 6.4 监控日志：SyncRun 记录每次运行统计与错误
  • Task 6.5 数据脱敏/留底校验：确认 raw_json 全留底
  • Task 6.6 部署文档：docker-compose（MySQL/Redis/app/yunkes）+ README

测试策略贯穿

  • 所有 scraper 用 tests/cassettes/ 录制响应 + respx，不依赖线上
  • storage/sync 用 SQLite 内存库做单测
  • API 用 FastAPI TestClient
  • 每个 Milestone 结束做一次真实账号小范围只读冒烟

风险锚点

  • 媒体外链过期 → 抓到即下载（M2.6/M3.5/M4.2）
  • 云客改版 → scraper 集中封装 + raw_json 留底 + 浏览器兜底（base.py 预留 BrowserFallback）
  • 大规模 → 消息分表 + 增量水位 + 媒体哈希去重
Reference in New Issue View Git Blame Copy Permalink