Pythonクローリング＆スクレイピング［増補改訂版］ ―データ収集・解析のための実践開発ガイドーその11

第5章クローリング・スクレイピングの実践とデータの活用

Twitter Streaming API

公開されているツイートを取得します。

import os
import tweepy

# 環境変数から認証情報を取得する
TWITTER_API_KEY = os.environ['TWITTER_API_KEY']
TWITTER_API_SECRET_KEY = os.environ['TWITTER_API_SECRET_KEY']
TWITTER_ACCESS_TOKEN = os.environ['TWITTER_ACCESS_TOKEN']
TWITTER_ACCESS_TOKEN_SECRET = os.environ['TWITTER_ACCESS_TOKEN_SECRET']

def main():

  auth = tweepy.OAuthHandler(TWITTER_API_KEY, TWITTER_API_SECRET_KEY)
  auth.set_access_token(TWITTER_ACCESS_TOKEN, TWITTER_ACCESS_TOKEN_SECRET)

  stream = tweepy.Stream(auth, MystreamListener())
  stream.sample(languages=['ja'])


class MystreamListener(tweepy.StreamListener):
  """
  Streaming API で取得したツイートを処理
  """

  def on_status(self, status: tweepy.Status):
    """
    ツイートを受信したときに呼び出されるメソッド
    """
    print('@' + status.author.screen_name, status.text)


if __name__=='__main__':
  main()

Amazon API

API のキーを取得したんですが、なぜかエラーになるので試せませんでした。

YouTube API

Google API の中の YouTube API を使います。

import os
from apiclient.discovery import build

YOUTUBE_API_KEY = os.environ['YOUTUBE_API_KEY']

youtube = build('youtube', 'v3', developerKey=YOUTUBE_API_KEY)

search_response = youtube.search().list(
  part='snippet',
  q='Python',
  type='video',
).execute()

for item in search_response['items']:
  print(item['snippet']['title'])

f:id:yamakasa3:20190905215258p:plain

MongoDB に YouTube の動画のメタ情報を保存

プログラミングというクエリで YouYube API を使い、視聴回数の上位 10 件までを表示します。

import os
import logging
from typing import Iterator, List

from apiclient.discovery import build
from pymongo import MongoClient, ReplaceOne, DESCENDING
from pymongo.collection import Collection

YOUTUBE_API_KEY = os.environ['YOUTUBE_API_KEY']
logging.getLogger('googleapiclient.discovery_cache').setLevel(logging.ERROR)


def main():
    """
    MongoDB に YouTube データを保存する
    """
    mongo_client = MongoClient('localhost', 27017)
    collection = mongo_client.youtube.videos

    # 動画を検索してアイテムを保存
    for item_per_page in search_videos('プログラミング'):
        save_to_mongodb(collection, item_per_page)

    show_top_videos(collection)


def search_videos(query: str, max_pages: int = 5) -> Iterator[List[dict]]:
    """
    query で検索して、アイテムのリストを返す
    max_page ページまで検索を行う
    """
    youtube = build('youtube', 'v3', developerKey=YOUTUBE_API_KEY)
    search_request = youtube.search().list(
        part='id',
        q=query,
        type='video',
        maxResults=50,
    )

    i = 0
    while search_request and i < max_pages:
        search_response = search_request.execute()
        # for item in search_response['items']:
        #     print(item['id']['videoId'])
        
        video_ids = [item['id']['videoId'] for item in search_response['items']]
        
        videos_response = youtube.videos().list(
            part='snippet,statistics',
            id=','.join(video_ids)
        ).execute()
        
        yield videos_response['items']
        search_request = youtube.search().list_next(search_request, search_response)
        i += 1
    

def save_to_mongodb(collection: Collection, items: List[dict]):
    """
    MongoDB に保存
    """
    for item in items:
        item['_id'] = item['id']  # id 属性を _id とする
        for key, value in item['statistics'].items():
            item['statistics'][key] = int(value)

    operations = [ReplaceOne({'_id': item['_id']},
                             item, upsert=True) for item in items]
    result = collection.bulk_write(operations)
    logging.info(f'Upserted {result.upserted_count} documents.')


def show_top_videos(collection: Collection):
    """
    MongoDB のコレクション内で視聴数の上位10を表示する
    """
    for item in collection.find().sort('statistics.viewCount', DESCENDING).limit(10):
        print(item['statistics']['viewCount'], item['snippet']['title'])

if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    main()