lami/lami-base-v1/backend/tests_python/export_live_data.py

#!/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())