[PYTHON] J'ai essayé d'utiliser l'API de données YOUTUBE V3

Aperçu

J'ai essayé d'utiliser YouTube Data API V3 dans le Scraping Primer. Je suis coincé de différentes manières, alors je vais laisser une trace.

Code implémenté

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('Veuillez spécifier la valeur de recherche.:')

    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()

Sabot ①

Après avoir implémenté le code et exécuté plusieurs fois, l'erreur suivante se produit.

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>.">

Au début, je n'étais pas sûr, alors j'ai cherché quelle en était la cause.

En conséquence, il semble que la cause était simplement la restriction à l'utilisation du "contingent". Créez un nouveau projet à partir de GCP et émettez une nouvelle clé API à partir de l'onglet Authentification. Définissez la clé sur la variable d'environnement ci-dessous et exécutez à nouveau le fichier python.

set YOUTUBE_API_KEY=<YOUTUBE_API_KEY>

En conséquence, il a été exécuté sans aucun problème et un fichier CSV a été généré.

(scraping3.7) C:\Users\user\scraping3.7\files>python  save_youtube_videos_matadata.py
Veuillez spécifier la valeur de recherche.:Hikakin
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.

Sabot ②

Parmi le code ci-dessus, les éléments suivants ne sont pas très clairs. Je n'ai pas pu m'empêcher d'arrêter le développement.

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

À la suite de recherches officielles, il a été constaté que l'argument est reçu dans le format suivant. Officiel

ReplaceOne(filter, replacement, options)

Dans les cas suivants

ReplaceOne({'city': 'Tokyo'}, {'city':'Gunma'}, upsert=True)

Si les données de «ville»: «Tokyo» existent, mettez à jour la ville «:« Gunma ». Si «ville»: «Tokyo» n'existe pas, insérez une nouvelle «ville»: «Gunma».

Livre de référence

amazon Python Crawling & Scraping [Édition augmentée et révisée] -Guide de développement pratique pour la collecte et l'analyse de données