如何用 TwexAPI 获取 XChat 群组 DM 历史

群组 DM 历史比一对一导出更需要结构。一个消息流里可能同时有客户、客服负责人、活动运营和多个审核人，所以导出时不应该只保存文本，还要保留 group ID、sender ID、时间、附件，以及说明归档停在哪里的分页游标。

TwexAPI 的 Get DM History (XChat v3) 端点通过 POST /v3/twitter/dm-history 读取 XChat 历史。群组场景下，把已有的 g... group ID 作为 recipient 传入。

这个端点只读取已有会话。它不会创建群、添加成员，也不会发送消息。

端点和群组请求字段

使用 POST /v3/twitter/dm-history，并传入已知的 XChat group ID。

字段	类型	必填	说明
`recipient`	string	是	以 `g...` 开头的已有 group ID。
`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": "g1234567890",
    "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": "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}

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

Python 群组历史导出脚本

建议使用 group registry，而不是让未审核的用户输入直接传原始 g... 值。这样导出结果才能绑定到明确的业务用途。

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")

生产环境里，请先写入当前页，再请求下一页。如果导出在第三页后失败，就从最后保存的 before 继续，而不是重新拉完整归档。

群组历史工作流

登记群组：保存允许读取的 g... ID、内部 key、负责人、用途和允许读取的账号。
请求一页：调用 POST /v3/twitter/dm-history，设置 all: false，页大小保持在 10 到 200。
持久化当前页：先保存规范化行和原始 message JSON，再更新 checkpoint。
推进游标：用当前页最旧消息的 sequence_id 作为下一页 before。
明确停止：当 has_more 为 false、没有消息，或达到任务配置的页数上限时停止。

页数上限是故意设计的。群组历史可能很长，有边界的任务更容易监控和重试。

什么时候用 all=true

all=true 适合很小的群组会话，且你明确希望一次请求导出全部历史。它不适合运营日志、客户证据或偏合规的归档，因为失败后缺少页级恢复点。

定期导出建议保持 all: false，保存游标，并逐页推进历史。

应该保存哪些字段

字段	作用
`group_key`	便于人理解的内部群组标签。
`group_id`	请求中使用的 XChat `g...` recipient 值。
`conversation_id`	API 返回的会话键。
`message_id`	去重和查询。
`sequence_id`	分页 checkpoint 和排序信号。
`sender_id`	说明消息由哪位参与者发送。
`recipient_id`	确认消息属于预期群组。
`time`	会话时间线。
`text`	搜索、工单备注和复核上下文。
`attachment` 和 `attachments`	媒体证据和 XChat v3 附件元数据。
`fetched_at`	说明快照采集时间。
原始 message JSON	字段映射变化时可以重新处理。

不要把完整账号 cookie 存在归档表里。只保存凭据引用，把密钥留在 secret manager。

常见错误

本来要读取群组，却传了 user ID 或 @handle。
让用户在导出表单里直接输入未复核的 g... ID。
用最新消息的 sequence_id 作为 before。
对很长的群组历史使用 all=true，导致无法做 checkpoint 恢复。
因为第一页只有文本消息，就丢弃附件字段。

总结

使用 POST /v3/twitter/dm-history，并把 recipient 设置为 g...，即可读取已有 XChat 群组历史。更可靠的模式是维护 group registry、用最旧 sequence_id 手动分页、保存原始 JSON，并记录能干净恢复的 checkpoint。