如何通过 API 向 direct-message Group 发送 DM

curl --request POST \
  --url https://api.twexapi.io/v3/twitter/send-dm \
  --header "Authorization: Bearer $TWEXAPI_BEARER_TOKEN" \
  --header "Content-Type: application/json" \
  --data '{
    "recipient": "g1234567890",
    "text": "Launch group 更新：导出修复已经上线。请先重新跑最新报告，再对外分享截图。",
    "cookie": "auth_token=...; ct0=...; twid=...",
    "pin": "1234"
  }'

1{
2  "code": 200,
3  "msg": "success",
4  "data": {
5    "id": "1234567890123456789",
6    "text": "Launch group 更新：导出修复已经上线。",
7    "time": "2024-01-01T12:00:00Z",
8    "attachment": null,
9    "attachments": null,
10    "sequence_id": "2076661179334979585",
11    "sender_id": "1234567890123456789",
12    "recipient_id": "g1234567890"
13  }
14}

1import os
2from datetime import datetime, timezone
3from typing import Any
4
5import requests
6
7API_URL = "https://api.twexapi.io/v3/twitter/send-dm"
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 validate_group_message(text: str) -> str:
27    cleaned = text.strip()
28    if not cleaned:
29        raise ValueError("text is required")
30    return cleaned
31
32def send_group_dm(
33    group_key: str,
34    text: str,
35    *,
36    media_urls: list[str] | None = None,
37    video_url: str | None = None,
38) -> dict[str, Any]:
39    if media_urls and video_url:
40        raise ValueError("media_urls and video_url are mutually exclusive")
41    if media_urls and len(media_urls) > 1:
42        raise ValueError("media_urls supports at most one image")
43
44    payload: dict[str, Any] = {
45        "recipient": resolve_group_id(group_key),
46        "text": validate_group_message(text),
47        "cookie": X_COOKIE,
48        "pin": XCHAT_PIN,
49    }
50
51    if media_urls:
52        payload["media_urls"] = media_urls
53    if video_url:
54        payload["video_url"] = video_url
55    if PROXY:
56        payload["proxy"] = PROXY
57
58    response = requests.post(
59        API_URL,
60        headers={
61            "Authorization": f"Bearer {BEARER_TOKEN}",
62            "Content-Type": "application/json",
63            "Accept": "application/json",
64        },
65        json=payload,
66        timeout=60,
67    )
68    response.raise_for_status()
69    data = response.json()
70
71    if data.get("code") != 200:
72        raise RuntimeError(data.get("msg", "TwexAPI returned an error"))
73
74    return data
75
76def normalize_group_send(
77    draft: dict[str, Any],
78    response: dict[str, Any],
79) -> dict[str, Any]:
80    message = response.get("data") or {}
81    return {
82        "draft_id": draft["draft_id"],
83        "group_key": draft["group_key"],
84        "group_id": resolve_group_id(draft["group_key"]),
85        "message_id": message.get("id"),
86        "sequence_id": message.get("sequence_id"),
87        "sender_id": message.get("sender_id"),
88        "recipient_id": message.get("recipient_id"),
89        "sent_at": message.get("time") or datetime.now(timezone.utc).isoformat(),
90        "api_code": response.get("code"),
91        "api_msg": response.get("msg"),
92        "raw_response": response,
93    }
94
95if __name__ == "__main__":
96    draft = {
97        "draft_id": "launch-group-update-2026-07-16",
98        "group_key": "launch_ops",
99        "text": "Launch group 更新：导出修复已经上线。请先重新跑最新报告，再对外分享截图。",
100        "status": "approved",
101    }
102
103    if draft["status"] != "approved":
104        raise ValueError("Group DM draft must be approved before sending")
105
106    result = send_group_dm(draft["group_key"], draft["text"])
107    print(normalize_group_send(draft, result))

TWEXAPI_BEARER_TOKEN="your_twexapi_bearer_token" \
TWEXAPI_X_COOKIE="auth_token=...; ct0=...; twid=..." \
TWEXAPI_XCHAT_PIN="1234" \
python send_xchat_group_dm.py

image_result = send_group_dm(
    "launch_ops",
    "这是最终 launch checklist 截图。",
    media_urls=["https://example.com/checklist.jpg"],
)

video_result = send_group_dm(
    "creator_review",
    "这是更新后 brief 的简短 walkthrough。",
    video_url="https://example.com/brief-walkthrough.mp4",
)