メディアツイートのみ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()