YOTUBE Data API V3を使ってみた

#概要
スクレイピング入門書でyoutube Data API V3を使ってみた。
色々と詰まりまくったのでその証跡を残しておく。

##実装したコード

youtube_api_metadata.py

import os
import logging
import csv

from apiclient.discovery import build
from pymongo import MongoClient, ReplaceOne, DESCENDING
from typing import Iterator, List

from pymongo.collection import Collection

YOUTUBE_API_KEY=os.environ['YOUTUBE_API_KEY']

logging.getLogger('googleapiclient.discovery_cache').setLevel(logging.ERROR)


def main():

    mongo_client = MongoClient('localhost', 27017)
    collections = mongo_client.youtube.videos

    query = input('検索値を指定して下さい。:')

    for items_per_page in  search_videos(query):
        save_to_momgo(collections, items_per_page)

    save_to_csv(collections)





def search_videos(query: str, max_pages: int=5):

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

    search_request = youtube.search().list(
        part='id',
        q=query,
        type='video',
        maxResults=15,
    )

    i = 0
    while search_request and i < max_pages:
        search_response = search_request.execute()

        video_ids = [item['id']['videoId'] for item in search_response['items']]
        video_response = youtube.videos().list(
            part='snippet,statistics',
            id=','.join(video_ids),
        ).execute()

        yield video_response['items']

        search_requst = youtube.search().list_next(search_request, search_response)
        i += 1


def save_to_momgo(collection: Collection, items: List[dict]):

    for item in items:
        item['_id'] = item['id']


        for key, value in item['statistics'].items():
            item['statistics'][key] = int(value)

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




def save_to_csv(collection: Collection):

    with open('top_videos_list.csv', 'w',newline='', encoding='utf-8-sig') as f:
        writer = csv.DictWriter(f, ['title', 'viewCount'])
        writer.writeheader()
        for item in collection.find().sort('statistics.viewCount', DESCENDING):
            writer.writerows([{'title' : item['snippet']['title'], 'viewCount': item['statistics']['viewCount']}])






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

##詰まり①

コードを実装し、何度か実行していたら以下エラーが発生。

sample.py

googleapiclient.errors.HttpError: <HttpError 403 when requesting https://www.googleapis.com/youtube/v3/videos?part=snippet%2Cstatistics&id=wXeR58bjCak%2CujxXyCrDnU0%2CQhhUVI0sxCc%2CKZz-7KSMjZA%2CWq-WeVQoE6U%2CG-qWwfG9mBE%2CYqUwEPSZQGQ%2CIuopzT_TWPQ%2CmcFspy1WhL8%2Ck9dcl7F6IFY%2C--Z5cvZ4JEw%2C3hidJgc9Zyw%2CdYSmEkcM_8s%2Ch6Hc4RuK8D8%2CRQfN2re3u4w&key=<YOUTUBE_API_KEY>&alt=json returned "The request cannot be completed because you have exceeded your <a href="/youtube/v3/getting-started#quota">quota</a>.">

最初はよくわからず、何が原因かを探しまくった。

結果、原因は単純に"quota"の使用制限がかかったのだと思われる。
GCPから新しいプロジェックトを作成し、認証タブからAPIキーを新規に発行。
キーを以下で環境変数に設定し、再度pythonファイルを実行。

set YOUTUBE_API_KEY=<YOUTUBE_API_KEY>

結果,問題なく実行されCSVファイルが出力された。

(scraping3.7) C:\Users\user\scraping3.7\files>python  save_youtube_videos_matadata.py
検索値を指定して下さい。:ひかきん
INFO:root:Upserted 15 documents.
INFO:root:Upserted 0 documents.
INFO:root:Upserted 0 documents.
INFO:root:Upserted 0 documents.
INFO:root:Upserted 0 documents.

##詰まり②
上記コードの内、以下が機能があまりよくわからず。
これが気になってしょうがなくなり開発がストップ。

operation = [ReplaceOne({'_id': item['_id']}, item, upsert=True) for item in items]

公式で調べまくった結果,以下のフォーマットで引数を受け取る事がわかった。
公式

ReplaceOne(filter, replacement, options)

以下の場合

ReplaceOne({'city': '東京'}, {'city':'群馬'}, upsert=True)

'city':'東京'のデータが存在する場合、city':'群馬'に更新。
'city':'東京'が存在しない場合、新たに'city':'群馬'を挿入。

##参考書
amazon
Pythonクローリング&スクレイピング[増補改訂版] -データ収集・解析のための実践開発ガイド