Skip to content

Instantly share code, notes, and snippets.

@Ryomasao Ryomasao/Twitter.py
Created Feb 23, 2017

Embed
What would you like to do?
ツイッターの情報を取得する
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from requests_oauthlib import OAuth1Session
import json
import requests
#認証情報
outh_key_dict = {
'consumer_key': 'JQW4vn3OFHmorXrf0KQziSrC3',
'consumer_secret': 'kGXQF7iTafgoeAvcxhugHCMx6W9plixlTS7ZeBjNg1LrXcsg5l',
'access_token': '3326721211-XPqRiB1yWMyQCHcBqSGptEyqcFpd1JPmAJD1Itr',
'access_token_secret': '8jrDbrf8xNBHQYZhPSVRuhF68HJYKrq0ID5lEgY4XDUqM'
}
def create_oauth_session(oauth_dict):
'''
Twitterの認証を行い、セッションを返す。
:param oauth_dict:
:return:
'''
oauth = OAuth1Session(
oauth_dict['consumer_key'],
oauth_dict['consumer_secret'],
oauth_dict['access_token'],
oauth_dict['access_token_secret'],
)
return oauth
def get_json_data_byHashtag(hash_tag):
'''
hash_tagを検索キーとして、ツイートの情報を取得します。
取得した結果は、jsonデータをpythonオブジェクトにして返却します。
:param hash_tag:
:return:
'''
# ツイート取得用のURL
url = "https://api.twitter.com/1.1/search/tweets.json?"
#リクエストパラメータ
params = {
'q': str(hash_tag),
'count': '100',
}
#認証
twitter = create_oauth_session(outh_key_dict)
#ツイートを取得する。
req = twitter.get(url, params=params)
#レスポンス確認
if req.status_code != 200:
print("Error:%d" % req.status_code)
return None
responses = json.loads(req.text)
#DEBUG用にレスポンス内容を書き出しておく
file = open('jsonTweet.txt', 'w')
file.write(json.dumps(responses, indent=4))
file.close()
return responses
def get_contributor_information(responses):
'''
ツイッターAPIのレスポンス情報から、ツイート内に含まれるmedia情報、および誰がツイートしたのかを取得する。
:param responses:
:return: 投稿者の情報をリスト形式で返します。
投稿者の情報は以下の構造で保持する。
[
{'screen_name':'TwitterのID',
'id':'投稿したTweetのID?',
'media_url':'画像がある場合は、画像のurl、ない場合は空白'
}
]
'''
#投稿者の情報
contributor_inf_list = []
for data in responses['statuses']:
screen_name = data['user']['screen_name']
id = data['id']
#写真がない場合はmedia_urlsはスペース
if 'media' not in data['entities']:
media_urls = ''
else:
urls = data['entities']['media']
#写真が複数添付されている場合、url[1]とかになるかも、検証してない
media_urls = urls[0]['media_url']
#投稿者情報を作成する
contributor_inf_list.append(
{
'screen_name': screen_name,
'id': id,
'media_urls': media_urls
}
)
return contributor_inf_list
def download_media(id, media_urls):
'''
urlsのファイルをダウンロードして、端末に保存します。
:param id: ダウンロードした画像を保存する際のファイル名。ここではTwitterの投稿のidを想定。
:param media_urls: ダウンロード対象のファイルのURL
:return: なし
'''
downloads = requests.get(media_urls).content
#ファイル名つけて保存する
filepath = '%s.jpg' % id
images = open(filepath, 'wb')
images.write(downloads)
images.close()
if __name__ == '__main__':
#Twitterでハッシュタグ検索
responses = get_json_data_byHashtag('#faceRecTest')
#レスポンスがない場合は、終了する
if responses is None:
print('Responses is None')
exit
#レスポンス内容をもとに、投稿者情報を取得する。
contributor_inf_list = get_contributor_information(responses)
#投稿者情報のから画像のURLを取得し、ダウンロードして保存する。
for contributor in contributor_inf_list:
#media_urlsが空白の投稿者については除外する。
if contributor['media_urls'] != '':
download_media(contributor['id'], contributor['media_urls'])
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.