如何通过 API 管理 X 互动服务订单

https://api.twexapi.io/twitter/action

curl --request POST \
  --url https://api.twexapi.io/twitter/action \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --data '{
    "service": "views",
    "link": "https://x.com/username/status/1234567890",
    "quantity": 100
  }'

{
  "code": 200,
  "msg": "success",
  "data": {
    "order_id": 12345
  }
}

GET https://api.twexapi.io/twitter/action/order-status?order_id=<order_id>

curl --request GET \
  --url 'https://api.twexapi.io/twitter/action/order-status?order_id=12345' \
  --header 'Authorization: Bearer <token>'

{
  "code": 200,
  "msg": "success",
  "data": {
    "order_id": 12345,
    "start_count": 0,
    "fee": 1,
    "status": "completed"
  }
}

1import json
2from datetime import datetime, timezone
3from pathlib import Path
4
5import requests
6
7TOKEN = "<your_bearer_token>"
8ORDER_URL = "https://api.twexapi.io/twitter/action"
9STATUS_URL = "https://api.twexapi.io/twitter/action/order-status"
10ORDER_LOG = Path("engagement-service-orders.jsonl")
11
12headers = {
13    "Authorization": f"Bearer {TOKEN}",
14    "Content-Type": "application/json",
15}
16
17ALLOWED_SERVICES = {"likes", "retweets", "views", "bookmarks", "followers"}
18
19def submit_order(service, link, quantity, approval_id, requester, reason):
20    if service not in ALLOWED_SERVICES:
21        raise ValueError(f"Unsupported service: {service}")
22    if not approval_id:
23        raise ValueError("approval_id is required")
24    if quantity <= 0:
25        raise ValueError("quantity must be positive")
26
27    payload = {
28        "service": service,
29        "link": link,
30        "quantity": quantity,
31    }
32
33    response = requests.post(ORDER_URL, headers=headers, json=payload, timeout=30)
34    response.raise_for_status()
35    body = response.json()
36    order_id = (body.get("data") or {}).get("order_id")
37
38    log_row = {
39        "approval_id": approval_id,
40        "requester": requester,
41        "reason": reason,
42        "submitted_at": datetime.now(timezone.utc).isoformat(),
43        "request": payload,
44        "response": body,
45        "order_id": order_id,
46    }
47    with ORDER_LOG.open("a", encoding="utf-8") as f:
48        f.write(json.dumps(log_row, ensure_ascii=False) + "\n")
49
50    return order_id
51
52def check_status(order_id):
53    response = requests.get(
54        STATUS_URL,
55        headers={"Authorization": f"Bearer {TOKEN}"},
56        params={"order_id": order_id},
57        timeout=30,
58    )
59    response.raise_for_status()
60    return response.json()
61
62order_id = submit_order(
63    service="views",
64    link="https://x.com/username/status/1234567890",
65    quantity=100,
66    approval_id="APPROVAL-2025-001",
67    requester="campaign-ops@example.com",
68    reason="Approved campaign operations request",
69)
70
71print(check_status(order_id))

1import time
2
3TERMINAL_STATUSES = {"completed", "failed"}
4
5def wait_for_terminal_status(order_id, poll_seconds=60, max_checks=30):
6    for _ in range(max_checks):
7        body = check_status(order_id)
8        status = (body.get("data") or {}).get("status")
9        print(f"order={order_id} status={status}")
10
11        if status in TERMINAL_STATUSES:
12            return body
13
14        time.sleep(poll_seconds)
15
16    raise TimeoutError(f"Order {order_id} did not reach a terminal status")