> ## Documentation Index
> Fetch the complete documentation index at: https://docs.sorsa.io/llms.txt
> Use this file to discover all available pages before exploring further.

# 营销活动验证

# 通过 API 验证 Twitter 互动行为：关注、转推、评论和引用

X（原 Twitter）奖励活动通常要求用户关注账号、转推帖子、发表评论或加入社群。公平发放奖励前，需要确认参与者确实完成了所声称的行为。人工检查难以扩展到大量用户，而依赖自我声明的勾选框容易吸引机器人和欺诈。

Sorsa API 提供专门的验证端点，回答用户是否关注某账号、转推或评论某推文、加入某社群等问题。每项检查通过一次 API 调用返回明确的是/否或状态结果，帮助你基于可验证、可审计的数据构建自动化任务系统、抽奖平台、推荐计划和互动活动。

本指南使用可运行代码介绍所有验证端点，并将它们组合成完整活动验证管道。新账号均有 100 次免费请求，无需信用卡，适用于所有端点，可先验证完整流程再选择套餐。

> **注意：** 更多工作流和端到端示例见博客上的 [Twitter 互动验证 API：完整活动指南](https://api.sorsa.io/blog/twitter-engagement-verification-api)。

***

## 可用的验证检查

以下列出可验证的行为、对应端点及返回结果：

| 行为      | 端点                        | 方法   | 返回结果                                             |
| :------ | :------------------------ | :--- | :----------------------------------------------- |
| 用户关注了账号 | `/check-follow`           | POST | `{"follow": true/false}`                         |
| 用户转推了推文 | `/check-retweet`          | POST | `{"retweet": true/false}`                        |
| 用户引用了推文 | `/check-quoted`           | POST | `{"status": "quoted" / "retweet" / "not_found"}` |
| 用户评论了推文 | `/check-comment`          | GET  | `{"commented": true/false}`                      |
| 用户是社群成员 | `/check-community-member` | POST | `{"is_member": true/false}`                      |

**无法验证的行为：点赞。** X 在 2024 年将点赞设为私有，因此包括官方 API 在内的任何 API 都无法检查特定用户是否给特定推文点了赞。请围绕以上五类行为设计活动。

***

## 检查 1：用户是否关注了账号？

最常见的活动任务，例如“关注 @YourBrand 参与抽奖”。

**端点：**`POST /v3/check-follow`

此端点检查“user\_2 是否关注 user\_1”。将 `user_1` 设为品牌，即被关注账号；将 `user_2` 设为参与者。

### 最简单的示例

```bash theme={null}
curl -X POST https://api.sorsa.io/v3/check-follow \
  -H "ApiKey: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "username_1": "YourBrand",
    "username_2": "participant_handle"
  }'
```

响应：

```json theme={null}
{
  "follow": true,
  "user_protected": false
}
```

### 参数

双方各提供一种标识符：

| 一方                  | 选项（提供一种）                                 |
| :------------------ | :--------------------------------------- |
| 品牌账号（`user_1`，被关注者） | `username_1`、`user_link_1` 或 `user_id_1` |
| 参与者（`user_2`）       | `username_2`、`user_link_2` 或 `user_id_2` |

### Python

```python theme={null}
import requests

API_KEY = "YOUR_API_KEY"
BASE = "https://api.sorsa.io/v3"
HEADERS = {"ApiKey": API_KEY, "Content-Type": "application/json"}

def check_follow(brand_handle: str, participant_handle: str) -> dict:
    resp = requests.post(
        f"{BASE}/check-follow",
        headers=HEADERS,
        json={"username_1": brand_handle, "username_2": participant_handle},
        timeout=15,
    )
    resp.raise_for_status()
    return resp.json()


result = check_follow("YourBrand", "participant123")
if result["follow"]:
    print("Follow verified.")
elif result.get("user_protected"):
    print("Account is private; follow cannot be confirmed.")
else:
    print("Not following.")
```

如果 `user_protected` 为 `true`，说明参与者账号为私有，无法验证其关注关系。

***

## 检查 2：用户是否转推了推文？

例如“转推此帖即可参与”。端点每次检查最多 100 次转推；更多转推需要分页。

**端点：**`POST /v3/check-retweet`

### 参数

| 参数                                   | 类型     | 必需     | 说明                  |
| :----------------------------------- | :----- | :----- | :------------------ |
| `tweet_link`                         | string | 是      | 待验证推文的 URL 或 ID。    |
| `username` / `user_link` / `user_id` | string | 是（三选一） | 参与者，只提供一种标识符。       |
| `next_cursor`                        | string | 否      | 转推超过 100 次时使用的分页游标。 |

### Python

```python theme={null}
def check_retweet(tweet_link: str, participant_handle: str) -> bool:
    cursor = None
    for _ in range(5):  # check up to 500 retweets total
        body = {"tweet_link": tweet_link, "username": participant_handle}
        if cursor:
            body["next_cursor"] = cursor
        resp = requests.post(f"{BASE}/check-retweet", headers=HEADERS, json=body, timeout=15)
        resp.raise_for_status()
        data = resp.json()
        if data["retweet"]:
            return True
        cursor = data.get("next_cursor")
        if not cursor:
            return False
    raise RuntimeError("Retweet verification incomplete: page limit reached")
```

每次调用检查最近 100 次转推。多数活动一次请求即可，因为用户通常在活动开始后不久转推，位于最近一批。热门推文中如果用户转推较早，应通过 `next_cursor` 继续分页。

***

## 检查 3：用户是否引用了推文？

例如“引用此帖并分享想法”。`/check-quoted` 区分引用推文与普通转推，并返回状态字符串。

**端点：**`POST /v3/check-quoted`

### Python

```python theme={null}
def check_quoted(tweet_link: str, participant_handle: str) -> dict:
    resp = requests.post(
        f"{BASE}/check-quoted",
        headers=HEADERS,
        json={"tweet_link": tweet_link, "username": participant_handle},
        timeout=15,
    )
    resp.raise_for_status()
    return resp.json()


data = check_quoted("https://x.com/YourBrand/status/1234567890", "participant123")

if data["status"] == "quoted":
    print(f"Quote verified! They wrote: {data['text']}")
elif data["status"] == "retweet":
    print("They retweeted but did not quote.")
else:
    print("No quote or retweet found.")
```

### 响应

```json theme={null}
{
  "status": "quoted",
  "date": "2026-03-10 14:22:09",
  "text": "This is amazing, everyone should check this out!",
  "user_protected": false
}
```

`status` 返回三种值之一：`"quoted"`（用户发布了引用推文）、`"retweet"`（未添加文字的转推）、`"not_found"`（未发现两种行为）。发现引用时还会返回日期和正文，可用于最低字数、必需话题标签或不当用语等内容质量检查。

```python theme={null}
def quote_is_acceptable(quote_text: str, min_length: int = 30, required_hashtag: str = None) -> bool:
    if len(quote_text.strip()) < min_length:
        return False
    if required_hashtag and required_hashtag.lower() not in quote_text.lower():
        return False
    return True
```

***

## 检查 4：用户是否评论了推文？

例如“在此帖下发表评论”。这是唯一使用 GET 而非 POST 的活动验证端点。

**端点：**`GET /v3/check-comment`

### 参数（查询字符串）

| 参数                                   | 类型     | 必需     | 说明            |
| :----------------------------------- | :----- | :----- | :------------ |
| `tweet_link`                         | string | 是      | 推文 URL 或 ID。  |
| `username` / `user_link` / `user_id` | string | 是（三选一） | 参与者，只提供一种标识符。 |

### Python

```python theme={null}
def check_comment(tweet_link: str, participant_handle: str) -> dict:
    resp = requests.get(
        f"{BASE}/check-comment",
        headers={"ApiKey": API_KEY},
        params={"tweet_link": tweet_link, "username": participant_handle},
        timeout=15,
    )
    resp.raise_for_status()
    return resp.json()


data = check_comment("https://x.com/YourBrand/status/1234567890", "participant123")

if data["commented"]:
    print(f"Comment verified: {data['tweet']['full_text'][:100]}")
else:
    print("No comment found.")
```

`commented` 为 `true` 时，响应包含评论本身的完整 `tweet` 对象，包括正文、互动指标和时间戳。除了确认回复存在，还可检查最低字数、必需话题标签或排除仅有表情符号的回复。

```python theme={null}
def comment_is_acceptable(comment: dict, min_length: int = 20, required_keyword: str = None) -> bool:
    text = comment.get("full_text", "").strip()
    if len(text) < min_length:
        return False
    if required_keyword and required_keyword.lower() not in text.lower():
        return False
    if len(text.split()) < 3:
        return False
    return True
```

***

## 检查 5：用户是否为社群成员？

将此项设为活动要求之前，请向[支持团队](https://docs.sorsa.io/zh-Hans/support)确认当前社群数据可用性，并阅读[列表与社群](https://docs.sorsa.io/zh-Hans/lists-and-communities)中的可用性说明。

例如“加入我们的 X 社群即可参与”，适用于以社群成员身份为前提的活动。

**端点：**`POST /v3/check-community-member`

### Python

```python theme={null}
def check_community_member(community_id: str, participant_handle: str) -> bool:
    resp = requests.post(
        f"{BASE}/check-community-member",
        headers=HEADERS,
        json={"community_id": community_id, "username": participant_handle},
        timeout=15,
    )
    resp.raise_for_status()
    return resp.json().get("is_member", False)


is_member = check_community_member("1966045657589813686", "participant123")
print("Member" if is_member else "Not a member")
```

社群 ID 是社群 URL（`x.com/i/communities/<id>`）中的数字字符串。

***

## 构建活动验证管道

真实活动通常要求完成多项任务。下方模式对单个参与者执行全部五项检查，返回结构化结果，并对评论和引用应用质量规则。

```python theme={null}
from dataclasses import dataclass, field

@dataclass
class CampaignConfig:
    brand_handle: str
    tweet_to_retweet: str
    tweet_to_quote: str
    tweet_to_comment: str
    community_id: str
    required_hashtag: str = ""
    min_quote_length: int = 30
    min_comment_length: int = 20

@dataclass
class ParticipantResult:
    username: str
    follow: bool = False
    retweet: bool = False
    quote: bool = False
    quote_text: str = ""
    comment: bool = False
    comment_text: str = ""
    community: bool = False
    completed: int = field(init=False, default=0)

    def total(self) -> int:
        return sum([self.follow, self.retweet, self.quote, self.comment, self.community])


def verify_participant(username: str, cfg: CampaignConfig) -> ParticipantResult:
    r = ParticipantResult(username=username)

    r.follow = check_follow(cfg.brand_handle, username)["follow"]
    r.retweet = check_retweet(cfg.tweet_to_retweet, username)

    quote_data = check_quoted(cfg.tweet_to_quote, username)
    if quote_data["status"] == "quoted":
        r.quote_text = quote_data.get("text", "")
        r.quote = quote_is_acceptable(r.quote_text, cfg.min_quote_length, cfg.required_hashtag)

    comment_data = check_comment(cfg.tweet_to_comment, username)
    if comment_data.get("commented"):
        r.comment_text = comment_data["tweet"].get("full_text", "")
        r.comment = comment_is_acceptable(comment_data["tweet"], cfg.min_comment_length)

    r.community = check_community_member(cfg.community_id, username)

    r.completed = r.total()
    return r


cfg = CampaignConfig(
    brand_handle="YourBrand",
    tweet_to_retweet="https://x.com/YourBrand/status/111111111",
    tweet_to_quote="https://x.com/YourBrand/status/222222222",
    tweet_to_comment="https://x.com/YourBrand/status/333333333",
    community_id="1966045657589813686",
    required_hashtag="#YourLaunch",
)

result = verify_participant("participant123", cfg)
print(f"@{result.username}: {result.completed}/5 tasks done")
```

五项检查各读第一页时消耗五次请求。转推分页和重试会增加请求，因此五次只是基线，并非固定成本。达到页面预算上限属于验证未完成，不能证明参与者未完成任务。

***

## 批量验证参与者

活动有数千名参与者时，可以批量验证。下方模式遵守速率限制，将结果写入 CSV，并可恢复进度：每处理一位参与者就写入一行，避免崩溃后丢失全部进展。

```python theme={null}
import csv
import time
from pathlib import Path

def verify_campaign_batch(usernames: list[str], cfg: CampaignConfig, output_file: str) -> None:
    fields = ["username", "follow", "retweet", "quote", "comment", "community",
              "completed", "quote_text", "comment_text"]

    already_done = set()
    out_path = Path(output_file)
    if out_path.exists():
        with out_path.open() as f:
            already_done = {row["username"] for row in csv.DictReader(f)}

    mode = "a" if out_path.exists() else "w"
    with out_path.open(mode, newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fields)
        if mode == "w":
            writer.writeheader()

        for i, username in enumerate(usernames):
            if username in already_done:
                continue

            for attempt in range(3):
                try:
                    r = verify_participant(username, cfg)
                    writer.writerow({
                        "username": r.username,
                        "follow": r.follow,
                        "retweet": r.retweet,
                        "quote": r.quote,
                        "comment": r.comment,
                        "community": r.community,
                        "completed": r.completed,
                        "quote_text": r.quote_text,
                        "comment_text": r.comment_text,
                    })
                    f.flush()
                    already_done.add(username)
                    print(f"[{i+1}/{len(usernames)}] @{username}: {r.completed}/5")
                    break
                except RuntimeError as e:
                    print(f"[{i+1}] @{username}: INCOMPLETE {e}")
                    break
                except requests.HTTPError as e:
                    if e.response.status_code == 429:
                        time.sleep(5)
                        continue  # retry the same participant
                    print(f"[{i+1}] @{username}: ERROR {e}")
                    break

            time.sleep(0.25)


participants = open("entries.txt").read().splitlines()
verify_campaign_batch(participants, cfg, "campaign_results.csv")
```

循环在参与者之间等待，并对 `429` 最多重试三次，但单个参与者的转推分页还可能增加调用。生产工作进程池应使用共享限流器。实际吞吐取决于分页深度和响应延迟。未完成或失败的参与者应保留重试资格，不能直接记为未完成任务。

***

## 验证账号所有权

用户参与之前，可能需要确认其确实拥有所提供的 X 用户名。常见方式：

1. 生成唯一代码，例如 `VERIFY-a8f3b2`，并展示给用户。
2. 要求用户发布包含该代码的推文。
3. 使用 `/user-tweets` 获取最近推文，检查代码是否出现。

```python theme={null}
import secrets

def generate_verification_code() -> str:
    return f"VERIFY-{secrets.token_hex(4)}"


def verify_account_ownership(username: str, expected_code: str) -> bool:
    resp = requests.post(
        f"{BASE}/user-tweets",
        headers=HEADERS,
        json={"username": username},
        timeout=15,
    )
    resp.raise_for_status()
    tweets = resp.json().get("tweets", [])

    for tweet in tweets:
        author = tweet.get("user") or {}
        if (author.get("username", "").lower() == username.lstrip("@").lower()
                and not tweet.get("retweeted_status")
                and expected_code in (tweet.get("full_text") or "")):
            return True
    return False


code = generate_verification_code()
print(f"Ask the user to tweet: {code}")
# ... after the user tweets ...
if verify_account_ownership("participant123", code):
    print("Account ownership confirmed.")
```

每个验证挑战应绑定登录参与者和目标 X 账号，设置较短有效期，并且只能使用一次。核对匹配推文的作者和创建时间。示例检查作者和文本；应用仍需实现挑战存储、过期和一次性消费。验证后参与者可删除推文。

***

## 防欺诈考虑

将下列检查作为可配置的资格或人工审核标准。账号年龄和计数不能证明账号是否真实：

* **最低账号年龄。** 通过 `/info` 获取资料并检查 `created_at`。可拒绝最近 30 天创建的账号，因为许多机器人群使用新账号。
* **最低活跃度。** 检查 `tweets_count` 和 `followers_count`。较低数值可触发额外审核，但不能据此断定账号是机器人。
* **评论质量。** `/check-comment` 返回完整正文，可检查最低字数、必需关键词或话题标签，并排除单字符或仅表情回复。
* **引用质量。** `/check-quoted` 返回引用正文，可应用与评论相同的质量规则。
* **完成速度。** 过快完成只是审核信号，不是自动化的证据。记录时间戳并标记异常快速完成的情况。

```python theme={null}
from datetime import datetime, timezone

def is_legitimate_account(
    username: str,
    min_age_days: int = 30,
    min_tweets: int = 10,
    min_followers: int = 5,
) -> tuple[bool, dict]:
    resp = requests.get(
        f"{BASE}/info",
        headers={"ApiKey": API_KEY},
        params={"username": username},
        timeout=15,
    )
    resp.raise_for_status()
    profile = resp.json()

    created = datetime.fromisoformat(profile["created_at"].replace("Z", "+00:00"))
    age_days = (datetime.now(timezone.utc) - created).days

    checks = {
        "account_age_ok": age_days >= min_age_days,
        "has_tweets": profile.get("tweets_count", 0) >= min_tweets,
        "has_followers": profile.get("followers_count", 0) >= min_followers,
        "not_protected": not profile.get("protected", False),
    }
    return all(checks.values()), checks
```

在五项验证前运行此检查。如果 `is_legitimate_account` 返回 `False`，可对本来就会被拒绝的参与者省去五次验证请求。

***

## 按影响力为参与者评分

不同参与者的潜在传播范围不同。对活动而言，50,000 位粉丝账号的转推价值可能高于只有 50 位粉丝的账号。通过 `/info` 获取参与者资料，并按粉丝数为奖励加权。

```python theme={null}
import math

BASE_POINTS = {"follow": 10, "retweet": 15, "quote": 25, "comment": 20, "community": 10}

def get_follower_count(username: str) -> int:
    resp = requests.get(
        f"{BASE}/info",
        headers={"ApiKey": API_KEY},
        params={"username": username},
        timeout=15,
    )
    resp.raise_for_status()
    return resp.json().get("followers_count", 0)


def calculate_weighted_points(result: ParticipantResult) -> dict:
    followers = get_follower_count(result.username)
    # log scaling: 100 followers -> 2x, 10K -> 4x, 1M -> 6x
    multiplier = max(1.0, math.log10(followers + 1))
    total = 0
    breakdown = {}
    for task, base in BASE_POINTS.items():
        if getattr(result, task):
            points = round(base * multiplier)
            breakdown[task] = points
            total += points
    return {"followers": followers, "multiplier": round(multiplier, 2),
            "breakdown": breakdown, "total": total}
```

加密货币活动可将粉丝数倍数替换为 [Sorsa Score](https://docs.sorsa.io/zh-Hans/sorsa-score-and-crypto-analytics)，衡量账号在加密货币 KOL、项目和 VC 中的认可度。

***

## 关于点赞

X（Twitter）在 2024 年将点赞设为私有。平台不再通过任何公开 API 暴露具体哪些用户点赞了特定推文，包括 Sorsa、官方 X API 和其他第三方工具。如果活动之前要求“点赞此推文”，请改为仍可完整验证的转推或评论。

***

## 后续步骤

* [搜索推文](https://docs.sorsa.io/zh-Hans/search-tweets)：按关键词查找活动相关推文，扩大监测范围
* [追踪提及](https://docs.sorsa.io/zh-Hans/search-mentions)：同时追踪自然品牌提及与活动带来的提及
* [实时监测](https://docs.sorsa.io/zh-Hans/real-time-monitoring)：轮询新行为，实现近实时验证
* [粉丝与关注](https://docs.sorsa.io/zh-Hans/followers-and-following)：提取自身粉丝列表，与活动参与者交叉核对
* [价格](https://api.sorsa.io/pricing)：估算活动成本，完整验证的基线为每人五次请求
* [API 参考](https://docs.sorsa.io/zh-Hans/api-reference-guide)：全部验证端点的完整规范