X(Twitter)の投稿・引用RTをAIで自動化する方法
Python×API実践ガイド

# search_posts.py

import requests
import json
import random

# 認証情報を読み込み
with open('credentials.json') as f:
    creds = json.load(f)

BEARER_TOKEN = creds['bearer_token']

# 検索テーマ（ランダムに選択）
SEARCH_THEMES = [
    {
        'name': 'AI自動化',
        'query': '(AI 自動化 OR AI 業務効率化) lang:ja'
                 ' -is:retweet min_faves:10',
    },
    {
        'name': 'AIツール',
        'query': '(Claude OR ChatGPT OR Cursor) (便利 OR すごい'
                 ' OR 使ってみた) lang:ja -is:retweet min_faves:5',
    },
    {
        'name': '海外AI活用',
        'query': '(AI agent OR AI automation OR no-code AI)'
                 ' -is:retweet min_faves:20',
    },
    {
        'name': 'GAS/ノーコード',
        'query': '(GAS OR Zapier OR Make) (自動化 OR 効率化)'
                 ' lang:ja -is:retweet min_faves:5',
    },
]

def search_posts(theme=None):
    """テーマに基づいてポストを検索"""
    if theme is None:
        theme = random.choice(SEARCH_THEMES)

    url = 'https://api.twitter.com/2/tweets/search/recent'
    headers = {
        'Authorization': f'Bearer {BEARER_TOKEN}',
    }
    params = {
        'query': theme['query'],
        'max_results': 20,
        'tweet.fields': 'public_metrics,author_id,lang,created_at',
        'expansions': 'author_id',
        'user.fields': 'username,name,public_metrics',
    }

    res = requests.get(url, headers=headers, params=params)
    if res.status_code != 200:
        raise Exception(f'Search API error: {res.status_code}')

    data = res.json()
    tweets = data.get('data', [])

    # エンゲージメント順にソート
    tweets.sort(
        key=lambda t: (
            t['public_metrics']['like_count']
            + t['public_metrics']['retweet_count'] * 3
        ),
        reverse=True,
    )

    return tweets[:10], theme['name']  # 上位10件

# generate_comment.py

import anthropic

client = anthropic.Anthropic()  # ANTHROPIC_API_KEY環境変数から自動取得

SYSTEM_PROMPT = """あなたはAI活用・業務自動化の専門家です。
Xの引用RTに付けるコメントを生成してください。

ルール:
- 120文字以内（短く、インパクトのある一言）
- 元ポストの良い点を具体的に触れる
- 自分の知見や経験を1つ添える
- 絵文字は1-2個まで
- ハッシュタグは付けない
- 「素晴らしい！」「いいですね！」だけの薄いコメントは禁止
- 英語ポストの場合は日本語に翻訳してコメントする

トーンの例:
- 「これ実際に導入したけど、○○の部分が特に効いた」
- 「○○が面白い。うちでは△△に応用してる」
- 「🇺🇸→🇯🇵 要約すると○○。日本でも十分使える」
"""

def generate_comment(tweet_text: str, is_english: bool = False):
    """ポストに対する引用コメントを生成"""

    user_msg = f"""以下のポストに対する引用RTコメントを生成してください。

【ポスト内容】
{tweet_text}

{'※ 英語ポストです。日本語で翻訳要約+コメントしてください。' if is_english else ''}

120文字以内のコメントのみを出力してください。"""

    response = client.messages.create(
        model='claude-sonnet-4-20250514',
        max_tokens=200,
        system=SYSTEM_PROMPT,
        messages=[{'role': 'user', 'content': user_msg}],
    )

    return response.content[0].text.strip()

# post_quote.py

from requests_oauthlib import OAuth1Session
import json
import time
import random

with open('credentials.json') as f:
    creds = json.load(f)

oauth = OAuth1Session(
    creds['api_key'],
    client_secret=creds['api_secret'],
    resource_owner_key=creds['access_token'],
    resource_owner_secret=creds['access_token_secret'],
)

def post_quote_tweet(tweet_id: str, comment: str):
    """引用RTを投稿"""
    url = 'https://api.twitter.com/2/tweets'

    # 引用RT = コメント + 元ツイートのURL
    quote_url = f'https://twitter.com/i/status/{tweet_id}'
    payload = {
        'text': comment,
        'quote_tweet_id': tweet_id,
    }

    res = oauth.post(url, json=payload)

    if res.status_code == 201:
        data = res.json()
        print(f'✅ 引用RT成功: {data["data"]["id"]}')
        return data['data']['id']
    else:
        print(f'❌ エラー: {res.status_code} {res.text}')
        return None

def run_curation(num_quotes=2):
    """メインフロー: 検索→AI分析→引用RT"""
    tweets, theme_name = search_posts()
    posted = 0
    done_ids = load_done_ids()  # 重複防止

    for tweet in tweets:
        if posted >= num_quotes:
            break
        if tweet['id'] in done_ids:
            continue

        # 英語判定
        is_english = tweet.get('lang', 'ja') != 'ja'

        # AIコメント生成
        comment = generate_comment(
            tweet['text'], is_english
        )

        # 投稿
        result = post_quote_tweet(tweet['id'], comment)
        if result:
            save_done_id(tweet['id'])
            posted += 1

            # スパム対策: 投稿間に15-30秒のランダム間隔
            if posted < num_quotes:
                wait = random.randint(15, 30)
                print(f'⏳ {wait}秒待機...')
                time.sleep(wait)

    print(f'🎉 {theme_name}テーマで{posted}件の引用RT完了')

def load_done_ids():
    """投稿済みIDを読み込み（重複防止）"""
    try:
        with open('done_ids.txt') as f:
            return set(f.read().strip().split('\n'))
    except FileNotFoundError:
        return set()

def save_done_id(tweet_id):
    """投稿済みIDを記録"""
    with open('done_ids.txt', 'a') as f:
        f.write(tweet_id + '\n')

# auto_reply.py（一次返信のみ）

def check_and_reply_mentions():
    """自分へのメンション・リプライをチェックして一次返信"""
    url = 'https://api.twitter.com/2/users/me/mentions'
    headers = {
        'Authorization': f'Bearer {BEARER_TOKEN}',
    }
    params = {
        'max_results': 10,
        'tweet.fields': 'public_metrics,in_reply_to_user_id',
    }

    res = requests.get(url, headers=headers, params=params)
    mentions = res.json().get('data', [])

    replied_ids = load_replied_ids()

    for mention in mentions:
        if mention['id'] in replied_ids:
            continue

        # AIで返信を生成
        reply_text = generate_reply(mention['text'])

        # リプライ投稿
        payload = {
            'text': reply_text,
            'reply': {
                'in_reply_to_tweet_id': mention['id']
            },
        }
        res = oauth.post(
            'https://api.twitter.com/2/tweets',
            json=payload
        )

        if res.status_code == 201:
            save_replied_id(mention['id'])
            print(f'✅ リプライ送信: {mention["id"]}')

        # スパム対策: 間隔を空ける
        time.sleep(random.randint(20, 45))

# analyze_performance.py

def analyze_daily_performance():
    """前日の投稿パフォーマンスを分析"""
    # 自分の直近ツイートを取得
    url = 'https://api.twitter.com/2/users/me/tweets'
    params = {
        'max_results': 20,
        'tweet.fields': 'public_metrics,created_at',
    }
    res = requests.get(url, headers=headers, params=params)
    tweets = res.json().get('data', [])

    # テーマ別にスコア集計
    theme_scores = {}
    for tweet in tweets:
        metrics = tweet['public_metrics']
        score = (
            metrics['like_count'] * 1
            + metrics['retweet_count'] * 3
            + metrics['reply_count'] * 2
            + metrics['impression_count'] * 0.01
        )
        theme = detect_theme(tweet['text'])  # テーマ推定
        if theme not in theme_scores:
            theme_scores[theme] = []
        theme_scores[theme].append(score)

    # 平均スコアでランキング
    ranking = sorted(
        theme_scores.items(),
        key=lambda x: sum(x[1]) / len(x[1]),
        reverse=True,
    )

    print('📊 テーマ別パフォーマンス:')
    for theme, scores in ranking:
        avg = sum(scores) / len(scores)
        print(f'  {theme}: 平均スコア {avg:.1f} ({len(scores)}件)')