如何用 TwexAPI 获取 XChat DM 历史

私信历史只有在能解释采集过程时才有价值。客服工单、CRM 同步、合作记录都需要 conversation ID、message ID、时间、sender 和 recipient、附件，以及说明导出停在哪里的分页游标。

TwexAPI 的 Get DM History (XChat v3) 端点可以读取 XChat 会话历史。一对一场景里，recipient 可以是 user ID、@handle，也可以是 1928096185664843776:1989842918455291904 这种 conversation ID。

端点和请求字段

使用 POST /v3/twitter/dm-history 读取 XChat 会话。

字段	类型	必填	说明
`recipient`	string	是	User ID、`@handle`、conversation ID `a:b`，或 group ID `g...`。
`cookie`	string	是	X 认证 cookie 或 `auth_token`。只保存在服务端。
`pin`	string 或 null	否	Identity PIN。API 默认值是 `1234`。
`count`	integer 或 null	否	每页数量，范围 10 到 200，默认 50。
`all`	boolean 或 null	否	为 `true` 时会跟随 `has_more` 返回所有页面。大历史慎用。
`before`	string 或 null	否	手动分页时，传上一页最旧消息的 `sequence_id`。
`proxy`	string 或 null	否	可选 HTTP proxy。

基础请求

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": "@example_user",
    "cookie": "auth_token=...; ct0=...; twid=...",
    "pin": "1234",
    "count": 50,
    "all": false
  }'

响应会返回 conversation_id、has_more 和 messages。

1{
2  "code": 200,
3  "msg": "success",
4  "data": {
5    "conversation_id": "1928096185664843776:1989842918455291904",
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": "1989842918455291904",
15        "attachment": null,
16        "attachments": null
17      }
18    ]
19  }
20}

如果 has_more 为 true，取当前页最旧消息的 sequence_id，在下一次请求里作为 before 传入。

Python 手动分页

手动分页比 all=true 更容易做 checkpoint。建议先写入当前页，再请求下一页。

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
13def fetch_dm_page(
14    recipient: str,
15    *,
16    before: str | None = None,
17    count: int = 50,
18) -> dict[str, Any]:
19    if count < 10 or count > 200:
20        raise ValueError("count must be between 10 and 200")
21
22    payload: dict[str, Any] = {
23        "recipient": recipient,
24        "cookie": X_COOKIE,
25        "pin": XCHAT_PIN,
26        "count": count,
27        "all": False,
28    }
29    if before:
30        payload["before"] = before
31    if PROXY:
32        payload["proxy"] = PROXY
33
34    response = requests.post(
35        API_URL,
36        headers={
37            "Authorization": f"Bearer {BEARER_TOKEN}",
38            "Content-Type": "application/json",
39            "Accept": "application/json",
40        },
41        json=payload,
42        timeout=60,
43    )
44    response.raise_for_status()
45    data = response.json()
46
47    if data.get("code") != 200:
48        raise RuntimeError(data.get("msg", "TwexAPI returned an error"))
49
50    return data["data"]
51
52def normalize_message(conversation_id: str, message: dict[str, Any]) -> dict[str, Any]:
53    return {
54        "conversation_id": conversation_id,
55        "message_id": message["id"],
56        "text": message.get("text"),
57        "time": message.get("time"),
58        "sequence_id": message.get("sequence_id"),
59        "sender_id": message.get("sender_id"),
60        "recipient_id": message.get("recipient_id"),
61        "attachment": message.get("attachment"),
62        "attachments": message.get("attachments"),
63        "fetched_at": datetime.now(timezone.utc).isoformat(),
64        "raw": message,
65    }
66
67def export_history(recipient: str, *, max_pages: int = 10) -> list[dict[str, Any]]:
68    before = None
69    rows: list[dict[str, Any]] = []
70
71    for _ in range(max_pages):
72        page = fetch_dm_page(recipient, before=before)
73        conversation_id = page["conversation_id"]
74        messages = page.get("messages") or []
75
76        rows.extend(normalize_message(conversation_id, msg) for msg in messages)
77
78        if not page.get("has_more") or not messages:
79            break
80
81        oldest = min(
82            messages,
83            key=lambda msg: int(msg.get("sequence_id") or 0),
84        )
85        before = oldest.get("sequence_id")
86        if not before:
87            break
88
89    return rows
90
91if __name__ == "__main__":
92    exported = export_history("@example_user", max_pages=5)
93    print(f"Exported {len(exported)} messages")

生产环境里，应该在请求下一页前先持久化当前页。如果任务中断，就从最后保存的 before 继续，而不是重新从第一页开始。

什么时候用 all=true

all=true 适合很小的会话，或者明确需要一次性导出的内部工具。客服系统、CRM 同步、长期归档更适合手动分页，因为每一页都有 checkpoint，失败后更容易恢复。

只有在确认会话足够小，且不需要页级恢复时，再使用 all=true。

应该保存哪些字段

字段	作用
`conversation_id`	API 返回的稳定会话键。
`message_id`	去重，并关联已发送消息。
`sequence_id`	分页 checkpoint 和排序信号。
`time`	会话时间线。
`sender_id` 和 `recipient_id`	判断消息方向和参与者映射。
`text`	用于搜索或工单上下文。
`attachment` 和 `attachments`	媒体证据和 XChat v3 附件元数据。
`fetched_at`	说明快照采集时间。
原始 message JSON	字段映射变化时可重新处理。

不要把账号 cookie 存在消息记录里。保存凭据引用即可。

常见错误

对很大的历史记录直接使用 all=true，导致无法做页级恢复。
当前页还没写入，就提前保存 before。
用最新消息的 sequence_id 分页，而不是当前页最旧消息的 sequence_id。
把 @handle 当成永久身份；应该保存返回的 conversation_id。
因为第一次测试只有文本消息，就丢弃附件字段。

总结

POST /v3/twitter/dm-history 可以按 user ID、@handle 或 conversation ID 读取 XChat 历史。更可靠的方式是手动分页：请求一页，保存规范化消息和原始 JSON，用最旧消息的 sequence_id 作为下一页 before，并在 has_more 为 true 时继续。