メディアツイートのみ5件取得する

・今回はPythonで取得してみる

・ツイートの取得はできた

・Conoha WingはWAFをオフにしても外部からデータベースにアクセスすることができない

import tweepy
import os
import time
from typing import List, Dict
import mysql.connector


# wp-config.php
BEARER_TOKEN = "べあらーとーくん"
DB_HOST = "mysql73.XXXXX.ne.jp"
DB_USER = "12bpr_5d7s75j3"
DB_PASS = "パスワード"
DB_NAME = "12bpr_x78df545"

# データベースに接続する
# ConohaWingのサーバー上で実行した場合のみデータベースへ接続可能
# WAFをオフにしても外部からの接続はできない
def connect_to_wordpress_db():
    try:
        return mysql.connector.connect(
            host=DB_HOST,
            port=3306,
            user=DB_USER,
            password=DB_PASS,
            database=DB_NAME,
            connection_timeout=20,
            ssl_disabled=True,
            pool_name=None
        )
    except mysql.connector.Error as err:
        print(f"接続エラー: {err}")


# wp_twitterテーブルに埋め込みコードを保存する
def save_twitter_embed(cursor, username: str, tweet_id: str, embed_code: str):
    query = """
    INSERT INTO wp_xxx
    (username, tweet_id, embed_code, created_at)
    VALUES (%s, %s, %s, NOW())
    ON DUPLICATE KEY UPDATE
    embed_code = %s, updated_at = NOW()
    """
    cursor.execute(query, (username, tweet_id, embed_code, embed_code))


# 指定されたユーザーのメディアツイートを5件取得する
# 返り値はList[Dict]: メディアを含むツイートのリスト
def get_media_tweets(username: str, max_tweets: int = 5, max_retries: int = 3) -> List[Dict]:
    for attempt in range(max_retries):
        try:
            # Twitter APIクライアントの初期化
            client = tweepy.Client(bearer_token=BEARER_TOKEN)

            # ユーザーIDを取得
            user_response = client.get_user(username=username)
            user_id = user_response.data.id
        
            # メディアツイートを取得
            media_tweets = client.get_users_tweets(
                id=user_id, 
                max_results=max_tweets,
                tweet_fields=['created_at', 'attachments', 'text'],
                media_fields=['url', 'type'],
                expansions=['attachments.media_keys']
            )

            # メディア情報を処理
            media_map = {}
            if media_tweets.includes and 'media' in media_tweets.includes:
                media_map = {
                    media.media_key: media 
                    for media in media_tweets.includes['media']
                }

            # メディアツイートを整形
            processed_tweets = []
            for tweet in media_tweets.data or []:
                media_info = []
                if tweet.attachments and tweet.attachments.get('media_keys'):
                    for media_key in tweet.attachments['media_keys']:
                        media = media_map.get(media_key)
                        if media and media.type == 'photo':
                            # 埋め込みコードを生成
                            embed_code = f'<blockquote class="twiiter-tweet">' \
                                         f'<p lang="ja" dir="ltr">{tweet.text}</p>' \
                                        f'— {username} (@{username}) ' \
                                        f'<a href="https://twitter.com/{username}/status/{tweet.id}">' \
                                        f'Tweet URL</a>' \
                                        f'</blockquote>'

                            media_info.append({
                                'media_url': media.url,
                                'tweet_text': tweet.text,
                                'embed_code': embed_code,
                                'tweet_id': tweet.id
                            })

                if media_info:
                    processed_tweets.append({
                        'tweet_id': tweet.id,
                        'created_at': tweet.created_at,
                        'media': media_info
                    })
        
            return processed_tweets

        except tweepy.TweepyException as e:
            # 429 Too Many Requestの場合は待つ
            if e.response and e.response.status.code == 429:
                wait_time = (2 ** attempt) * 15
                print("429のため" + wait_time + "秒間お待ちください")
                time.sleep(wait_time)
            else:
                print(f"Twitter API エラー: {e}")
                return []
        except Exception as e:
            print(f"予期せぬエラー: {e}")
            return []

def main():
    username = 'example_user'
    
    # データベース接続
    db_connection = connect_to_wordpress_db()
    cursor = db_connection.cursor()
    
    try:
        # メディアツイート取得
        media_tweets = get_media_tweets(username)
        
        # 各ツイートをデータベースに保存
        for tweet in media_tweets:
            for media in tweet['media']:
                save_twitter_embed(
                    cursor, 
                    username, 
                    media['tweet_id'], 
                    media['embed_code']
                )
        
        # 変更をコミット
        db_connection.commit()
        print("ツイート埋め込みコードを正常に保存しました。")
    
    except Exception as e:
        db_connection.rollback()
        print(f"エラーが発生しました: {e}")
    
    finally:
        cursor.close()
        db_connection.close()

if __name__ == "__main__":
    main()