""" CONFIDENTIAL. Copyright 2020 Robots & Humans. For review by the YouTube Data Api Team. Contact: joel@whtlst.in We are a record label. The following is the script we are using to find channels that for niche topics that might want to user our artists' music. The calls are highlighted with a comment containing APICALL, if you want to do a search for those. The output of this script is a spreadsheet that we use to create contact lists to contact the channel owners. This is a simple script, but since it uses the search endpoint, it can use a lot of quota. The example of what we search for is at the bottom, it's for 'reaction' style channels. """ import pandas as pd import numpy as np import re import os from datetime import datetime from apiclient.discovery import build import googleapiclient.errors DEVELOPER_KEY = os.environ["GOOGLE_API_KEY"] YOUTUBE_API_SERVICE_NAME = "youtube" YOUTUBE_API_VERSION = "v3" def chunks(l, n): """Yield successive n-sized chunks from l.""" for i in range(0, len(l), n): yield l[i : i + n] def setup_youtube(): return build( YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION, developerKey=DEVELOPER_KEY, cache_discovery=False, ) DATA_FOLDER = os.environ["DATA_FOLDER"] musicTopicId = "/m/04rlf" def search_and_get_video_details( search_query, num_results, topicId=None, order=None, publishedAfter=None, request_size=50, ): yt = setup_youtube() max_size = request_size request_result_sizes = ( # eg, [50, 50, 15] if num_results is 115 [max_size] * (num_results // max_size) ) + ([num_results % max_size] if num_results % max_size != 0 else []) all_ids = [] all_items = [] pageToken = None vid_parts = [ "id", "snippet", "statistics", ] for i, maxResults in enumerate(request_result_sizes): # APICALL: search for keywords 'search_query' search_results = ( yt.search() .list( type="video", q=search_query, part="snippet", maxResults=maxResults, topicId=topicId, order=order, publishedAfter=publishedAfter, pageToken=pageToken, ) .execute() ) these_ids = [v["id"]["videoId"] for v in search_results["items"]] all_ids.extend(these_ids) # APICALL: get video details from 50 video id: video_results = ( yt.videos().list(id=",".join(these_ids), part=",".join(vid_parts)).execute() ) all_items.extend(video_results["items"]) pageToken = search_results.get("nextPageToken") if pageToken is None and i < len(request_result_sizes) + 1: print(f"No more results found, total len{all_items}") break return all_items def visit_channel_stats( channel_ids: list, each_stat_visitor: callable, not_found_visitor: callable, part="statistics", ): # loop through channel ids in chunkc of 50 for channel_ids_chunk in chunks(channel_ids, 50): ids_arg = ",".join(channel_ids_chunk) max_results = len(channel_ids_chunk) yt = setup_youtube() found_ids = set() # APICALL: Get channel results for 50 channel ids: results = ( yt.channels().list(id=ids_arg, part=part, maxResults=max_results).execute() ) items = results.get("items") or [] for item in items: id = item["id"] found_ids.add(id) each_stat_visitor(item) for id in channel_ids_chunk: if id not in found_ids: not_found_visitor(id) def get_video_search_data_frame( search_query, num_videos, topicId, order, publishedAfter, ): videos = search_and_get_video_details( search_query, num_results=num_videos, topicId=topicId, order=order, publishedAfter=publishedAfter, ) vid_columns = [ "id", "snippet.publishedAt", "snippet.channelId", "snippet.title", "snippet.description", "statistics.viewCount", "statistics.likeCount", "statistics.dislikeCount", "statistics.favoriteCount", "statistics.commentCount", ] viddf = pd.json_normalize(videos)[vid_columns] return viddf def get_channel_dataframe(uniq_channel_ids): collected_channels = [] def add_channel(d): collected_channels.append(d) def not_found(d): print("Not found", d) visit_channel_stats( uniq_channel_ids, add_channel, not_found, part="statistics,snippet" ) channel_cols = [ "id", "snippet.title", "snippet.description", "snippet.country", "statistics.viewCount", "statistics.commentCount", "statistics.subscriberCount", "statistics.videoCount", ] channelinfo = pd.json_normalize(collected_channels)[channel_cols].set_index("id") channelinfo = channelinfo.rename( columns={c: "ch." + c.split(".")[-1] for c in channelinfo.columns} ) return channelinfo def get_merged_dataframe(search_query, num_videos, topicId, order, publishedAfter): dfcols = get_video_search_data_frame( search_query, num_videos=num_videos, topicId=topicId, order=order, publishedAfter=publishedAfter, ) uniq_channel_ids = list(set(dfcols["snippet.channelId"])) chandf = get_channel_dataframe(uniq_channel_ids) dfMerge = pd.merge( dfcols, chandf, left_on="snippet.channelId", right_index=True ).replace({np.nan: None}) return dfMerge def augment_merged_dataframe(search_query, dfMerge): dfMerge["Search Term"] = search_query dfMerge["Channel Link"] = dfMerge["snippet.channelId"].apply( lambda x: f"https://www.youtube.com/channel/{x}" ) dfMerge["Video Link"] = dfMerge.id.apply( lambda x: f"https://www.youtube.com/watch?v={x}" ) dfMerge["Channel Email"] = dfMerge["ch.description"].apply(get_email) dfMerge["Video Email"] = dfMerge["snippet.description"].apply(get_email) dfMerge["Channel Insta"] = dfMerge["ch.description"].apply(get_insta) dfMerge["Video Insta"] = dfMerge["snippet.description"].apply(get_insta) dfMerge = dfMerge.rename(columns={k: v for k, v in final_column_settings})[ [c[1] for c in final_column_settings] ] return dfMerge def make_safe_filename(s): def safe_char(c): if c.isalnum(): return c else: return "_" return "".join(safe_char(c) for c in s).rstrip("_") final_column_settings = [ ("Search Term", "Search Term"), ("ch.title", "Channel Title"), ("Channel Link", "Channel Link"), ("Channel Email", "Channel Email"), ("Channel Insta", "Channel Insta"), ("Video Email", "Video Email"), ("Video Insta", "Video Insta"), ("ch.description", "Channel Description"), ("ch.subscriberCount", "Subscribers"), ("ch.viewCount", "Channel Views"), ("ch.videoCount", "Total Videos"), ("ch.country", "Channel Country"), ("Video Link", "Video Link"), ("snippet.title", "Video Title"), ("snippet.description", "Video Description"), ("snippet.publishedAt", "Video Published"), ("statistics.viewCount", "Video Views"), ("statistics.likeCount", "Video Likes"), ("statistics.commentCount", "Video Comments"), ] def get_email(s): m = next(re.finditer(r"[a-z0-9\.\-+_]+@[a-z0-9\.\-+_]+\.[a-z]+", s.lower()), None) return m.group(0) if m else None def get_insta(s): m = next(re.finditer(r"instagram.com\/([a-z0-9\.\-_]+)/", s.lower()), None) return ("@" + m.group(1)) if m else None def create_csv_from_search( search_query, topicId=None, order="viewCount", publishedAfter="2020-01-01T00:00:00Z", num_videos=100, folder=DATA_FOLDER, ): dfMerge = get_merged_dataframe( search_query, num_videos=num_videos, topicId=topicId, order=order, publishedAfter=publishedAfter, ) dfMerge = augment_merged_dataframe(search_query, dfMerge) ts = int(datetime.today().timestamp()) from pathlib import Path Path(folder).mkdir(parents=True, exist_ok=True) filename = "vid_" + make_safe_filename(search_query) + str(ts) + ".csv" dfMerge.to_csv(os.path.join(folder, filename)) return dfMerge if __name__ == '__main__': reaction_searches = [ dict(search_query='new song reviews', topicId=musicTopicId, num_videos=50), dict(search_query='song reaction', topicId=musicTopicId), dict(search_query='music reaction', topicId=musicTopicId), dict(search_query='reaction videos', topicId=musicTopicId), dict(search_query='new pop reactions', topicId=musicTopicId), dict(search_query='new release reaction', topicId=musicTopicId), ] reaction_folder = os.path.join(DATA_FOLDER, 'reactions') for search in reaction_searches: kwargs = dict( publishedAfter='2020-01-01T00:00:00Z', order='viewCount', num_videos=300, topicId=None, folder=reaction_folder ) kwargs.update(search) create_csv_from_search(**kwargs)