如何用 TwexAPI 构建 X 互动工作流

互动自动化如果被当成“增长开关”，很快会失控。真正能上线的流程，通常不是谁都可以随手给任意链接加量，而是先确认内容已批准、预算已锁定、数量符合限制，再把每一次提交都记录成可追踪订单。

TwexAPI 的 POST /twitter/action 适合放在活动队列后面：提交 service、link 和 quantity 后，保存返回的 order_id，再用 GET /twitter/action/order-status 查询交付状态。这样互动动作不会漂在脚本日志里，而是进入预算、审批、重试和复盘系统。

工作流边界

互动工作流边界 指只对自有、授权或已批准推广的内容运行订单化互动流程。它不是替代内容判断的按钮，也不应该绕过活动审批、平台规则或内部预算限制。

上线前先固定这几条规则：

自有或已授权内容：只对允许推广的账号和内容运行互动流程。
公开目标：目标推文或账号应保持公开，避免订单无法交付。
预算上限：按活动、服务和目标 URL 设置上限，不允许脚本无限提交。
审计日志：保存目标 URL、服务类型、数量、操作者、活动、响应和 order_id。
停止条件：当内容下架、账号状态变化、订单失败或市场环境变化时，暂停后续提交。

接口链路

接口链路 指先提交互动订单，再查询订单状态。提交端点返回 order_id，状态端点返回进度、费用、起始计数和当前状态。

步骤	端点	用途
1	`POST /twitter/action`	提交 `likes`、`retweets`、`views`、`bookmarks` 或 `followers` 订单
2	`GET /twitter/action/order-status?order_id=...`	查询 `pending`、`in_progress`、`completed`、`partial`、`failed` 等状态

请求使用 Bearer Token：

Authorization: Bearer <your_token>
Content-Type: application/json

服务和数量需要提前校验：

服务	数量范围	目标
`likes`	`10` 到 `5000`	推文 URL
`retweets`	`10` 到 `500`	推文 URL
`views`	`100` 到 `9999999`	推文 URL
`bookmarks`	`10` 到 `5000`	推文 URL
`followers`	`10` 到 `30000`	账号或 Profile URL

Python 示例：提交和追踪订单

Python 示例 指一个服务端活动队列：先校验活动计划，再提交订单，保存 order_id，最后轮询订单状态。生产环境应把提交动作放在审批后的队列中，而不是让运营页面直接无限调用 API。

1import requests
2import time
3from datetime import datetime, timezone
4
5API_BASE = "https://api.twexapi.io"
6TOKEN = "YOUR_TWEXAPI_BEARER_TOKEN"
7
8SERVICE_LIMITS = {
9    "likes": (10, 5000),
10    "retweets": (10, 500),
11    "views": (100, 9999999),
12    "bookmarks": (10, 5000),
13    "followers": (10, 30000),
14}
15
16def headers():
17    return {
18        "Authorization": f"Bearer {TOKEN}",
19        "Accept": "application/json",
20        "Content-Type": "application/json"
21    }
22
23def validate_order(order):
24    service = order["service"]
25    quantity = int(order["quantity"])
26
27    if service not in SERVICE_LIMITS:
28        raise ValueError(f"Unsupported service: {service}")
29
30    min_qty, max_qty = SERVICE_LIMITS[service]
31    if quantity < min_qty or quantity > max_qty:
32        raise ValueError(f"{service} quantity must be between {min_qty} and {max_qty}")
33
34    if not order["link"].startswith(("https://x.com/", "https://twitter.com/")):
35        raise ValueError("link must be a full X/Twitter URL")
36
37def submit_order(order, campaign_id, requested_by):
38    validate_order(order)
39
40    response = requests.post(
41        API_BASE + "/twitter/action",
42        headers=headers(),
43        json={
44            "service": order["service"],
45            "link": order["link"],
46            "quantity": int(order["quantity"]),
47        },
48        timeout=30,
49    )
50    response.raise_for_status()
51    payload = response.json()
52
53    if payload.get("code", 200) >= 400:
54        raise RuntimeError(payload.get("msg", "TwexAPI returned an error"))
55
56    return {
57        "campaign_id": campaign_id,
58        "requested_by": requested_by,
59        "service": order["service"],
60        "link": order["link"],
61        "quantity": int(order["quantity"]),
62        "order_id": payload["data"]["order_id"],
63        "submitted_at": datetime.now(timezone.utc).isoformat(),
64        "raw_response": payload,
65    }
66
67def get_order_status(order_id):
68    response = requests.get(
69        API_BASE + "/twitter/action/order-status",
70        headers=headers(),
71        params={"order_id": order_id},
72        timeout=30,
73    )
74    response.raise_for_status()
75    payload = response.json()
76
77    if payload.get("code", 200) >= 400:
78        raise RuntimeError(payload.get("msg", "TwexAPI returned an error"))
79
80    return payload["data"]
81
82campaign_orders = [
83    {"service": "views", "link": "https://x.com/yourbrand/status/123456789", "quantity": 5000},
84    {"service": "likes", "link": "https://x.com/yourbrand/status/123456789", "quantity": 100},
85]
86
87submitted = []
88for order in campaign_orders:
89    submitted_order = submit_order(order, campaign_id="launch-2026-04", requested_by="growth_ops")
90    submitted.append(submitted_order)
91    print("submitted", submitted_order["order_id"], submitted_order["service"])
92    time.sleep(2)
93
94for item in submitted:
95    status = get_order_status(item["order_id"])
96    print(item["order_id"], status.get("status"), status)

状态值进入你的任务系统后，可以这样处理：pending 和 in_progress 继续观察，completed 进入复盘，partial 需要人工判断是否补单，failed 则保留错误证据并停止同一目标的后续订单。

和搜索信号配合

搜索信号配合 指在互动订单之前，先用搜索、回复和趋势数据判断内容是否值得进入分发队列。互动动作应该服务于明确活动目标，而不是把每个趋势词都变成投放对象。

发现：用 Advanced Search 找到细分领域里的活跃讨论。
创作：发布真正有用、贴合主题的帖子或回复。
审批：确认目标 URL、服务、数量、预算、负责人和停止条件。
提交：调用 POST /twitter/action，保存 order_id 和原始响应。
追踪：用订单状态端点同步进度，复盘完成、部分完成和失败订单。

这样自动化会服务于编辑判断，而不是把每个趋势都变成推广目标。

预算与审计字段

预算与审计字段 指每个订单必须能解释“为什么提交、谁批准、花了多少、现在到哪一步”。开始活动前，先确定要保存和复核的字段：

控制字段	为什么重要
活动 ID	把每个动作关联到发布或实验
目标 URL	避免把动作提交到错误帖子
服务和数量	方便复核预算和节奏
操作者或任务 ID	说明是谁或哪个任务发起动作
`order_id`	后续查询状态、对账和复盘的主键
状态历史	区分 `pending`、`in_progress`、`completed`、`partial`、`failed`
原始响应	为重试、客服排查和审计保留证据

错误处理和停止条件

错误处理和停止条件 指把接口错误、订单失败和业务风险分开处理。不要因为一个订单失败就无限重试，也不要因为网络超时就误判订单没有创建。

提交前失败：校验服务、数量和 URL；不符合范围时不要调用 API。
提交后超时：先查本地是否已有 order_id；没有证据时再决定是否重试，避免重复订单。
订单失败：保存失败状态和原始响应，暂停同一目标的后续订单。
部分完成：不要自动补单，先看活动目标和预算是否仍然有效。
内容状态变化：目标内容删除、设为私密或活动取消时，停止后续提交。

小结

小结指把 X 互动自动化当成活动订单系统，而不是一次性脚本：选择已批准内容，设置服务和数量范围，调用 /twitter/action 提交订单，保存 order_id，再用 /twitter/action/order-status 追踪交付状态。TwexAPI 提供 API 层，真正让流程稳定的是外围的审批、预算、日志和停止机制。