如何通过 API 发送 direct-message 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": "@example_user",
    "text": "你好，感谢你反馈导出问题。我们已经发布修复，可以帮你确认最新任务是否正常。",
    "cookie": "auth_token=...; ct0=...; twid=...",
    "pin": "1234"
  }'

1{
2  "code": 200,
3  "msg": "success",
4  "data": {
5    "id": "1234567890123456789",
6    "text": "你好，感谢你反馈导出问题。",
7    "time": "2024-01-01T12:00:00Z",
8    "attachment": null,
9    "attachments": null,
10    "sequence_id": "2076661179334979585",
11    "sender_id": "1234567890123456789",
12    "recipient_id": "9876543210987654321"
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
13def validate_dm_payload(
14    recipient: str,
15    text: str,
16    *,
17    media_urls: list[str] | None = None,
18    video_url: str | None = None,
19) -> None:
20    if not recipient or not recipient.strip():
21        raise ValueError("recipient is required")
22    if not text or not text.strip():
23        raise ValueError("text is required")
24    if media_urls and video_url:
25        raise ValueError("media_urls and video_url are mutually exclusive")
26    if media_urls and len(media_urls) > 1:
27        raise ValueError("media_urls supports at most one image")
28
29def send_xchat_dm(
30    recipient: str,
31    text: str,
32    *,
33    media_urls: list[str] | None = None,
34    video_url: str | None = None,
35) -> dict[str, Any]:
36    validate_dm_payload(
37        recipient,
38        text,
39        media_urls=media_urls,
40        video_url=video_url,
41    )
42
43    payload: dict[str, Any] = {
44        "recipient": recipient.strip(),
45        "text": text.strip(),
46        "cookie": X_COOKIE,
47        "pin": XCHAT_PIN,
48    }
49
50    if media_urls:
51        payload["media_urls"] = media_urls
52    if video_url:
53        payload["video_url"] = video_url
54    if PROXY:
55        payload["proxy"] = PROXY
56
57    response = requests.post(
58        API_URL,
59        headers={
60            "Authorization": f"Bearer {BEARER_TOKEN}",
61            "Content-Type": "application/json",
62            "Accept": "application/json",
63        },
64        json=payload,
65        timeout=60,
66    )
67    response.raise_for_status()
68    data = response.json()
69
70    if data.get("code") != 200:
71        raise RuntimeError(data.get("msg", "TwexAPI returned an error"))
72
73    return data
74
75def normalize_send_result(
76    draft: dict[str, Any],
77    response: dict[str, Any],
78) -> dict[str, Any]:
79    message = response.get("data") or {}
80    return {
81        "draft_id": draft["draft_id"],
82        "recipient": draft["recipient"],
83        "message_id": message.get("id"),
84        "sequence_id": message.get("sequence_id"),
85        "sender_id": message.get("sender_id"),
86        "recipient_id": message.get("recipient_id"),
87        "sent_at": message.get("time") or datetime.now(timezone.utc).isoformat(),
88        "api_code": response.get("code"),
89        "api_msg": response.get("msg"),
90        "raw_response": response,
91    }
92
93if __name__ == "__main__":
94    draft = {
95        "draft_id": "support-followup-1042",
96        "recipient": "@example_user",
97        "text": "你好，感谢你反馈导出问题。我们已经发布修复，可以帮你确认最新任务是否正常。",
98        "status": "approved",
99    }
100
101    if draft["status"] != "approved":
102        raise ValueError("DM draft must be approved before sending")
103
104    result = send_xchat_dm(draft["recipient"], draft["text"])
105    print(normalize_send_result(draft, result))

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

image_result = send_xchat_dm(
    "@example_user",
    "这是刚才提到的截图。",
    media_urls=["https://example.com/screenshot.jpg"],
)

video_result = send_xchat_dm(
    "@example_user",
    "这里是一段简短演示。",
    video_url="https://example.com/walkthrough.mp4",
)