tukiyururu · September 12, 2024 00:54
diff --git a/twitter_media.py b/twitter_media.py
 # -*- coding: utf-8 -*-
 """
 特定ユーザのTwitterの画像・動画を全取得する
 """
 import os
 import re
 from sys import argv
 from urllib.request import urlopen
 from urllib.parse import urlparse
 from json import loads
 import tweepy
 from bs4 import BeautifulSoup
 from dotenv import load_dotenv


 def save(media_url, media):
    """ 画像・動画を保存する """
    if not os.path.exists(media):
        # メディアの取得
        with urlopen(media_url) as res:
            print('[GET] ' + media_url)

            # メディアの保存
            with open(media, 'wb') as file:
                file.write(res.read())
                print(' => ' + media)


 def get_images(name, tweet_ids, param=''):
    """ 連続して画像を保存する """
    # メディアJSONデータのURL
    url = 'https://twitter.com/i/profiles/show/%s/media_timeline%s' % (name, param)

    # メディアJSONデータの取得
    res = urlopen(url)
    json = loads(res.read())
    res.close()

    # JSONデータのHTMLからツイート情報を取得
    soup = BeautifulSoup(json['items_html'], 'html.parser')
    tweets = soup.select('.tweet')
    # ツイートID
    tweet_id = ''

    for tweet in tweets:
        num = 0
        # ツイートIDの取得
        tweet_id = tweet['data-tweet-id']

        # 動画のサムネイルがあるか
        reg = re.compile(
            r'https://pbs\.twimg\.com/ext_tw_video_thumb/[\d]+/pu/img/[\w\-]+\.(jpg|png)')
        if reg.findall(str(tweet)):
            # 動画のあるツイートIDを格納
            tweet_ids.append(tweet_id)

        # 画像情報の取得
        photos = tweet.select('.AdaptiveMedia-photoContainer')

        for photo in photos:
            num += 1
            # 画像URL
            img_url = photo['data-image-url']
            # 拡張子
            ext = os.path.splitext(img_url)[1]

            # ファイル名: {スクリーンネーム}_{ツイートID}_{番号}{拡張子}
            img = '%s_%s_%d%s' % (name, tweet_id, num, ext)

            save(img_url + ':orig', img)

    # 次ページがあれば処理を続行
    if json['has_more_items']:
        # 最後のツイートIDから -1
        max_id = int(tweet_id) - 1
        # パラメータを渡して再帰呼出し
        get_images(name, tweet_ids, '?last_note_ts=%s&max_position=%s' % (tweet_id, max_id))


 def get_videos(api, name, tweet_ids):
    """ 連続して動画を保存する """
    for i in range(0, len(tweet_ids), 100):
        # 100件ずつツイートのJSONデータを取得
        statuses = api.statuses_lookup(tweet_ids[i:i + 100])

        for status in statuses:
            # 動画情報の取得
            video_info = status.extended_entities['media'][0]['video_info']['variants']
            # 動画情報をbitrateの値で降順ソート
            info = sorted(video_info, key=lambda x: 'bitrate' in x and -x['bitrate'])
            # 動画URL
            video_url = info[0]['url']
            # 動画URLのpathを取得
            path = urlparse(video_url).path

            # 拡張子
            ext = os.path.splitext(path)[1]
            # ファイル名: {スクリーンネーム}_{ツイートID}{拡張子}
            video = '%s_%s%s' % (name, status.id_str, ext)

            save(video_url, video)


 def twitter_media(name, env):
    """ 画像・動画を全取得する """
    # 動画を含むツイートを格納する配列
    tweet_ids = []

    get_images(name, tweet_ids)

    # .envファイルの取得
    load_dotenv(env)

    # コンシューマキーやアクセストークンの取得
    consumer_key = os.environ.get('CONSUMER_KEY')
    consumer_secret = os.environ.get('CONSUMER_SECRET')
    access_token = os.environ.get('ACCESS_TOKEN')
    access_token_secret = os.environ.get('ACCESS_TOKEN_SECRET')

    # OAuth認証
    auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)
    api = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True)

    get_videos(api, name, tweet_ids)


 def main():
    """ メイン処理 """
    # スクリーンネーム
    name = argv[1]
    # 実行スクリプトのディレクトリ
    work_dir = os.path.dirname(os.path.abspath(__file__))
    # 保存ディレクトリ
    media_dir = work_dir + '/media/' + name
    # .envファイル
    env = work_dir + '/.env'

    # 保存ディレクトリの作成
    if not os.path.exists(media_dir):
        os.makedirs(media_dir)
    # 保存ディレクトリに移動
    os.chdir(media_dir)

    twitter_media(name, env)


 if __name__ == '__main__':
    main()
	# -- coding: utf-8 --
	"""
	特定ユーザのTwitterの画像・動画を全取得する
	"""
	import os
	import re
	from sys import argv
	from urllib.request import urlopen
	from urllib.parse import urlparse
	from json import loads
	import tweepy
	from bs4 import BeautifulSoup
	from dotenv import load_dotenv


	def save(media_url, media):
	""" 画像・動画を保存する """
	if not os.path.exists(media):
	# メディアの取得
	with urlopen(media_url) as res:
	print('[GET] ' + media_url)

	# メディアの保存
	with open(media, 'wb') as file:
	file.write(res.read())
	print(' => ' + media)


	def get_images(name, tweet_ids, param=''):
	""" 連続して画像を保存する """
	# メディアJSONデータのURL
	url = 'https://twitter.com/i/profiles/show/%s/media_timeline%s' % (name, param)

	# メディアJSONデータの取得
	res = urlopen(url)
	json = loads(res.read())
	res.close()

	# JSONデータのHTMLからツイート情報を取得
	soup = BeautifulSoup(json['items_html'], 'html.parser')
	tweets = soup.select('.tweet')
	# ツイートID
	tweet_id = ''

	for tweet in tweets:
	num = 0
	# ツイートIDの取得
	tweet_id = tweet['data-tweet-id']

	# 動画のサムネイルがあるか
	reg = re.compile(
	r'https://pbs\.twimg\.com/ext_tw_video_thumb/[\d]+/pu/img/[\w\-]+\.(jpg\|png)')
	if reg.findall(str(tweet)):
	# 動画のあるツイートIDを格納
	tweet_ids.append(tweet_id)

	# 画像情報の取得
	photos = tweet.select('.AdaptiveMedia-photoContainer')

	for photo in photos:
	num += 1
	# 画像URL
	img_url = photo['data-image-url']
	# 拡張子
	ext = os.path.splitext(img_url)[1]

	# ファイル名: {スクリーンネーム}_{ツイートID}_{番号}{拡張子}
	img = '%s_%s_%d%s' % (name, tweet_id, num, ext)

	save(img_url + ':orig', img)

	# 次ページがあれば処理を続行
	if json['has_more_items']:
	# 最後のツイートIDから -1
	max_id = int(tweet_id) - 1
	# パラメータを渡して再帰呼出し
	get_images(name, tweet_ids, '?last_note_ts=%s&max_position=%s' % (tweet_id, max_id))


	def get_videos(api, name, tweet_ids):
	""" 連続して動画を保存する """
	for i in range(0, len(tweet_ids), 100):
	# 100件ずつツイートのJSONデータを取得
	statuses = api.statuses_lookup(tweet_ids[i:i + 100])

	for status in statuses:
	# 動画情報の取得
	video_info = status.extended_entities['media'][0]['video_info']['variants']
	# 動画情報をbitrateの値で降順ソート
	info = sorted(video_info, key=lambda x: 'bitrate' in x and -x['bitrate'])
	# 動画URL
	video_url = info[0]['url']
	# 動画URLのpathを取得
	path = urlparse(video_url).path

	# 拡張子
	ext = os.path.splitext(path)[1]
	# ファイル名: {スクリーンネーム}_{ツイートID}{拡張子}
	video = '%s_%s%s' % (name, status.id_str, ext)

	save(video_url, video)


	def twitter_media(name, env):
	""" 画像・動画を全取得する """
	# 動画を含むツイートを格納する配列
	tweet_ids = []

	get_images(name, tweet_ids)

	# .envファイルの取得
	load_dotenv(env)

	# コンシューマキーやアクセストークンの取得
	consumer_key = os.environ.get('CONSUMER_KEY')
	consumer_secret = os.environ.get('CONSUMER_SECRET')
	access_token = os.environ.get('ACCESS_TOKEN')
	access_token_secret = os.environ.get('ACCESS_TOKEN_SECRET')

	# OAuth認証
	auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
	auth.set_access_token(access_token, access_token_secret)
	api = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True)

	get_videos(api, name, tweet_ids)


	def main():
	""" メイン処理 """
	# スクリーンネーム
	name = argv[1]
	# 実行スクリプトのディレクトリ
	work_dir = os.path.dirname(os.path.abspath(__file__))
	# 保存ディレクトリ
	media_dir = work_dir + '/media/' + name
	# .envファイル
	env = work_dir + '/.env'

	# 保存ディレクトリの作成
	if not os.path.exists(media_dir):
	os.makedirs(media_dir)
	# 保存ディレクトリに移動
	os.chdir(media_dir)

	twitter_media(name, env)


	if __name__ == '__main__':
	main()