如何通过 API 获取 XChat 群组 DM 历史

curl --request POST \
  --url https://api.twexapi.io/v3/twitter/dm-history \
  --header "Authorization: Bearer $TWEXAPI_BEARER_TOKEN" \
  --header "Content-Type: application/json" \
  --data '{
    "recipient": "g1234567890",
    "cookie": "auth_token=...; ct0=...; twid=...",
    "pin": "1234",
    "count": 50,
    "all": false
  }'

1{
2  "code": 200,
3  "msg": "success",
4  "data": {
5    "conversation_id": "g1234567890",
6    "has_more": true,
7    "messages": [
8      {
9        "id": "1234567890123456789",
10        "text": "发布报告已经可以审核了。",
11        "time": "2024-01-01T12:00:00Z",
12        "sequence_id": "2076661179334979585",
13        "sender_id": "1928096185664843776",
14        "recipient_id": "g1234567890",
15        "attachment": null,
16        "attachments": null
17      }
18    ]
19  }
20}

1import os
2from datetime import datetime, timezone
3from typing import Any
4
5import requests
6
7API_URL = "https://api.twexapi.io/v3/twitter/dm-history"
8BEARER_TOKEN = os.environ["TWEXAPI_BEARER_TOKEN"]
9X_COOKIE = os.environ["TWEXAPI_X_COOKIE"]
10XCHAT_PIN = os.environ.get("TWEXAPI_XCHAT_PIN", "1234")
11PROXY = os.environ.get("TWEXAPI_PROXY")
12
13GROUPS = {
14    "launch_ops": "g1234567890",
15    "creator_review": "g9876543210",
16}
17
18def resolve_group_id(group_key: str) -> str:
19    group_id = GROUPS.get(group_key)
20    if not group_id:
21        raise ValueError(f"Unknown group key: {group_key}")
22    if not group_id.startswith("g"):
23        raise ValueError("XChat group recipient must start with 'g'")
24    return group_id
25
26def fetch_group_history_page(
27    group_key: str,
28    *,
29    before: str | None = None,
30    count: int = 50,
31) -> dict[str, Any]:
32    if count < 10 or count > 200:
33        raise ValueError("count must be between 10 and 200")
34
35    payload: dict[str, Any] = {
36        "recipient": resolve_group_id(group_key),
37        "cookie": X_COOKIE,
38        "pin": XCHAT_PIN,
39        "count": count,
40        "all": False,
41    }
42    if before:
43        payload["before"] = before
44    if PROXY:
45        payload["proxy"] = PROXY
46
47    response = requests.post(
48        API_URL,
49        headers={
50            "Authorization": f"Bearer {BEARER_TOKEN}",
51            "Content-Type": "application/json",
52            "Accept": "application/json",
53        },
54        json=payload,
55        timeout=60,
56    )
57    response.raise_for_status()
58    data = response.json()
59
60    if data.get("code") != 200:
61        raise RuntimeError(data.get("msg", "TwexAPI returned an error"))
62
63    return data["data"]
64
65def normalize_group_message(
66    group_key: str,
67    group_id: str,
68    conversation_id: str,
69    message: dict[str, Any],
70) -> dict[str, Any]:
71    return {
72        "group_key": group_key,
73        "group_id": group_id,
74        "conversation_id": conversation_id,
75        "message_id": message["id"],
76        "text": message.get("text"),
77        "time": message.get("time"),
78        "sequence_id": message.get("sequence_id"),
79        "sender_id": message.get("sender_id"),
80        "recipient_id": message.get("recipient_id"),
81        "attachment": message.get("attachment"),
82        "attachments": message.get("attachments"),
83        "fetched_at": datetime.now(timezone.utc).isoformat(),
84        "raw": message,
85    }
86
87def export_group_history(
88    group_key: str,
89    *,
90    max_pages: int = 10,
91) -> list[dict[str, Any]]:
92    before = None
93    rows: list[dict[str, Any]] = []
94    group_id = resolve_group_id(group_key)
95
96    for _ in range(max_pages):
97        page = fetch_group_history_page(group_key, before=before)
98        conversation_id = page["conversation_id"]
99        messages = page.get("messages") or []
100
101        rows.extend(
102            normalize_group_message(group_key, group_id, conversation_id, msg)
103            for msg in messages
104        )
105
106        if not page.get("has_more") or not messages:
107            break
108
109        oldest = min(
110            messages,
111            key=lambda msg: int(msg.get("sequence_id") or 0),
112        )
113        before = oldest.get("sequence_id")
114        if not before:
115            break
116
117    return rows
118
119if __name__ == "__main__":
120    exported = export_group_history("launch_ops", max_pages=5)
121    print(f"Exported {len(exported)} group messages")