#!/usr/bin/env python3 """ 导出 QiWe 真实联调数据（给交付 / 上级查看） 不跑本地假数据用例，只请求真实企微接口，生成： - live_data_report.md 人类可读 - live_data_snapshot.json 原始摘要（可贴进报告） 用法（先 pnpm dev）： cd backend/tests_python python export_live_data.py """ from __future__ import annotations import json import sys from datetime import datetime, timezone from pathlib import Path import requests from client import ApiClient from load_env import load_backend_env ROOT = Path(__file__).resolve().parent PC_BASE = "http://localhost:3101/api" OUT_MD = ROOT / "live_data_report.md" OUT_JSON = ROOT / "live_data_snapshot.json" def _session_room_ids(data: dict) -> list[str]: ids: list[str] = [] for key in ("collectList", "shieldList", "topList", "markList"): for item in data.get(key) or []: if item.get("sessionType") == 1 and item.get("sessionId"): ids.append(str(item["sessionId"])) return ids def _parse_room_ids(env: dict[str, str]) -> list[str]: raw = env.get("QIWEI_ROOM_IDS", "").strip() return [x.strip() for x in raw.split(",") if x.strip()] def main() -> int: env = load_backend_env() guid = env.get("QIWEI_GUID", "").strip() token_ok = bool(env.get("QIWEI_TOKEN")) room_ids = _parse_room_ids(env) if not token_ok or not guid: print("请在 backend/.env 配置 QIWEI_TOKEN 与 QIWEI_GUID") return 1 try: requests.get(f"{PC_BASE}/health", timeout=3).raise_for_status() except requests.RequestException: print("PC 服务未启动，请先: cd backend && pnpm dev") return 1 api = ApiClient(PC_BASE) snapshot: dict = { "exportedAt": datetime.now(timezone.utc).isoformat(), "guid": guid, "roomIdsConfigured": room_ids, "sections": {}, } lines = [ "# QiWe 真实联调数据导出", "", f"- 导出时间（UTC）：{snapshot['exportedAt']}", f"- GUID：`{guid}`", f"- 配置群数：{len(room_ids)}", "", "> 由 `python export_live_data.py` 生成，数据来自真实 QiWe API。", "", ] def section(title: str, key: str, fn): print(f" 拉取: {title}...") try: data = fn() snapshot["sections"][key] = data lines.append(f"## {title}") lines.append("") lines.append("```json") lines.append(json.dumps(data, ensure_ascii=False, indent=2)) lines.append("```") lines.append("") return data except Exception as e: err = {"error": str(e)} snapshot["sections"][key] = err lines.append(f"## {title}") lines.append("") lines.append(f"**失败：** {e}") lines.append("") return None # 1. 在线状态 staff = section( "1. 员工在线状态", "staffStatus", lambda: api.assert_success(api.get(f"/qiwei/staff/{guid}/status")), ) if staff: lines.append( f"**摘要：** 昵称/账号 `{staff.get('nickname') or staff.get('userId')}`，" f"在线状态 `{staff.get('userOnlineStatus')}`" ) lines.append("") # 2. 会话列表（真实群 id 来源） session_data = section( "2. 会话列表 getSessionList", "sessionList", lambda: api.assert_success( api.post( "/qiwei/proxy", json={"method": "/session/getSessionList", "params": {"guid": guid}}, ) ), ) session_rooms = _session_room_ids(session_data or {}) if session_data: lines.append(f"- 会话中群聊（sessionType=1）共 **{len(session_rooms)}** 个") lines.append("") # 3. getRoomList（可能为空） room_list = section( "3. 群分页 getRoomList（可能为空）", "getRoomList", lambda: api.assert_success( api.post( "/qiwei/proxy", json={ "method": "/room/getRoomList", "params": {"guid": guid, "nextStartIndex": 0}, }, ) ), ) # 4. 按 QIWEI_ROOM_IDS 同步群 ids_to_sync = room_ids or session_rooms[:5] if ids_to_sync: synced = section( "4. 按 roomIdList 同步群（真实群名/人数）", "roomsSyncByIds", lambda: api.assert_success( api.post( "/rooms/sync", json={"guid": guid, "roomIdList": ids_to_sync}, ) ), ) if synced and synced.get("rooms"): lines.append("### 群摘要") lines.append("") lines.append("| roomId | 群名 | 人数 |") lines.append("|--------|------|------|") for r in synced["rooms"]: lines.append( f"| {r.get('roomId')} | {r.get('roomName', '')} | {r.get('memberCount', '')} |" ) lines.append("") else: snapshot["sections"]["roomsSyncByIds"] = { "skipped": "未配置 QIWEI_ROOM_IDS 且会话列表无群" } # 5. 每个群的详情与健康度 for i, rid in enumerate(ids_to_sync or [], 1): detail = section( f"5.{i} 群详情 GET /rooms/{rid}", f"roomDetail_{rid}", lambda r=rid: api.assert_success(api.get(f"/rooms/{r}")), ) section( f"5.{i}b 群健康度", f"roomHealth_{rid}", lambda r=rid: api.assert_success(api.get(f"/rooms/{r}/health")), ) # 6. 同步消息（真实消息条数） msgs = section( "6. 同步消息 syncMsg（最近若干条）", "syncMessages", lambda: api.assert_success( api.post("/qiwei/sync", json={"guid": guid, "msgSeq": 0, "limit": 10}) ), ) if msgs: n = len(msgs.get("messages") or []) lines.append(f"- 本批拉取消息数：**{n}**，hasMore={msgs.get('hasMore')}") lines.append("") # 7. 外部联系人 contacts = section( "7. 外部联系人同步", "contactsSync", lambda: api.assert_success( api.post( "/contacts/sync", json={"guid": guid, "currentSeq": 0, "limit": 10}, ) ), ) if contacts: lines.append(f"- 联系人数：**{len(contacts.get('contacts') or [])}**") lines.append("") if ids_to_sync: import json as _json import time as _time import uuid as _uuid fix = Path(__file__).parent / "fixtures" / "qiwei_webhook_sample.json" wh = _json.loads(fix.read_text(encoding="utf-8")) rid = ids_to_sync[0] docid = f"DOC_EXPORT_{_uuid.uuid4().hex[:8]}" ts = int(_time.time()) for i, item in enumerate(wh.get("data", [])): item["guid"] = guid item["fromRoomId"] = rid item["timestamp"] = ts + i if item.get("msgType") == 13: item["msgData"]["linkUrl"] = ( f"https://doc.weixin.qq.com/txdoc/excel?docid={docid}" ) section("8. Webhook 登记群文档", "webhookLive", lambda: api.assert_success( api.post("/qiwei/webhook", json=wh) )) section( "8b. 群文档台账", "roomDocLive", lambda: api.assert_success(api.get("/qiwei/room-docs", params={"roomId": rid})), ) msgs = section( "8c. 已存消息列表", "messagesLive", lambda: api.assert_success( api.get("/qiwei/messages", params={"roomId": rid, "limit": 10}) ), ) if msgs is not None: lines.append(f"- 消息条数：**{len(msgs)}**") lines.append("") if env.get("QIWEI_LIVE_NOTIFY", "").strip() == "1" and ids_to_sync: room_name = "" rooms_data = snapshot["sections"].get("roomsSyncByIds") or {} if rooms_data.get("rooms"): room_name = rooms_data["rooms"][0].get("roomName", "") tpl = section( "9. 整改通知模板", "notifyTemplate", lambda: api.assert_success( api.get( "/notifications/template", params={"issueType": "missing_doc", "roomName": room_name or "联调群"}, ) ), ) if tpl and ids_to_sync: section( "10. 真实发送通知 sendText（发到第一个配置群）", "notifySendLive", lambda: api.assert_success( api.post( "/notifications/send", json={ "type": "compliance", "receiverId": 1, "title": tpl["title"], "content": tpl["content"] + " [export_live_data]", "refId": 1, "guid": guid, "toId": ids_to_sync[0], }, ), status=201, ), ) OUT_JSON.write_text( json.dumps(snapshot, ensure_ascii=False, indent=2), encoding="utf-8" ) OUT_MD.write_text("\n".join(lines), encoding="utf-8") print() print("真实联调数据已导出：") print(f" {OUT_MD}") print(f" {OUT_JSON}") return 0 if __name__ == "__main__": raise SystemExit(main())