Developer Examples

Python Automation Scenarios

Ready-to-run Python scripts covering common Twitter/X operations. Each scenario is implemented three ways — choose the approach that fits your setup.

Single-account assumption: These examples assume only one TweetClaw extension is connected to TweetPilot, or only one OAuth account is managed.
Ask TweetPilot AI how to use examples with multiple accounts →
8 scenarios  ·  3 implementations each Download All Examples (.zip)
S01
Scheduled Tweet
Post a tweet on a fixed schedule — ideal for daily updates, market recaps, or broadcast-style content.
"""
S01 · Scheduled Tweet — ClawBot SDK
=====================================
EN: Post via TweetClaw browser extension.
    Add to TweetPilot Task Manager for automatic scheduling.
中文：通过 TweetClaw 浏览器扩展发推。加入任务管理器实现定时自动发送。

Manual run / 手动运行:
  PYTHONPATH="$HOME/.tweetpilot/clawbot" python3 clawbot.py
"""

from datetime import datetime
from clawbot import ClawBotClient

# ── Config — edit TWEET_TEXT; {date} inserts today's date (YYYY-MM-DD)
# 配置 — 修改 TWEET_TEXT；{date} 将被替换为当天日期
TWEET_TEXT = "Daily update {date} — powered by TweetPilot"


def main():
    client = ClawBotClient()

    # Auto-select the first connected TweetClaw instance.
    # For multi-account use, ask TweetPilot AI for guidance.
    # 自动选择第一个已连接的 TweetClaw 实例；多账号请询问 TweetPilot AI。
    instances = client.x.status.get_instances()
    instance_id = None
    if isinstance(instances, list) and instances:
        first = instances[0]
        instance_id = first.get("instanceId") or first.get("id")

    text = TWEET_TEXT.format(date=datetime.now().strftime("%Y-%m-%d"))
    result = client.x.actions.create_tweet(text, instance_id=instance_id)

    print(f"Tweet sent / 发推成功: {text}")
    print(f"Result: {result}")


if __name__ == "__main__":
    main()
"""
S01 · Scheduled Tweet — xdk SDK
==================================
EN: Post via Twitter OAuth using xdk.
    TweetPilot provides the token automatically — no manual setup.
中文：通过 Twitter OAuth + xdk 客户端发推，TweetPilot 自动提供 token。

pip install xdk requests
"""

import sys
import requests
from datetime import datetime
from xdk import Client

# ── Config — replace with your Twitter numeric ID
# 配置 — 替换为你的 Twitter 数字 ID（TweetPilot → 账号设置 → 账号 ID）
TWITTER_ID = "YOUR_TWITTER_ID"
TWEET_TEXT = "Daily update {date} — powered by TweetPilot"


def get_access_token(twitter_id: str) -> str:
    # Fetch OAuth token from TweetPilot rust-bridge (port 20088)
    # 从 TweetPilot rust-bridge（端口 20088）获取 OAuth token
    try:
        resp = requests.post(
            "http://127.0.0.1:20088/api/v1/x/oauth/access-token",
            json={"twitter_id": twitter_id}, timeout=10,
        )
        resp.raise_for_status()
        return resp.json()["access_token"]
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running / TweetPilot 未在运行")
        sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code} — {e.response.text}")
        sys.exit(1)


def main():
    # Guard: TWITTER_ID must be replaced before running (OAuth accounts only)
    # 运行前检查：TWITTER_ID 必须替换为真实 ID（仅支持 OAuth 授权账号）
    if TWITTER_ID == "YOUR_TWITTER_ID":
        print("ERROR: Set TWITTER_ID to your numeric Twitter ID (OAuth accounts only).")
        print("错误：请将 TWITTER_ID 替换为数字 ID，仅限 TweetPilot 中 x 开头的 OAuth 账号。")
        sys.exit(1)

    # ⚠️ Must use access_token= (User OAuth 2.0), NOT bearer_token= (read-only App token)
    # ⚠️ 必须用 access_token= 传入，bearer_token= 是只读的 App token，写操作会报错
    client = Client(access_token=get_access_token(TWITTER_ID))
    text   = TWEET_TEXT.format(date=datetime.now().strftime("%Y-%m-%d"))
    result = client.posts.create(body={"text": text})

    print(f"Tweet sent / 发推成功: {text}")
    print(f"Tweet ID: {result.data.id}")


if __name__ == "__main__":
    main()
This tab shows how to call TweetPilot LocalBridge via raw HTTP — the browser extension acts as proxy, no OAuth token needed. To call Twitter's official API directly, refer to Twitter's official docs.
本 Tab 演示通过裸 HTTP 调用 TweetPilot LocalBridge（浏览器扩展代理），无需 OAuth token。如需直接调用推特官方 API，请参考推特官方文档，网上资料也很丰富，本页不再重复说明。 
"""
S01 · Scheduled Tweet — HTTP REST API (LocalBridge)
=====================================================
EN: Call TweetPilot's LocalBridge REST API directly — no OAuth token needed.
    The browser extension handles authentication transparently.
    Requires TweetClaw extension online in the browser.
中文：直接调用 TweetPilot LocalBridge REST API 发推，无需 OAuth token。
     浏览器扩展自动处理认证，需 TweetClaw 扩展在线。

pip install requests
"""

import sys
import requests
from datetime import datetime

LOCAL_BRIDGE = "http://127.0.0.1:20088"
TWEET_TEXT   = "Daily update {date} — powered by TweetPilot"


def post_tweet(text: str) -> str:
    # POST /api/v1/x/tweets — LocalBridge proxies to Twitter via browser extension
    # LocalBridge 通过浏览器扩展代理发推，无需手动传 token
    try:
        resp = requests.post(
            f"{LOCAL_BRIDGE}/api/v1/x/tweets",
            json={"text": text},
            timeout=15,
        )
        resp.raise_for_status()
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running / TweetPilot 未在运行")
        sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code} — {e.response.text}")
        sys.exit(1)

    data = resp.json()

    # LocalBridge returns HTTP 200 even when Twitter rejects the tweet.
    # Must check the response body for GraphQL-level errors (e.g. code 187 = duplicate).
    # LocalBridge 即使推特拒绝也返回 HTTP 200，必须检查响应体中的 GraphQL 错误。
    errors = data.get("data", {}).get("errors")
    if errors:
        for err in errors:
            print(f"ERROR: Twitter rejected — code {err.get('code')}: {err.get('message')}")
        sys.exit(1)

    try:
        return data["data"]["data"]["create_tweet"]["tweet_results"]["result"]["rest_id"]
    except (KeyError, TypeError):
        return "unknown"


def main():
    text     = TWEET_TEXT.format(date=datetime.now().strftime("%Y-%m-%d"))
    tweet_id = post_tweet(text)
    print(f"Tweet sent / 发推成功: {text}")
    print(f"Tweet ID: {tweet_id}")


if __name__ == "__main__":
    main()
Equivalent curl command
# Post a tweet via TweetPilot LocalBridge — no token needed
# 通过 TweetPilot LocalBridge 发推，无需 token
curl -s -X POST http://127.0.0.1:20088/api/v1/x/tweets \
  -H "Content-Type: application/json" \
  -d '{"text": "Daily update — powered by TweetPilot"}'
↓ clawbot.py xdk.py http_rest.py
S02
Tweet with Image
Upload a local image and attach it to a tweet — great for product screenshots, charts, or daily visual content.
"""
S02 · Tweet with Image — ClawBot SDK
======================================
EN: Upload a local image and post it as a tweet via browser extension.
    ClawBot handles chunked upload + tweet creation in one call.
    Requires TweetClaw extension online in the browser.
中文：通过浏览器扩展上传本地图片并发推。
     ClawBot 封装了分块上传和发推的完整流程，一行调用即可完成。

pip install requests  (clawbot pre-installed via TweetPilot)
"""

import sys, os
from clawbot import ClawBotClient

IMAGE_PATH = "image.jpg"  # ← Replace / 替换为你的图片路径
TWEET_TEXT = "🚀 TweetPilot v1 is live — the Social AI Agent built for X.\nAutomate posts, replies, likes & research while you stay focused.\nBuilt for creators, builders & teams → tweetpilot.ai"


def main():
    if not os.path.exists(IMAGE_PATH):
        print(f"ERROR: Image not found: {IMAGE_PATH}")
        sys.exit(1)

    client = ClawBotClient()

    # Resolve instance_id for multi-extension environments
    # 多扩展实例环境下，获取第一个可用的 instanceId
    instances   = client.x.status.get_instances()
    instance_id = None
    if isinstance(instances, list) and instances:
        instance_id = instances[0].get("instanceId") or instances[0].get("id")

    # post_tweet() uploads the image then creates the tweet in one call
    # post_tweet() 内部自动完成上传 + 发推，无需手动处理 media_id
    result = client.media.post_tweet(
        text=TWEET_TEXT,
        file_paths=[IMAGE_PATH],
        instance_id=instance_id,
    )
    print(f"Tweet sent / 发推成功: {TWEET_TEXT}")
    print(f"Result: {result}")


if __name__ == "__main__":
    main()
"""
S02 · Tweet with Image — xdk SDK
===================================
EN: Upload image via Twitter media API and attach to a tweet.
    TweetPilot provides the OAuth access token automatically.
    Requires OAuth account authorized in TweetPilot (x-prefix accounts).
    Note: media.upload() requires the 'media.write' OAuth scope.
中文：通过 Twitter media API 上传图片并发推。
     TweetPilot 自动提供 OAuth token，需 x 开头的 OAuth 授权账号。
     注：media.upload() 需要 media.write OAuth scope。

pip install xdk requests
"""

import sys, os, base64
import requests
from xdk import Client

TWITTER_ID = "YOUR_TWITTER_ID"  # ← OAuth account numeric ID / OAuth 账号数字 ID
IMAGE_PATH = "image.jpg"        # ← Replace / 替换为你的图片路径
TWEET_TEXT = "🚀 TweetPilot v1 is live — the Social AI Agent built for X.\nAutomate posts, replies, likes & research while you stay focused.\nBuilt for creators, builders & teams → tweetpilot.ai"


def get_access_token(twitter_id):
    try:
        resp = requests.post(
            "http://127.0.0.1:20088/api/v1/x/oauth/access-token",
            json={"twitter_id": twitter_id}, timeout=10,
        )
        resp.raise_for_status()
        return resp.json()["access_token"]
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running / TweetPilot 未在运行")
        sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code} — {e.response.text}")
        sys.exit(1)


def main():
    if TWITTER_ID == "YOUR_TWITTER_ID":
        print("ERROR: Set TWITTER_ID to your numeric Twitter ID (OAuth accounts only).")
        sys.exit(1)
    if not os.path.exists(IMAGE_PATH):
        print(f"ERROR: Image not found: {IMAGE_PATH}")
        sys.exit(1)

    # Must use access_token= (User OAuth 2.0), NOT bearer_token= (read-only)
    # 必须用 access_token= 传入，bearer_token= 是只读的 App token
    client = Client(access_token=get_access_token(TWITTER_ID))

    # Read image and base64-encode for Twitter media API (requires media.write scope)
    # 读取图片并 base64 编码（需要 media.write scope）
    with open(IMAGE_PATH, "rb") as f:
        media_b64 = base64.b64encode(f.read()).decode()

    upload_result = client.media.upload(body={
        "media": media_b64,
        "media_category": "tweet_image",
    })
    media_id = (upload_result.data["id"] if isinstance(upload_result.data, dict)
                else upload_result.data.id)

    result = client.posts.create(body={
        "text": TWEET_TEXT,
        "media": {"media_ids": [media_id]},
    })
    tweet_id = result.data["id"] if isinstance(result.data, dict) else result.data.id
    print(f"Tweet sent / 发推成功: {TWEET_TEXT[:50]}...")
    print(f"Tweet ID: {tweet_id}")


if __name__ == "__main__":
    main()
Calls LocalBridge task API for image upload (browser extension proxy), then posts tweet. To use Twitter's official media API directly, see Twitter docs.
通过 LocalBridge task API 上传图片（浏览器扩展代理），再发推。如需直接调用推特官方媒体 API，请参考推特官方文档。 
"""
S02 · Tweet with Image — HTTP REST API (LocalBridge)
pip install requests
"""

import sys, os, time, mimetypes
import requests

LOCAL_BRIDGE = "http://127.0.0.1:20088"
IMAGE_PATH   = "image.jpg"  # ← Replace / 替换为你的图片路径
TWEET_TEXT   = "Check this out — powered by TweetPilot"


def get_instance_id():
    resp = requests.get(f"{LOCAL_BRIDGE}/api/v1/x/instances", timeout=10)
    resp.raise_for_status()
    instances = resp.json()
    if not instances:
        print("ERROR: No TweetClaw instance connected"); sys.exit(1)
    return instances[0]["instanceId"]


def upload_image(path, instance_id):
    # Step 1: create task / 创建任务
    task_id = requests.post(f"{LOCAL_BRIDGE}/api/v1/tasks", json={
        "clientName": "tweetClaw", "instanceId": instance_id,
        "taskKind": "x.media_upload", "inputMode": "chunked_binary", "params": {},
    }, timeout=10).json()["taskId"]

    # Step 2: upload file (single chunk for images < 5 MB) / 上传文件
    with open(path, "rb") as f: data = f.read()
    requests.put(f"{LOCAL_BRIDGE}/api/v1/tasks/{task_id}/input/0",
        data=data, headers={"Content-Type": "application/octet-stream"}, timeout=30).raise_for_status()

    # Step 3: seal → start → poll → result
    ct = mimetypes.guess_type(path)[0] or "image/jpeg"
    requests.post(f"{LOCAL_BRIDGE}/api/v1/tasks/{task_id}/seal",
        json={"totalParts": 1, "totalBytes": len(data), "contentType": ct}, timeout=10).raise_for_status()
    requests.post(f"{LOCAL_BRIDGE}/api/v1/tasks/{task_id}/start", timeout=10).raise_for_status()

    deadline = time.time() + 60
    while time.time() < deadline:
        st = requests.get(f"{LOCAL_BRIDGE}/api/v1/tasks/{task_id}", timeout=10).json()
        if st["state"] == "completed": break
        if st["state"] in ("failed","cancelled"):
            print(f"ERROR: {st.get('errorMessage')}"); sys.exit(1)
        time.sleep(2)

    result = requests.get(f"{LOCAL_BRIDGE}/api/v1/tasks/{task_id}/result", timeout=10).json()
    return result["mediaId"]


def main():
    if not os.path.exists(IMAGE_PATH):
        print(f"ERROR: Image not found: {IMAGE_PATH}"); sys.exit(1)
    try:
        instance_id = get_instance_id()
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running"); sys.exit(1)

    print(f"Uploading / 上传中: {IMAGE_PATH}")
    media_id = upload_image(IMAGE_PATH, instance_id)
    print(f"Upload done / 上传完成, media_id: {media_id}")

    resp = requests.post(f"{LOCAL_BRIDGE}/api/v1/x/tweets",
        json={"text": TWEET_TEXT, "media_ids": [media_id]}, timeout=15)
    resp.raise_for_status()
    data = resp.json()
    errors = data.get("data", {}).get("errors")
    if errors:
        for e in errors: print(f"ERROR: code {e.get('code')}: {e.get('message')}")
        sys.exit(1)
    try:
        tweet_id = data["data"]["data"]["create_tweet"]["tweet_results"]["result"]["rest_id"]
    except (KeyError, TypeError):
        tweet_id = "unknown"
    print(f"Tweet sent / 发推成功: {TWEET_TEXT}")
    print(f"Tweet ID: {tweet_id}")


if __name__ == "__main__":
    main()
Equivalent curl command
# Step 1: Get instance ID / 获取实例 ID
INSTANCE=$(curl -s http://127.0.0.1:20088/api/v1/x/instances \
  | python3 -c "import sys,json; print(json.load(sys.stdin)[0]['instanceId'])")

# Step 2: Create upload task / 创建上传任务
TASK_ID=$(curl -s -X POST http://127.0.0.1:20088/api/v1/tasks \
  -H "Content-Type: application/json" \
  -d "{\"clientName\":\"tweetClaw\",\"instanceId\":\"$INSTANCE\",\"taskKind\":\"x.media_upload\",\"inputMode\":\"chunked_binary\",\"params\":{}}" \
  | python3 -c "import sys,json; print(json.load(sys.stdin)['taskId'])")

# Step 3: Upload image / 上传图片
curl -s -X PUT http://127.0.0.1:20088/api/v1/tasks/$TASK_ID/input/0 \
  -H "Content-Type: application/octet-stream" \
  --data-binary @image.jpg

# Step 4: Seal + Start / 封装并启动任务
SIZE=$(wc -c < image.jpg | tr -d ' ')
curl -s -X POST http://127.0.0.1:20088/api/v1/tasks/$TASK_ID/seal \
  -H "Content-Type: application/json" \
  -d "{\"totalParts\":1,\"totalBytes\":$SIZE,\"contentType\":\"image/jpeg\"}"
curl -s -X POST http://127.0.0.1:20088/api/v1/tasks/$TASK_ID/start

# Step 5: Get media_id from result / 从结果中提取 media_id
MEDIA_ID=$(curl -s http://127.0.0.1:20088/api/v1/tasks/$TASK_ID/result \
  | python3 -c "import sys,json; print(json.load(sys.stdin)['mediaId'])")

# Step 6: Post tweet with image / 带图发推
curl -s -X POST http://127.0.0.1:20088/api/v1/x/tweets \
  -H "Content-Type: application/json" \
  -d "{\"text\":\"Check this out — powered by TweetPilot\",\"media_ids\":[\"$MEDIA_ID\"]}"
↓ clawbot.py xdk.py http_rest.py
S03
Search & Auto-Reply
Search tweets by keyword and automatically reply to matching results — useful for brand monitoring and audience engagement.
Rate-limit warning: Only the first matching tweet is replied to per run. Replying to many tweets at once risks spam detection or account suspension. Use TweetPilot's scheduler to run this every 30+ minutes.
⚠️ 每次只回复第一条匹配推文。批量回复极易触发推特限流或封号。建议配合 TweetPilot 定时任务，每 30 分钟以上执行一次。 
"""
S03 · Search & Auto-Reply — ClawBot SDK
==========================================
EN: Search tweets by keyword, reply ONLY to the first result.
    ⚠️ Replying to many tweets at once risks rate-limiting or suspension.
    Requires TweetClaw extension online in the browser.
中文：按关键词搜索推文，只回复第一条。
     ⚠️ 批量回复极易触发限流，建议每 30 分钟以上定时执行。

pip install requests  (clawbot pre-installed via TweetPilot)
"""

import sys
from clawbot import ClawBotClient

SEARCH_QUERY = "TweetPilot"  # ← Keyword / 搜索关键词
REPLY_TEXT   = "Thanks for mentioning us! 🙌 — TweetPilot"


def main():
    client = ClawBotClient()
    instances   = client.x.status.get_instances()
    instance_id = None
    if isinstance(instances, list) and instances:
        instance_id = instances[0].get("instanceId") or instances[0].get("id")

    print(f'Searching: "{SEARCH_QUERY}"')
    tweets = client.x.search.search_tweets(SEARCH_QUERY, count=5, instance_id=instance_id)

    if not tweets:
        print("No tweets found / 未找到相关推文"); return

    # ⚠️ Reply ONLY to the first tweet — never loop over all results
    # ⚠️ 只回复第一条，不循环全部结果
    tweet = tweets[0]
    print(f"Replying to {tweet.id}: {(tweet.text or '')[:60]!r}")
    try:
        result = client.x.actions.reply(tweet_id=tweet.id, text=REPLY_TEXT, instance_id=instance_id)
        print(f"Done / 完成: {result}")
    except Exception as e:
        print(f"Failed / 失败: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()
Rate-limit warning: Only the first matching tweet is replied to per run. Schedule this every 30+ minutes.
⚠️ 每次只回复第一条匹配推文，建议配合定时任务每 30 分钟以上执行一次。 
"""
S03 · Search & Auto-Reply — xdk SDK
=======================================
EN: Search recent tweets via Twitter API v2, reply to the first replyable result.
    Tries each result in order; skips retweets and reply-restricted tweets.
    Requires OAuth account authorized in TweetPilot Account Settings.
中文：通过 Twitter API v2 搜索推文，回复第一条可回复的推文。
     按顺序尝试，自动跳过转发和有回复限制的推文。

pip install xdk requests
"""

import sys
import requests
from xdk import Client

TWITTER_ID   = "YOUR_TWITTER_ID"  # ← OAuth account numeric ID
SEARCH_QUERY = "TweetPilot"
REPLY_TEXT   = "Thanks for mentioning us! 🙌 — TweetPilot"


def get_access_token(twitter_id):
    try:
        resp = requests.post("http://127.0.0.1:20088/api/v1/x/oauth/access-token",
            json={"twitter_id": twitter_id}, timeout=10)
        resp.raise_for_status()
        return resp.json()["access_token"]
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running"); sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code} — {e.response.text}"); sys.exit(1)


def main():
    if TWITTER_ID == "YOUR_TWITTER_ID":
        print("ERROR: Set TWITTER_ID to your numeric Twitter ID (OAuth only).")
        sys.exit(1)

    # Must use access_token= (User OAuth 2.0), NOT bearer_token= (read-only)
    client = Client(access_token=get_access_token(TWITTER_ID))

    print(f'Searching: "{SEARCH_QUERY}"')
    first_page = next(iter(client.posts.search_recent(query=SEARCH_QUERY, max_results=10)), None)
    tweets = (first_page.data or []) if first_page else []

    if not tweets:
        print("No tweets found / 未找到相关推文"); return

    # Try each tweet; skip RTs; on 403 (reply restricted) move to next.
    # 逐条尝试；跳过转发；遇到 403（回复被限制）换下一条。
    replied = False
    for t in tweets:
        t_text   = (t.get("text") if isinstance(t, dict) else getattr(t, "text", "") or "")
        tweet_id = (t.get("id")   if isinstance(t, dict) else t.id)
        if t_text.startswith("RT @"):
            continue
        print(f"Trying {tweet_id}: {t_text[:60]!r}")
        try:
            result = client.posts.create(body={
                "text":  REPLY_TEXT,
                "reply": {"in_reply_to_tweet_id": tweet_id},
            })
            reply_id = (result.data["id"] if isinstance(result.data, dict) else result.data.id)
            print(f"Done / 完成, reply ID: {reply_id}")
            replied = True
            break
        except Exception as e:
            err = getattr(getattr(e, "response", None), "text", str(e))
            print(f"  Skipped: {err}")
            continue

    if not replied:
        print("Failed / 失败: no replyable tweet found / 搜索结果中无可回复推文")
        sys.exit(1)


if __name__ == "__main__":
    main()
print(f"Failed / 失败: {e}") sys.exit(1) if __name__ == "__main__": main() main()
Calls LocalBridge REST API — browser extension acts as proxy, no OAuth token needed. To call Twitter's official API directly, see Twitter docs.
Rate-limit warning: Only the first matching tweet is replied to per run. Schedule every 30+ minutes. / ⚠️ 每次只回复第一条，建议每 30 分钟以上定时执行一次。 
"""
S03 · Search & Auto-Reply — HTTP REST API (LocalBridge)
EN: Search tweets via LocalBridge, reply ONLY to the first result.
    ⚠️ Replying to many tweets at once risks rate-limiting or suspension.
pip install requests
"""

import sys
import requests

LOCAL_BRIDGE = "http://127.0.0.1:20088"
SEARCH_QUERY = "TweetPilot"  # ← Keyword / 搜索关键词
REPLY_TEXT   = "Thanks for mentioning us! 🙌 — TweetPilot"


def search_tweets(query, count):
    try:
        resp = requests.get(f"{LOCAL_BRIDGE}/api/v1/x/search",
            params={"query": query, "count": count}, timeout=15)
        resp.raise_for_status()
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running"); sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code}"); sys.exit(1)

    tweets = []
    try:
        instructions = (resp.json().get("data", {})
            .get("data", {})
            .get("search_by_raw_query", {}).get("search_timeline", {})
            .get("timeline", {}).get("instructions", []))
        for instr in instructions:
            for entry in instr.get("entries", []):
                r = entry.get("content", {}).get("itemContent", {}).get("tweet_results", {}).get("result", {})
                if not r: continue
                t = r.get("tweet") or r
                rid = t.get("rest_id") or r.get("rest_id")
                if rid:
                    tweets.append({"id": rid, "text": t.get("legacy", {}).get("full_text", "")})
    except Exception as e:
        print(f"WARNING: parse error: {e}")
    return tweets


def reply_to_tweet(tweet_id, text):
    resp = requests.post(f"{LOCAL_BRIDGE}/api/v1/x/replies",
        json={"tweetId": tweet_id, "text": text}, timeout=15)
    resp.raise_for_status()
    data = resp.json()
    errors = data.get("data", {}).get("errors")
    if errors:
        raise RuntimeError("; ".join(f"code {e.get('code')}: {e.get('message')}" for e in errors))
    try:
        return data["data"]["data"]["create_tweet"]["tweet_results"]["result"]["rest_id"]
    except (KeyError, TypeError):
        return "unknown"


def main():
    print(f'Searching: "{SEARCH_QUERY}"')
    tweets = search_tweets(SEARCH_QUERY, count=10)
    if not tweets:
        print("No tweets found / 未找到相关推文"); sys.exit(1)

    # Try each tweet; skip RTs; on failure move to next.
    # 逐条尝试；跳过转发；失败则换下一条。
    replied = False
    for tweet in tweets:
        if tweet["text"].startswith("RT @"):
            continue
        print(f"Trying tweet {tweet['id']}: {tweet['text'][:60]!r}")
        try:
            reply_id = reply_to_tweet(tweet["id"], REPLY_TEXT)
            print(f"Done / 完成, reply ID: {reply_id}")
            replied = True
            break
        except Exception as e:
            print(f"  Skipped: {e}"); continue

    if not replied:
        print("Failed / 失败: no replyable tweet found / 搜索结果中无可回复推文")
        sys.exit(1)


if __name__ == "__main__":
    main()
Equivalent curl command
# Step 1: Search tweets / 搜索推文
curl -s "http://127.0.0.1:20088/api/v1/x/search?query=TweetPilot&count=5"

# Step 2: Reply to a tweet / 回复推文（替换 TWEET_ID）
curl -s -X POST http://127.0.0.1:20088/api/v1/x/replies \
  -H "Content-Type: application/json" \
  -d '{"tweetId": "TWEET_ID", "text": "Thanks for mentioning us! — TweetPilot"}'
↓ clawbot.py xdk.py http_rest.py
S04
Timeline Auto-Like
Read the home timeline and like tweets that match a keyword filter — a simple engagement automation.
Rate-limit warning: Only likes ONE matching tweet per run. Schedule every 30+ minutes. / ⚠️ 每次只点赞一条，建议每 30 分钟以上定时执行一次。 
"""
S04 · Timeline Auto-Like — ClawBot SDK
EN: Read home timeline, like ONLY the first tweet matching a keyword.
    ⚠️ Only ONE like per run — schedule every 30+ min to avoid rate-limits.
中文：读取主页时间线，只点赞第一条匹配关键词的推文。⚠️ 每次只点赞一条。

pip install requests  (clawbot pre-installed via TweetPilot)
"""

import sys
from clawbot import ClawBotClient

KEYWORD = "AI"  # ← Keyword filter (case-insensitive), "" = like first tweet / 关键词（空=点赞第一条）


def main():
    client = ClawBotClient()
    instances   = client.x.status.get_instances()
    instance_id = None
    if isinstance(instances, list) and instances:
        instance_id = instances[0].get("instanceId") or instances[0].get("id")

    print("Fetching timeline / 获取时间线...")
    tweets = client.x.timeline.list_timeline_tweets(instance_id=instance_id)
    if not tweets:
        print("No tweets in timeline / 时间线为空"); return

    # Find the first matching tweet — like ONLY that one
    # 找到第一条匹配推文，只点赞这一条
    target = next(
        (t for t in tweets if not KEYWORD or KEYWORD.lower() in (t.text or "").lower()),
        None
    )
    if not target:
        print(f'No tweet matched "{KEYWORD}" / 未找到包含关键词的推文'); return

    print(f"Liking {target.id}: {(target.text or '')[:60]!r}")
    try:
        result = client.x.actions.like(tweet_id=target.id, instance_id=instance_id)
        print(f"Done / 完成: {result}")
    except Exception as e:
        print(f"Failed / 失败: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()
Rate-limit warning: Only likes ONE matching tweet per run. Schedule every 30+ minutes. / ⚠️ 每次只点赞一条，建议每 30 分钟以上定时执行一次。 
"""
S04 · Timeline Auto-Like — xdk SDK
EN: Read home timeline via Twitter API v2, like ONLY the first matching tweet.
    ⚠️ Only ONE like per run — schedule every 30+ min to avoid rate-limits.
    Requires OAuth account authorized in TweetPilot (x-prefix accounts).
中文：通过 Twitter API v2 读取主页时间线，只点赞第一条匹配推文。

pip install xdk requests
"""

import sys
import requests
from xdk import Client

TWITTER_ID = "YOUR_TWITTER_ID"  # ← OAuth account numeric ID
KEYWORD    = "AI"               # ← Keyword filter, "" = like first tweet


def get_access_token(twitter_id):
    try:
        resp = requests.post("http://127.0.0.1:20088/api/v1/x/oauth/access-token",
            json={"twitter_id": twitter_id}, timeout=10)
        resp.raise_for_status()
        return resp.json()["access_token"]
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running"); sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code} — {e.response.text}"); sys.exit(1)


def main():
    if TWITTER_ID == "YOUR_TWITTER_ID":
        print("ERROR: Set TWITTER_ID to your numeric Twitter ID (OAuth only).")
        sys.exit(1)

    # Must use access_token= (User OAuth 2.0), NOT bearer_token= (read-only)
    # 必须用 access_token= 传入，bearer_token= 是只读的 App token
    client = Client(access_token=get_access_token(TWITTER_ID))

    print("Fetching timeline / 获取时间线...")
    target = None
    try:
        for page in client.users.get_timeline(id=TWITTER_ID, max_results=20):
            for tweet in (page.data or []):
                if not KEYWORD or KEYWORD.lower() in (tweet.text or "").lower():
                    target = tweet
                    break
            if target: break
    except Exception as e:
        print(f"ERROR: {e}"); sys.exit(1)

    if not target:
        print(f'No tweet matched "{KEYWORD}" / 未找到包含关键词的推文'); return

    print(f"Liking {target.id}: {(target.text or '')[:60]!r}")
    try:
        client.users.like_post(id=TWITTER_ID, body={"tweet_id": target.id})
        print(f"Done / 完成: liked tweet {target.id}")
    except Exception as e:
        print(f"Failed / 失败: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()
Calls LocalBridge REST API — browser extension acts as proxy, no OAuth token needed. To call Twitter's official API directly, see Twitter docs.
Rate-limit warning: Only likes ONE matching tweet per run. Schedule every 30+ min. / ⚠️ 每次只点赞一条，建议每 30 分钟以上执行一次。 
"""
S04 · Timeline Auto-Like — HTTP REST API (LocalBridge)
EN: Read home timeline via LocalBridge, like ONLY the first matching tweet.
    ⚠️ Only ONE like per run — schedule every 30+ min to avoid rate-limits.
pip install requests
"""

import sys
import requests

LOCAL_BRIDGE = "http://127.0.0.1:20088"
KEYWORD      = "AI"  # ← Keyword filter, "" = like first tweet / 关键词（空=点赞第一条）


def get_timeline():
    try:
        resp = requests.get(f"{LOCAL_BRIDGE}/api/v1/x/timeline", timeout=15)
        resp.raise_for_status()
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running"); sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code}"); sys.exit(1)

    tweets = []
    try:
        instructions = (resp.json().get("data", {}).get("data", {}).get("home", {})
            .get("home_timeline_urt", {}).get("instructions", []))
        for instr in instructions:
            for entry in instr.get("entries", []):
                r = entry.get("content", {}).get("itemContent", {}).get("tweet_results", {}).get("result", {})
                if not r: continue
                t = r.get("tweet") or r
                rid = t.get("rest_id") or r.get("rest_id")
                if rid:
                    tweets.append({"id": rid, "text": t.get("legacy", {}).get("full_text", "")})
    except Exception as e:
        print(f"WARNING: parse error: {e}")
    return tweets


def main():
    print("Fetching timeline / 获取时间线...")
    tweets = get_timeline()
    if not tweets:
        print("No tweets in timeline / 时间线为空"); return

    # Find the first matching tweet — like ONLY that one
    # 找到第一条匹配推文，只点赞这一条
    target = next(
        (t for t in tweets if not KEYWORD or KEYWORD.lower() in t["text"].lower()),
        None
    )
    if not target:
        print(f'No tweet matched "{KEYWORD}" / 未找到包含关键词的推文'); return

    print(f"Liking {target['id']}: {target['text'][:60]!r}")
    try:
        resp = requests.post(f"{LOCAL_BRIDGE}/api/v1/x/likes",
            json={"tweetId": target["id"]}, timeout=15)
        resp.raise_for_status()
        if resp.json().get("ok"):
            print(f"Done / 完成: liked tweet {target['id']}")
        else:
            print(f"Failed / 失败: {resp.json()}")
            sys.exit(1)
    except Exception as e:
        print(f"Failed / 失败: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()
Equivalent curl command
# Step 1: Fetch timeline / 获取时间线
curl -s http://127.0.0.1:20088/api/v1/x/timeline

# Step 2: Like a tweet / 点赞推文（替换 TWEET_ID）
curl -s -X POST http://127.0.0.1:20088/api/v1/x/likes \
  -H "Content-Type: application/json" \
  -d '{"tweetId": "TWEET_ID"}'
↓ clawbot.py xdk.py http_rest.py
S05
Batch Follow
Follow a list of user IDs in bulk — handy for building a target audience or tracking competitors.
"""
S05 · Batch Follow — ClawBot SDK
EN: Follow a list of user IDs in bulk via the TweetClaw browser extension.
    Adds a delay between each follow to avoid rate-limiting.
中文：通过 TweetClaw 浏览器扩展批量关注一组用户 ID，每次关注间隔等待。

pip install requests  (clawbot pre-installed via TweetPilot)
"""

import sys, time
from clawbot import ClawBotClient

TARGET_IDS = [
    "44196397",   # @elonmusk
    "783214",     # @x
]
DELAY_SECONDS = 5  # ← seconds between each follow / 每次关注间隔秒数


def main():
    client = ClawBotClient()
    instances   = client.x.status.get_instances()
    instance_id = None
    if isinstance(instances, list) and instances:
        instance_id = instances[0].get("instanceId") or instances[0].get("id")

    for i, user_id in enumerate(TARGET_IDS):
        print(f"Following {user_id} ({i + 1}/{len(TARGET_IDS)})...")
        try:
            # follow() → POST /api/v1/x/follows via LocalBridge
            result = client.x.actions.follow(user_id=user_id, instance_id=instance_id)
            print(f"  Done / 完成: {result}")
        except Exception as e:
            print(f"  Failed / 失败: {e}"); sys.exit(1)
        if i < len(TARGET_IDS) - 1:
            time.sleep(DELAY_SECONDS)

    print(f"\nAll done / 全部完成: followed {len(TARGET_IDS)} users")


if __name__ == "__main__":
    main()
"""
S05 · Batch Follow — xdk SDK
EN: Follow a list of user IDs in bulk via Twitter API v2.
    Adds a delay between each follow to avoid rate-limiting.
    Requires OAuth account authorized in TweetPilot.
中文：通过 Twitter API v2 批量关注一组用户 ID，每次关注间隔等待。

pip install xdk requests
"""

import sys, time, requests
from xdk import Client

TWITTER_ID    = "YOUR_TWITTER_ID"  # ← OAuth account numeric ID / OAuth 账号数字 ID
TARGET_IDS    = ["44196397", "783214"]  # ← IDs to follow / 要关注的用户 ID 列表
DELAY_SECONDS = 5


def get_access_token(twitter_id):
    try:
        resp = requests.post("http://127.0.0.1:20088/api/v1/x/oauth/access-token",
            json={"twitter_id": twitter_id}, timeout=10)
        resp.raise_for_status()
        return resp.json()["access_token"]
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running"); sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code} — {e.response.text}"); sys.exit(1)


def main():
    if TWITTER_ID == "YOUR_TWITTER_ID":
        print("ERROR: Set TWITTER_ID to your numeric Twitter ID."); sys.exit(1)

    # Must use access_token= (User OAuth 2.0), NOT bearer_token= (read-only)
    client = Client(access_token=get_access_token(TWITTER_ID))

    for i, target_id in enumerate(TARGET_IDS):
        print(f"Following {target_id} ({i + 1}/{len(TARGET_IDS)})...")
        try:
            # follow_user requires your own user ID + target user ID
            client.users.follow_user(id=TWITTER_ID, body={"target_user_id": target_id})
            print(f"  Done / 完成: followed {target_id}")
        except Exception as e:
            body = e.response.text if hasattr(e, "response") and e.response else ""
            print(f"  Failed / 失败: {body or e}"); sys.exit(1)
        if i < len(TARGET_IDS) - 1:
            time.sleep(DELAY_SECONDS)

    print(f"\nAll done / 全部完成: followed {len(TARGET_IDS)} users")


if __name__ == "__main__":
    main()
Calls LocalBridge REST API — browser extension acts as proxy, no OAuth token needed. To call Twitter's official API directly, see Twitter docs. 
"""
S05 · Batch Follow — HTTP REST API (LocalBridge)
EN: Follow a list of user IDs in bulk via LocalBridge.
    Adds a delay between each follow to avoid rate-limiting.
pip install requests
"""

import sys, time, requests

LOCAL_BRIDGE  = "http://127.0.0.1:20088"
TARGET_IDS    = ["44196397", "783214"]  # ← IDs to follow / 要关注的用户 ID 列表
DELAY_SECONDS = 5


def follow_user(user_id):
    # POST /api/v1/x/follows — body: {"userId": "..."}
    try:
        resp = requests.post(f"{LOCAL_BRIDGE}/api/v1/x/follows",
            json={"userId": user_id}, timeout=15)
        resp.raise_for_status()
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running"); sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code} — {e.response.text}"); sys.exit(1)
    data = resp.json()
    if not data.get("ok") and not data.get("success"):
        raise RuntimeError(f"Follow failed: {data}")


def main():
    for i, user_id in enumerate(TARGET_IDS):
        print(f"Following {user_id} ({i + 1}/{len(TARGET_IDS)})...")
        try:
            follow_user(user_id)
            print(f"  Done / 完成: followed {user_id}")
        except Exception as e:
            print(f"  Failed / 失败: {e}"); sys.exit(1)
        if i < len(TARGET_IDS) - 1:
            time.sleep(DELAY_SECONDS)

    print(f"\nAll done / 全部完成: followed {len(TARGET_IDS)} users")


if __name__ == "__main__":
    main()
Equivalent curl command
# Follow a user via LocalBridge / 通过 LocalBridge 关注用户（替换 USER_ID）
curl -s -X POST http://127.0.0.1:20088/api/v1/x/follows \
  -H "Content-Type: application/json" \
  -d '{"userId": "USER_ID"}'
↓ clawbot.py xdk.py http_rest.py
S06
Tweet Metrics Report
Fetch engagement metrics (likes, retweets, replies) for a specific tweet and print or save a report.
"""
S06 · Tweet Metrics Report — ClawBot SDK
EN: Fetch engagement metrics for a tweet via TweetClaw browser extension.
    Metrics live in the raw GraphQL "legacy" field.
中文：通过 TweetClaw 浏览器扩展获取指定推文的互动数据（点赞/转推/回复/书签）。

pip install requests  (clawbot pre-installed via TweetPilot)
"""

import sys
from clawbot import ClawBotClient

TWEET_ID = "1234567890123456789"  # ← Replace with target tweet ID / 替换为目标推文 ID


def main():
    client = ClawBotClient()
    instances   = client.x.status.get_instances()
    instance_id = None
    if isinstance(instances, list) and instances:
        instance_id = instances[0].get("instanceId") or instances[0].get("id")

    print(f"Fetching tweet {TWEET_ID}...")
    try:
        # get_tweet() fetches full tweet object; metrics are in tweet.raw["legacy"]
        tweet = client.x.tweets.get_tweet(tweet_id=TWEET_ID, instance_id=instance_id)
    except Exception as e:
        print(f"ERROR: {e}"); sys.exit(1)

    if not tweet or not tweet.id:
        print("ERROR: Tweet not found / 未找到该推文"); sys.exit(1)

    legacy = tweet.raw.get("legacy", {})
    print(f"\n── Tweet Metrics Report / 推文互动报告 ──")
    print(f"ID        : {tweet.id}")
    print(f"Author    : @{tweet.author_screen_name}")
    print(f"Text      : {(tweet.text or '')[:120]}")
    print(f"Likes     : {legacy.get('favorite_count', 'N/A')}")
    print(f"Retweets  : {legacy.get('retweet_count',  'N/A')}")
    print(f"Replies   : {legacy.get('reply_count',    'N/A')}")
    print(f"Quotes    : {legacy.get('quote_count',    'N/A')}")
    print(f"Bookmarks : {legacy.get('bookmark_count', 'N/A')}")


if __name__ == "__main__":
    main()
"""
S06 · Tweet Metrics Report — xdk SDK
EN: Fetch engagement metrics via Twitter API v2 public_metrics field.
    Requires OAuth account authorized in TweetPilot.
中文：通过 Twitter API v2 public_metrics 字段获取推文互动数据。

pip install xdk requests
"""

import sys, requests
from xdk import Client

TWITTER_ID = "YOUR_TWITTER_ID"       # ← OAuth account numeric ID / OAuth 账号数字 ID
TWEET_ID   = "1234567890123456789"   # ← Replace with target tweet ID / 替换为目标推文 ID


def get_access_token(twitter_id):
    try:
        resp = requests.post("http://127.0.0.1:20088/api/v1/x/oauth/access-token",
            json={"twitter_id": twitter_id}, timeout=10)
        resp.raise_for_status()
        return resp.json()["access_token"]
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running"); sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code} — {e.response.text}"); sys.exit(1)


def main():
    if TWITTER_ID == "YOUR_TWITTER_ID":
        print("ERROR: Set TWITTER_ID to your numeric Twitter ID."); sys.exit(1)

    client = Client(access_token=get_access_token(TWITTER_ID))
    print(f"Fetching tweet {TWEET_ID}...")
    try:
        # get_by_id with public_metrics returns likes/retweets/replies/impressions
        resp = client.posts.get_by_id(id=TWEET_ID,
            tweet_fields=["public_metrics", "text", "author_id"])
    except Exception as e:
        body = e.response.text if hasattr(e, "response") and e.response else ""
        print(f"ERROR: {body or e}"); sys.exit(1)

    tweet = resp.data if hasattr(resp, "data") else resp
    if not tweet:
        print("ERROR: Tweet not found"); sys.exit(1)

    # xdk may return a dict or an object — handle both
    text    = tweet.get("text", "") if isinstance(tweet, dict) else (tweet.text or "")
    metrics = tweet.get("public_metrics", {}) if isinstance(tweet, dict) else (tweet.public_metrics or {})
    m = metrics if isinstance(metrics, dict) else {}

    print(f"\n── Tweet Metrics Report / 推文互动报告 ──")
    print(f"ID          : {TWEET_ID}")
    print(f"Text        : {text[:120]}")
    print(f"Likes       : {m.get('like_count',       'N/A')}")
    print(f"Retweets    : {m.get('retweet_count',    'N/A')}")
    print(f"Replies     : {m.get('reply_count',      'N/A')}")
    print(f"Quotes      : {m.get('quote_count',      'N/A')}")
    print(f"Impressions : {m.get('impression_count', 'N/A')}")


if __name__ == "__main__":
    main()
Calls LocalBridge REST API — browser extension acts as proxy, no OAuth token needed. To call Twitter's official API directly, see Twitter docs. 
"""
S06 · Tweet Metrics Report — HTTP REST API (LocalBridge)
EN: Fetch tweet engagement metrics via LocalBridge GET /api/v1/x/tweets.
    Metrics live in the GraphQL response under the "legacy" key.
pip install requests
"""

import sys, requests

LOCAL_BRIDGE = "http://127.0.0.1:20088"
TWEET_ID     = "1234567890123456789"  # ← Replace with target tweet ID / 替换为目标推文 ID


def get_tweet(tweet_id):
    # GET /api/v1/x/tweets?tweetId=<id>
    # Response: {"success": true, "data": {"data": {"threaded_conversation_with_injections_v2": {...}}}}
    try:
        resp = requests.get(f"{LOCAL_BRIDGE}/api/v1/x/tweets",
            params={"tweetId": tweet_id}, timeout=15)
        resp.raise_for_status()
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running"); sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code}"); sys.exit(1)
    return resp.json().get("data", {}).get("data", {})


def main():
    print(f"Fetching tweet {TWEET_ID}...")
    data = get_tweet(TWEET_ID)
    if not data:
        print("ERROR: Tweet not found"); sys.exit(1)

    # Navigate threaded_conversation → instructions → first TimelineTweet entry
    instructions = (data.get("threaded_conversation_with_injections_v2", {})
                        .get("instructions", []))
    tweet = None
    for instr in instructions:
        for entry in instr.get("entries", []):
            item = entry.get("content", {}).get("itemContent", {})
            if item.get("itemType") == "TimelineTweet":
                r = item.get("tweet_results", {}).get("result", {})
                tweet = r.get("tweet") or r
                break
        if tweet: break

    if not tweet:
        print("ERROR: Could not parse tweet data"); sys.exit(1)

    legacy      = tweet.get("legacy", {})
    user_result = tweet.get("core", {}).get("user_results", {}).get("result", {})
    screen_name = (user_result.get("core", {}).get("screen_name")
                   or user_result.get("legacy", {}).get("screen_name", "unknown"))

    print(f"\n── Tweet Metrics Report / 推文互动报告 ──")
    print(f"ID        : {TWEET_ID}")
    print(f"Author    : @{screen_name}")
    print(f"Text      : {legacy.get('full_text', '')[:120]}")
    print(f"Likes     : {legacy.get('favorite_count', 'N/A')}")
    print(f"Retweets  : {legacy.get('retweet_count',  'N/A')}")
    print(f"Replies   : {legacy.get('reply_count',    'N/A')}")
    print(f"Quotes    : {legacy.get('quote_count',    'N/A')}")
    print(f"Bookmarks : {legacy.get('bookmark_count', 'N/A')}")


if __name__ == "__main__":
    main()
Equivalent curl command
# Fetch tweet metrics via LocalBridge / 获取推文互动数据（替换 TWEET_ID）
curl -s "http://127.0.0.1:20088/api/v1/x/tweets?tweetId=TWEET_ID" | \
  python3 -c "
import json,sys
d=json.load(sys.stdin).get('data',{}).get('data',{})
r=d.get('tweetResult',{}).get('result',{})
t=r.get('tweet') or r
lg=t.get('legacy',{})
print('Likes:',lg.get('favorite_count'))
print('Retweets:',lg.get('retweet_count'))
print('Replies:',lg.get('reply_count'))
"
↓ clawbot.py xdk.py http_rest.py
S07
Mention Monitor + Feishu Alert
Poll the mentions timeline and send a Feishu Webhook notification to your team when new mentions arrive.
Feishu prerequisite / 飞书前提： Feishu must be configured in TweetPilot settings before these scripts will work. The /api/v1/feishu/send endpoint is only accessible locally — it listens on 127.0.0.1 at the REST API port you set in TweetPilot's system settings (default 20088). Remote or public IPs will be rejected.
飞书需在 TweetPilot 系统设置中完成配置后脚本才能使用。/api/v1/feishu/send 仅监听本机 127.0.0.1 + 系统设置中配置的 REST API 端口（默认 20088），外部 IP 访问无效。 
"""
S07 · Mention Monitor + Feishu Alert — ClawBot SDK
EN: Search recent @mentions via TweetClaw, send Feishu alert for new ones.
    Uses TweetPilot's built-in Feishu channel — no Webhook URL needed.
    Stores last-seen tweet ID locally so only new mentions are reported.
中文：通过 TweetClaw 搜索最新提及，通过 TweetPilot 内置飞书通道推送新提及。
     无需配置飞书 Webhook，每次只上报新提及。

pip install requests  (clawbot pre-installed via TweetPilot)
"""

import sys, requests
from pathlib import Path
from clawbot import ClawBotClient

LOCAL_BRIDGE = "http://127.0.0.1:20088"
SCREEN_NAME  = "your_handle"  # ← Your Twitter handle (no @) / 你的用户名（不含 @）
STATE_FILE   = Path.home() / ".tweetpilot" / "s07_last_mention_id.txt"


def load_last_id():
    try: return STATE_FILE.read_text().strip() or None
    except FileNotFoundError: return None

def save_last_id(tid):
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    STATE_FILE.write_text(tid)

def send_feishu(text, tweet_id, author):
    url = f"https://x.com/{author}/status/{tweet_id}"
    msg = f"🔔 New mention\n@{author}: {text[:200]}\n{url}"
    # POST /api/v1/feishu/send — TweetPilot built-in Feishu channel, no Webhook needed
    r = requests.post(f"{LOCAL_BRIDGE}/api/v1/feishu/send",
        json={"text": msg}, timeout=10)
    r.raise_for_status()
    if not r.json().get("ok"):
        raise RuntimeError(f"Feishu send failed: {r.json()}")


def main():
    if SCREEN_NAME == "your_handle":
        print("ERROR: Set SCREEN_NAME to your Twitter handle."); sys.exit(1)

    client = ClawBotClient()
    instances   = client.x.status.get_instances()
    instance_id = None
    if isinstance(instances, list) and instances:
        instance_id = instances[0].get("instanceId") or instances[0].get("id")

    last_id = load_last_id()
    print(f"Searching @{SCREEN_NAME} mentions (since_id={last_id or 'none'})...")

    try:
        # search_tweets returns newest-first results for the query
        tweets = client.x.tweets.search_tweets(
            query=f"@{SCREEN_NAME}", count=20, instance_id=instance_id)
    except Exception as e:
        print(f"ERROR: {e}"); sys.exit(1)

    # Keep only tweets newer than last seen ID
    new_tweets = [t for t in tweets if not last_id or t.id > last_id]
    if not new_tweets:
        print("No new mentions / 无新提及"); return

    print(f"Found {len(new_tweets)} new mention(s) / 发现 {len(new_tweets)} 条新提及")
    for t in reversed(new_tweets):
        author = getattr(t, "author_screen_name", "unknown") or "unknown"
        text   = getattr(t, "text", "") or ""
        print(f"  @{author}: {text[:80]!r}")
        try: send_feishu(text, t.id, author)
        except Exception as e: print(f"  Feishu failed: {e}")

    save_last_id(new_tweets[0].id)
    print(f"Done / 完成. Last ID: {new_tweets[0].id}")


if __name__ == "__main__":
    main()
"""
S07 · Mention Monitor + Feishu Alert — xdk SDK
EN: Poll recent mentions via Twitter API v2, send Feishu alert for new ones.
    Uses TweetPilot's built-in Feishu channel — no Webhook URL needed.
    Passes since_id so only mentions newer than last run are fetched.
    Requires OAuth account authorized in TweetPilot.
中文：通过 Twitter API v2 轮询最新提及，通过 TweetPilot 内置飞书通道推送新提及。
     无需配置飞书 Webhook，传入 since_id 只获取上次运行后的新提及。

pip install xdk requests
"""

import sys, requests
from pathlib import Path
from xdk import Client

LOCAL_BRIDGE = "http://127.0.0.1:20088"
TWITTER_ID   = "YOUR_TWITTER_ID"  # ← OAuth account numeric ID / OAuth 账号数字 ID
STATE_FILE   = Path.home() / ".tweetpilot" / "s07_last_mention_id.txt"


def get_access_token(twitter_id):
    try:
        resp = requests.post(f"{LOCAL_BRIDGE}/api/v1/x/oauth/access-token",
            json={"twitter_id": twitter_id}, timeout=10)
        resp.raise_for_status()
        return resp.json()["access_token"]
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running"); sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code} — {e.response.text}"); sys.exit(1)

def load_last_id():
    try: return STATE_FILE.read_text().strip() or None
    except FileNotFoundError: return None

def save_last_id(tid):
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    STATE_FILE.write_text(tid)

def send_feishu(text, tweet_id):
    url = f"https://x.com/i/web/status/{tweet_id}"
    msg = f"🔔 New mention\n{text[:200]}\n{url}"
    # POST /api/v1/feishu/send — TweetPilot built-in Feishu channel, no Webhook needed
    r = requests.post(f"{LOCAL_BRIDGE}/api/v1/feishu/send",
        json={"text": msg}, timeout=10)
    r.raise_for_status()
    if not r.json().get("ok"):
        raise RuntimeError(f"Feishu send failed: {r.json()}")


def main():
    if TWITTER_ID == "YOUR_TWITTER_ID":
        print("ERROR: Set TWITTER_ID to your numeric Twitter ID."); sys.exit(1)

    # Must use access_token= (User OAuth 2.0), NOT bearer_token= (read-only)
    client = Client(access_token=get_access_token(TWITTER_ID))

    last_id = load_last_id()
    print(f"Fetching mentions for {TWITTER_ID} (since_id={last_id or 'none'})...")

    new_tweets = []
    try:
        # get_mentions returns newest-first paginated mention timeline
        # Pass since_id to only fetch mentions newer than the last run
        for page in client.users.get_mentions(
                id=TWITTER_ID, max_results=20, since_id=last_id,
                tweet_fields=["text", "author_id"]):
            new_tweets.extend(page.data or [])
            break  # one page per run is enough
    except Exception as e:
        body = e.response.text if hasattr(e, "response") and e.response else ""
        print(f"ERROR: {body or e}"); sys.exit(1)

    if not new_tweets:
        print("No new mentions / 无新提及"); return

    print(f"Found {len(new_tweets)} new mention(s) / 发现 {len(new_tweets)} 条新提及")
    for t in reversed(new_tweets):
        # xdk get_mentions returns dicts — use dict access
        text     = t.get("text", "") if isinstance(t, dict) else (t.text or "")
        tweet_id = t.get("id")      if isinstance(t, dict) else t.id
        print(f"  {tweet_id}: {text[:80]!r}")
        try: send_feishu(text, tweet_id)
        except Exception as e: print(f"  Feishu failed: {e}")

    first = new_tweets[0]
    newest_id = first.get("id") if isinstance(first, dict) else first.id
    save_last_id(newest_id)
    print(f"Done / 完成. Last ID: {newest_id}")


if __name__ == "__main__":
    main()
Calls LocalBridge REST API — browser extension acts as proxy, no OAuth token needed. To call Twitter's official API directly, see Twitter docs. 
"""
S07 · Mention Monitor + Feishu Alert — HTTP REST API (LocalBridge)
EN: Search @mentions via LocalBridge, send Feishu alert for new ones.
    Uses TweetPilot's built-in Feishu channel — no Webhook URL needed.
    Stores last-seen tweet ID locally so only new mentions are reported.
pip install requests
"""

import sys, requests
from pathlib import Path

LOCAL_BRIDGE  = "http://127.0.0.1:20088"
SCREEN_NAME   = "your_handle"  # ← Your Twitter handle (no @) / 你的用户名（不含 @）
STATE_FILE    = Path.home() / ".tweetpilot" / "s07_last_mention_id.txt"


def load_last_id():
    try: return STATE_FILE.read_text().strip() or None
    except FileNotFoundError: return None

def save_last_id(tid):
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    STATE_FILE.write_text(tid)

def search_mentions(query):
    # GET /api/v1/x/search?query=@handle&count=20
    # Response: {"success": true, "data": {"data": {...GraphQL...}}}
    try:
        resp = requests.get(f"{LOCAL_BRIDGE}/api/v1/x/search",
            params={"query": query, "count": 20}, timeout=15)
        resp.raise_for_status()
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running"); sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code}"); sys.exit(1)

    tweets = []
    try:
        instructions = (resp.json().get("data", {}).get("data", {})
            .get("search_by_raw_query", {}).get("search_timeline", {})
            .get("timeline", {}).get("instructions", []))
        for instr in instructions:
            for entry in instr.get("entries", []):
                r = entry.get("content", {}).get("itemContent", {}).get("tweet_results", {}).get("result", {})
                if not r: continue
                t   = r.get("tweet") or r
                rid = t.get("rest_id") or r.get("rest_id")
                sn  = (t.get("core", {}).get("user_results", {})
                        .get("result", {}).get("core", {}).get("screen_name", "unknown"))
                if rid:
                    tweets.append({"id": rid, "text": t.get("legacy", {}).get("full_text", ""),
                                   "author_screen_name": sn})
    except Exception as e:
        print(f"WARNING: parse error: {e}")
    return tweets

def send_feishu(text, tweet_id, author):
    url = f"https://x.com/{author}/status/{tweet_id}"
    msg = f"🔔 New mention\n@{author}: {text[:200]}\n{url}"
    # POST /api/v1/feishu/send — TweetPilot built-in Feishu channel, no Webhook needed
    r = requests.post(f"{LOCAL_BRIDGE}/api/v1/feishu/send",
        json={"text": msg}, timeout=10)
    r.raise_for_status()
    if not r.json().get("ok"):
        raise RuntimeError(f"Feishu send failed: {r.json()}")


def main():
    if SCREEN_NAME == "your_handle":
        print("ERROR: Set SCREEN_NAME to your Twitter handle."); sys.exit(1)

    last_id = load_last_id()
    print(f"Searching @{SCREEN_NAME} mentions (since_id={last_id or 'none'})...")

    tweets = search_mentions(f"@{SCREEN_NAME}")
    new_tweets = [t for t in tweets if not last_id or t["id"] > last_id]

    if not new_tweets:
        print("No new mentions / 无新提及"); return

    print(f"Found {len(new_tweets)} new mention(s) / 发现 {len(new_tweets)} 条新提及")
    for t in reversed(new_tweets):
        print(f"  @{t['author_screen_name']}: {t['text'][:80]!r}")
        try: send_feishu(t["text"], t["id"], t["author_screen_name"])
        except Exception as e: print(f"  Feishu failed: {e}")

    save_last_id(new_tweets[0]["id"])
    print(f"Done / 完成. Last ID: {new_tweets[0]['id']}")


if __name__ == "__main__":
    main()
Equivalent curl command
# Search @mentions via LocalBridge / 搜索提及（替换 your_handle）
curl -s "http://127.0.0.1:20088/api/v1/x/search?query=%40your_handle&count=20"

# Send Feishu alert via TweetPilot built-in channel / 通过 TweetPilot 内置飞书通道发送通知
curl -s -X POST "http://127.0.0.1:20088/api/v1/feishu/send" \
  -H "Content-Type: application/json" \
  -d '{"text":"🔔 New mention: MESSAGE"}'
↓ clawbot.py xdk.py http_rest.py
S08
Retweet Trending Topics
Fetch current trending topics, find high-engagement tweets within a trend, and retweet them to ride the wave.
"""
S08 · Retweet Trending Topics — ClawBot SDK
============================================
EN: Search for high-engagement tweets on a given topic via TweetClaw,
    sort by likes, then retweet the top N results.
    No trending API needed — uses keyword search + engagement filtering.
中文：通过 TweetClaw 搜索指定话题的高互动推文，
     按点赞数排序后转推前 N 条。
     无需 trending API，以关键词搜索 + 互动数过滤实现。

Requirements / 依赖:
  pip install requests
  TweetPilot running + TweetClaw browser extension online
"""

import sys
import time
from clawbot import ClawBotClient

# ── Config / 配置 ────────────────────────────────────────────────────
# EN: Topic keyword or hashtag to search.
# 中文：搜索关键词或 hashtag。
TOPIC = "#AI"  # ← Replace / 替换为你想搜索的话题

# EN: Only retweet tweets with at least this many likes.
# 中文：仅转推点赞数不低于此值的推文。
MIN_LIKES = 100  # ← Minimum likes threshold / 最低点赞数

# EN: Maximum number of tweets to retweet per run.
# 中文：每次最多转推条数。
TOP_N = 3  # ← Max retweets per run / 每次最多转推数

# EN: Seconds to wait between retweets (avoid rate limiting).
# 中文：每次转推之间的等待秒数（避免频率限制）。
DELAY_SECONDS = 3


def parse_search_raw(raw: dict) -> list:
    """
    EN: Parse raw LocalBridge search response into a flat list of tweet dicts.
        Each dict has: id, text, author_screen_name, likes.
    中文：解析 LocalBridge 搜索原始响应，返回推文字典列表。
         每条含 id、text、author_screen_name、likes。
    """
    tweets = []
    try:
        instructions = (
            raw.get("data", {})
               .get("data", {})
               .get("search_by_raw_query", {})
               .get("search_timeline", {})
               .get("timeline", {})
               .get("instructions", [])
        )
        for instruction in instructions:
            for entry in instruction.get("entries", []):
                result = (
                    entry.get("content", {})
                         .get("itemContent", {})
                         .get("tweet_results", {})
                         .get("result", {})
                )
                if not result:
                    continue
                # EN: Handle TweetWithVisibilityResults wrapper.
                # 中文：兼容 TweetWithVisibilityResults 包装格式。
                tweet   = result.get("tweet") or result
                rest_id = tweet.get("rest_id") or result.get("rest_id")
                legacy  = tweet.get("legacy", {})
                text    = legacy.get("full_text", "")
                likes   = legacy.get("favorite_count", 0)
                screen_name = (
                    tweet.get("core", {})
                         .get("user_results", {})
                         .get("result", {})
                         .get("core", {})
                         .get("screen_name", "unknown")
                )
                if rest_id:
                    tweets.append({
                        "id": rest_id,
                        "text": text,
                        "author_screen_name": screen_name,
                        "likes": likes,
                    })
    except Exception as e:
        print(f"WARNING: Could not parse search results: {e}")
    return tweets


def main():
    client = ClawBotClient()

    # EN: Auto-select first connected TweetClaw instance.
    # 中文：自动选择第一个已连接的 TweetClaw 实例。
    instances = client.x.status.get_instances()
    instance_id = None
    if isinstance(instances, list) and instances:
        instance_id = instances[0].get("instanceId") or instances[0].get("id")

    print(f"Searching tweets for topic {TOPIC!r}...")

    try:
        raw = client.x.tweets.transport.search_raw(
            query=TOPIC,
            count=20,
            instance_id=instance_id,
        )
    except Exception as e:
        print(f"ERROR: Search failed: {e}")
        sys.exit(1)

    tweets = parse_search_raw(raw)

    if not tweets:
        print("No tweets found / 未找到推文")
        return

    # EN: Sort by likes descending, then apply MIN_LIKES filter.
    # 中文：按点赞数降序排序，再过滤低于门槛的推文。
    sorted_tweets = sorted(tweets, key=lambda t: t["likes"], reverse=True)
    top_tweets = [t for t in sorted_tweets if t["likes"] >= MIN_LIKES][:TOP_N]

    if not top_tweets:
        print(f"No tweets with >= {MIN_LIKES} likes / 没有点赞数达到 {MIN_LIKES} 的推文")
        return

    print(f"Retweeting top {len(top_tweets)} tweet(s) / 转推前 {len(top_tweets)} 条...")

    for t in top_tweets:
        print(f"  [{t['likes']} likes] @{t['author_screen_name']}: {t['text'][:80]!r}")
        try:
            client.x.actions.retweet(tweet_id=t["id"], instance_id=instance_id)
            print(f"  ✓ Retweeted / 已转推 {t['id']}")
        except Exception as e:
            print(f"  Retweet failed: {e}")
        time.sleep(DELAY_SECONDS)

    print("Done / 完成")


if __name__ == "__main__":
    main()
"""
S08 · Retweet Trending Topics — xdk SDK
=========================================
EN: Search for high-engagement tweets on a given topic via Twitter API v2,
    sort by public metrics, then retweet the top N results.
    TweetPilot provides the OAuth access token automatically.
中文：通过 Twitter API v2 搜索指定话题的高互动推文，
     按互动数排序后转推前 N 条。
     TweetPilot 自动提供 OAuth access token。

Requirements / 依赖:
  pip install xdk requests
  TweetPilot running + OAuth account authorized in Account Settings
"""

import sys
import time
import requests
from xdk import Client

# ── Config / 配置 ────────────────────────────────────────────────────
LOCAL_BRIDGE = "http://127.0.0.1:20088"
TWITTER_ID   = "YOUR_TWITTER_ID"  # ← OAuth account numeric ID / OAuth 账号数字 ID

# EN: Topic keyword or hashtag to search.
# 中文：搜索关键词或 hashtag。
TOPIC = "#AI"  # ← Replace / 替换为你想搜索的话题

# EN: Only retweet tweets with at least this many likes.
# 中文：仅转推点赞数不低于此值的推文。
MIN_LIKES = 100  # ← Minimum likes threshold / 最低点赞数

# EN: Maximum number of tweets to retweet per run.
# 中文：每次最多转推条数。
TOP_N = 3  # ← Max retweets per run / 每次最多转推数

# EN: Seconds to wait between retweets (avoid rate limiting).
# 中文：每次转推之间的等待秒数（避免频率限制）。
DELAY_SECONDS = 3


def get_access_token(twitter_id: str) -> str:
    try:
        resp = requests.post(
            f"{LOCAL_BRIDGE}/api/v1/x/oauth/access-token",
            json={"twitter_id": twitter_id},
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json()["access_token"]
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running / TweetPilot 未在运行")
        sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code} — {e.response.text}")
        sys.exit(1)


def get_likes(tweet) -> int:
    """Extract like count from tweet dict or object."""
    if isinstance(tweet, dict):
        m = tweet.get("public_metrics") or {}
        return m.get("like_count", 0)
    m = getattr(tweet, "public_metrics", None) or {}
    return m.get("like_count", 0) if isinstance(m, dict) else 0


def main():
    if TWITTER_ID == "YOUR_TWITTER_ID":
        print("ERROR: Set TWITTER_ID to your numeric Twitter ID."); sys.exit(1)

    # EN: Must use access_token= (User OAuth 2.0), NOT bearer_token=.
    # 中文：必须用 access_token= 传入，bearer_token= 是只读的 App token。
    client = Client(access_token=get_access_token(TWITTER_ID))

    print(f"Searching tweets for topic {TOPIC!r}...")

    tweets = []
    try:
        for page in client.posts.search_recent(
            query=TOPIC,
            max_results=20,
            tweet_fields=["public_metrics", "text", "author_id"],
        ):
            tweets.extend(page.data or [])
            break  # EN: One page per run is enough. / 每次运行取一页即可。
    except Exception as e:
        err_body = getattr(getattr(e, "response", None), "text", "") or str(e)
        print(f"ERROR: Search failed: {err_body}")
        sys.exit(1)

    if not tweets:
        print("No tweets found / 未找到推文")
        return

    # EN: Sort by likes descending, then apply MIN_LIKES filter.
    # 中文：按点赞数降序排序，再过滤低于门槛的推文。
    sorted_tweets = sorted(tweets, key=get_likes, reverse=True)
    top_tweets = [t for t in sorted_tweets if get_likes(t) >= MIN_LIKES][:TOP_N]

    if not top_tweets:
        print(f"No tweets with >= {MIN_LIKES} likes / 没有点赞数达到 {MIN_LIKES} 的推文")
        return

    print(f"Retweeting top {len(top_tweets)} tweet(s) / 转推前 {len(top_tweets)} 条...")

    for t in top_tweets:
        tweet_id = t.get("id") if isinstance(t, dict) else t.id
        text     = (t.get("text", "") if isinstance(t, dict) else getattr(t, "text", "") or "")[:80]
        likes    = get_likes(t)
        print(f"  [{likes} likes] {tweet_id}: {text!r}")
        try:
            # EN: repost_post uses body with tweet_id.
            # 中文：repost_post 需要 body 中包含 tweet_id。
            client.users.repost_post(
                id=TWITTER_ID,
                body={"tweet_id": tweet_id},
            )
            print(f"  ✓ Retweeted / 已转推 {tweet_id}")
        except Exception as e:
            err_body = getattr(getattr(e, "response", None), "text", "") or str(e)
            print(f"  Retweet failed: {err_body}")
        time.sleep(DELAY_SECONDS)

    print("Done / 完成")


if __name__ == "__main__":
    main()
Calls LocalBridge REST API — browser extension acts as proxy, no OAuth token needed. To call Twitter's official API directly, see Twitter docs. 
"""
S08 · Retweet Trending Topics — HTTP REST API (LocalBridge)
=============================================================
EN: Search for high-engagement tweets on a given topic via LocalBridge,
    sort by likes, then retweet the top N results.
    Uses the browser extension as proxy — no OAuth token needed.
中文：通过 LocalBridge 搜索指定话题的高互动推文，
     按点赞数排序后转推前 N 条。
     以浏览器扩展作为代理，无需 OAuth token。

Requirements / 依赖:
  pip install requests
  TweetPilot running + TweetClaw browser extension online
"""

import sys
import time
import requests

# ── Config / 配置 ────────────────────────────────────────────────────
LOCAL_BRIDGE = "http://127.0.0.1:20088"

# EN: Topic keyword or hashtag to search.
# 中文：搜索关键词或 hashtag。
TOPIC = "#AI"  # ← Replace / 替换为你想搜索的话题

# EN: Only retweet tweets with at least this many likes.
# 中文：仅转推点赞数不低于此值的推文。
MIN_LIKES = 100  # ← Minimum likes threshold / 最低点赞数

# EN: Maximum number of tweets to retweet per run.
# 中文：每次最多转推条数。
TOP_N = 3  # ← Max retweets per run / 每次最多转推数

# EN: Seconds to wait between retweets (avoid rate limiting).
# 中文：每次转推之间的等待秒数（避免频率限制）。
DELAY_SECONDS = 3


def search_tweets(query: str) -> list:
    """
    EN: Search tweets via LocalBridge GET /api/v1/x/search.
        Returns a list of tweet dicts with id, text, author_screen_name, and likes.
    中文：通过 LocalBridge GET /api/v1/x/search 搜索推文。
         返回含 id、text、author_screen_name、likes 的推文字典列表。
    """
    try:
        resp = requests.get(
            f"{LOCAL_BRIDGE}/api/v1/x/search",
            params={"query": query, "count": 20},
            timeout=15,
        )
        resp.raise_for_status()
    except requests.exceptions.ConnectionError:
        print("ERROR: TweetPilot not running / TweetPilot 未在运行")
        sys.exit(1)
    except requests.exceptions.HTTPError as e:
        print(f"ERROR: {e.response.status_code} — {e.response.text}")
        sys.exit(1)

    tweets = []
    try:
        # EN: LocalBridge wraps response: {"success": true, "data": {"data": {...GraphQL...}}}
        # 中文：LocalBridge 包装响应：{"success": true, "data": {"data": {...GraphQL...}}}
        instructions = (
            resp.json()
                .get("data", {})
                .get("data", {})
                .get("search_by_raw_query", {})
                .get("search_timeline", {})
                .get("timeline", {})
                .get("instructions", [])
        )
        for instruction in instructions:
            for entry in instruction.get("entries", []):
                result = (
                    entry.get("content", {})
                         .get("itemContent", {})
                         .get("tweet_results", {})
                         .get("result", {})
                )
                if not result:
                    continue
                # EN: Handle TweetWithVisibilityResults wrapper.
                # 中文：兼容 TweetWithVisibilityResults 包装格式。
                tweet   = result.get("tweet") or result
                rest_id = tweet.get("rest_id") or result.get("rest_id")
                legacy  = tweet.get("legacy", {})
                text    = legacy.get("full_text", "")
                likes   = legacy.get("favorite_count", 0)
                screen_name = (
                    tweet.get("core", {})
                         .get("user_results", {})
                         .get("result", {})
                         .get("core", {})
                         .get("screen_name", "unknown")
                )
                if rest_id:
                    tweets.append({
                        "id": rest_id,
                        "text": text,
                        "author_screen_name": screen_name,
                        "likes": likes,
                    })
    except Exception as e:
        print(f"WARNING: Could not parse search results: {e}")
    return tweets


def retweet(tweet_id: str) -> None:
    """
    EN: Retweet via LocalBridge POST /api/v1/x/retweets.
    中文：通过 LocalBridge POST /api/v1/x/retweets 转推。
    """
    resp = requests.post(
        f"{LOCAL_BRIDGE}/api/v1/x/retweets",
        json={"tweetId": tweet_id},
        timeout=15,
    )
    resp.raise_for_status()


def main():
    print(f"Searching tweets for topic {TOPIC!r}...")

    tweets = search_tweets(TOPIC)

    if not tweets:
        print("No tweets found / 未找到推文")
        return

    # EN: Sort by likes descending, then apply MIN_LIKES filter.
    # 中文：按点赞数降序排序，再过滤低于门槛的推文。
    sorted_tweets = sorted(tweets, key=lambda t: t["likes"], reverse=True)
    top_tweets = [t for t in sorted_tweets if t["likes"] >= MIN_LIKES][:TOP_N]

    if not top_tweets:
        print(f"No tweets with >= {MIN_LIKES} likes / 没有点赞数达到 {MIN_LIKES} 的推文")
        return

    print(f"Retweeting top {len(top_tweets)} tweet(s) / 转推前 {len(top_tweets)} 条...")

    for t in top_tweets:
        print(f"  [{t['likes']} likes] @{t['author_screen_name']}: {t['text'][:80]!r}")
        try:
            retweet(t["id"])
            print(f"  ✓ Retweeted / 已转推 {t['id']}")
        except Exception as e:
            print(f"  Retweet failed: {e}")
        time.sleep(DELAY_SECONDS)

    print("Done / 完成")


if __name__ == "__main__":
    main()
Equivalent curl command
# Search trending tweets via LocalBridge / 搜索热点推文（替换 %23AI 为你的话题，%23 = #）
curl -s "http://127.0.0.1:20088/api/v1/x/search?query=%23AI&count=20"

# Retweet a tweet via LocalBridge / 转推推文（替换 TWEET_ID）
curl -s -X POST "http://127.0.0.1:20088/api/v1/x/retweets" \
  -H "Content-Type: application/json" \
  -d '{"tweetId":"TWEET_ID"}'
↓ clawbot.py xdk.py http_rest.py