Youtube Watched History Analyzer

[Aerosmite]

Hey again! I often watch Youtube, so I made an easy script to collect interesting data from my watched history, thought it was a bit scared
I share it for you, if you want to beat my data

Features

• Allow several accounts history
  
• Return Json data in a result dir
  
• Average values of videos watched per amount of time and average video length
  
• An easy to read watched_history with only interesting data
  
• Most viewed channels, tags, topics...
  
• Custom your results files
  

Installing

• Youtube watched history data from Google
  Hide/Show

  1. Go here
     
  2. Ensure that you're login with the right account (top right)
     
  3. Click 'Select none'
     
  4. Enable Youtube
     
  5. Show Youtube options, change 'OPML (RSS)' to 'JSON'
     
  6. Click 'Next' > 'Create Archive' and download it!
     
• Your API key
  
• Youtube API module $ pip install --upgrade google-api-python-client
  
• Update the '# Init' value of the code ('API_key','files','results_dir')
  

#!/usr/bin/python #-*- coding: utf-8 -*- import os import re import json from datetime import datetime, timedelta from apiclient import discovery from apiclient.discovery import build # Init API_key = "REPLACE_ME" # List of your watch-history files (allow several accounts) files = ['/Path/to/watch-history.json','/Path/to/watch-history.json'] results_dir = '/Path/to/Results_dir' # Load only videos after this date min_date = "01/01/00" # DD/MM/YY # List of allowed results files results = {'clean_history': True, 'average': True, 'channels': True, 'topics': True, 'tags': True, 'days': True, 'months': True, 'years': True} # Average percentage of videos watched watch_percentage = 60 # % service = build("youtube", "v3",developerKey=API_key) Day = ['Monday','Tuesday','Friday','Wednesday','Thursday','Sunday','Saturday'] PT_format = re.compile(r'PT((?P<hours>\d+?)hr)?((?P<minutes>\d+?)M)?((?P<seconds>\d+?)S)?') min_date = datetime.strptime(min_date, '%d/%m/%y') watch_percentage /= 100 date_sorter = [] videos = [] average_li = [] sorted_li = [] channel_hm = {} topic_hm = {} tag_hm = {} duration_hm = {} day_name_hm = {} day_hm = {} month_hm = {} year_hm = {} # remove keyword arguments that are not set def remove_empty_kwargs(**kwargs):	good_kwargs = {}	if kwargs is not None:	for key, value in kwargs.iteritems():	if value:	good_kwargs[key] = value	return good_kwargs # sample python code for channels.list def videos_list_by_id(service, **kwargs):	kwargs = remove_empty_kwargs(**kwargs)	return service.videos().list(**kwargs).execute() def parse_time(time_str): parts = PT_format.match(time_str) if not parts: return parts = parts.groupdict() time_params = {} for (name, param) in parts.iteritems(): if param: time_params[name] = int(param) return timedelta(**time_params) def clear_videos_vars():	video_title = video_description = video_duration = video_thumbnails = video_location = video_topics = video_tags = video_date = channel_id = channel_title = None def count_data(hm_name,key):	try:	hm_name[key] = {'nb':hm_name[key]['nb'] + 1}	except:	hm_name[key] = {'nb':1} def average(dict):	nb = 0	length = 0	for key in dict:	nb += dict[key]['nb']	length += 1	return float(nb)/length def sorted_list(hm_name):	sorted_li = []	for key in hm_name:	sorted_li.append({'name':key,'nb':hm_name[key]['nb']})	return sorted(sorted_li, key=lambda k: k['nb'], reverse=True) # create new path if necessary if os.path.isdir(results_dir) == False:	os.makedirs(results_dir) # save watching date of videos for f in files:	for x in json.load(open(f)):	date = datetime.strptime(x['snippet']['publishedAt'], '%Y-%m-%dT%H:%M:%S.000Z')	delta_date = date - min_date	if delta_date.total_seconds() > 0:	date_sorter.append(date) # sort videos per watching date print(str(len(date_sorter)) + " videos detected" ) date_sorter.sort(reverse=True) for f in files:	for x in json.load(open(f)):	date = datetime.strptime(x['snippet']['publishedAt'], '%Y-%m-%dT%H:%M:%S.000Z')	delta_date = date - min_date	if delta_date.total_seconds() < 0:	continue	# define video position	pos = 0	for sorted_date in date_sorter:	if sorted_date == date:	break	else:	pos += 1	# access to video data	if x['snippet']['title'] != "Deleted video" and x['snippet']['title'] != "Private video":	try:	video_data = videos_list_by_id(service,part='snippet,contentDetails,topicDetails,recordingDetails',id=x['contentDetails']['videoId'])	try:	video_location = (video_data['items'][0]['recordingDetails']['location']['latitude'],video_data['items'][0]['recordingDetails']['location']['longitude'])	except KeyError:	video_location = None	try:	video_topics = video_data['items'][0]['topicDetails']['topicCategories']	except KeyError:	video_topics = []	try:	video_tags = video_data['items'][0]['snippet']['tags']	except KeyError:	video_tags = []	video_title = video_data['items'][0]['snippet']['title']	video_description = video_data['items'][0]['snippet']['description']	video_duration = parse_time(video_data['items'][0]['contentDetails']['duration'])	video_thumbnails = video_data['items'][0]['snippet']['thumbnails']['default']['url']	video_date = datetime.strptime(video_data['items'][0]['snippet']['publishedAt'],'%Y-%m-%dT%H:%M:%S.000Z').strftime('%d/%m/%y %H:%M')	channel_id = video_data['items'][0]['snippet']['channelId']	channel_title = video_data['items'][0]['snippet']['channelTitle']	# save video data in dicts	for topic in video_topics:	count_data(topic_hm,topic)	for tag in video_tags:	count_data(tag_hm,tag.lower())	if pos > 0 and pos < len(date_sorter):	next_date = date_sorter[pos-1]	if video_duration > next_date - date:	video_duration = next_date - date	else:	video_duration *= watch_percentage	count_data(duration_hm,video_duration)	count_data(channel_hm,channel_title)	except (KeyError,IndexError):	print("missing information for video '" + x['contentDetails']['videoId'] + "'")	clear_videos_vars()	except:	print("unknown error for video '" + x['contentDetails']['videoId'] + "'")	clear_videos_vars()	else:	print("unable to access video '" + x['contentDetails']['videoId'] + "'")	clear_videos_vars()	# save date data in dicts	count_data(year_hm,str(date.year))	count_data(month_hm,str(date.year) + "/" + str(format(date.month, '02')))	count_data(day_hm,str(date.year) + "/" + str(format(date.month, '02')) + "/" + str(format(date.day, '02')))	count_data(day_name_hm,date.strftime("%A"))	# add all data to 'clean_history'	videos.append({'pos':pos,'date':date.strftime('%d/%m/%y %H:%M'),'video':{'title':video_title,'description':video_description,'duration':str(video_duration),'thumbnails':video_thumbnails,'date':video_date,'location':video_location,'topics':video_topics,'tags':video_tags},'channel':{'title':channel_title,'id':channel_id}})	print("account " + str(json.load(open(f))[0]['snippet']['channelTitle']) + " done") # sort and save dicts in results files for list_result, value in results.iteritems():	if value:	if list_result == 'clean_history':	with open(os.path.join(results_dir,"clean_history.json"), 'w') as outfile:	json.dump(videos, outfile,indent=4)	elif list_result == 'average':	for day in Day:	try:	day_name_hm[day]['nb'] = day_name_hm[day]['nb'] / (len(day_hm) / 7.0)	except KeyError:	day_name_hm[day] = {'nb':0}	average_li.append({'videos per':{'day of the week':{Day[0]:day_name_hm[Day[0]]['nb'],Day[1]:day_name_hm[Day[1]]['nb'],Day[2]:day_name_hm[Day[2]]['nb'],Day[3]:day_name_hm[Day[3]]['nb'],Day[4]:day_name_hm[Day[4]]['nb'],Day[5]:day_name_hm[Day[5]]['nb'],Day[6]:day_name_hm[Day[6]]['nb']},'year':average(year_hm),'month':average(month_hm),'day':average(day_hm),'channel':average(channel_hm)}})	nb = dur_sum = 0	for key in duration_hm:	dur_sum += duration_hm[key]['nb'] * key.total_seconds()	nb += duration_hm[key]['nb']	average_li.append({'video length (min)':dur_sum/nb/60})	with open(os.path.join(results_dir,"average.json"), 'w') as outfile:	json.dump(average_li, outfile,indent=4)	elif list_result == 'channels':	with open(os.path.join(results_dir,"channels.json"), 'w') as outfile:	json.dump(sorted_list(channel_hm), outfile,indent=4)	elif list_result == 'topics':	with open(os.path.join(results_dir,"topics.json"), 'w') as outfile:	json.dump(sorted_list(topic_hm), outfile,indent=4)	elif list_result == 'tags':	with open(os.path.join(results_dir,"tags.json"), 'w') as outfile:	json.dump(sorted_list(tag_hm), outfile,indent=4)	elif list_result == 'days':	with open(os.path.join(results_dir,"days.json"), 'w') as outfile:	json.dump(sorted_list(day_hm), outfile,indent=4)	elif list_result == 'months':	with open(os.path.join(results_dir,"months.json"), 'w') as outfile:	json.dump(sorted_list(month_hm), outfile,indent=4)	elif list_result == 'years':	with open(os.path.join(results_dir,"years.json"), 'w') as outfile:	json.dump(sorted_list(year_hm), outfile,indent=4)

---

My average.json, if you try to beat me :P

Output:
[ { "videos per": { "month": 200.36363636363637, "year": 1102.0, "day": 15.412587412587413, "channel": 2.287974683544304, "day of the week": { "Monday": 17.132867132867133, "Tuesday": 18.552447552447553, "Friday": 14.734265734265735, "Wednesday": 16.88811188811189, "Thursday": 9.447552447552448, "Sunday": 13.951048951048952, "Saturday": 17.181818181818183 } } }, { "video length (min)": 2.282149992315967 } ]

[Aerosmite]

New update!

Changelog
 

• Add country with most viewed videos
  
• Add most viewed youtube category
  
• Support of video with new id
  
• Optimization and more self-explanatory code
  
• Handle more specific exceptions
  
• 'video_duration' bug fixed
  

#!/usr/bin/python #-*- coding: utf-8 -*- import os import re import json import operator from datetime import datetime, timedelta from apiclient import discovery from apiclient.discovery import build # Init API_key = ";)" # List of your watch-history files (allow several accounts) files = ['/Users/mathieu/python/Historique Youtube/Aerosmite/watch-history.json','/Users/mathieu/python/Historique Youtube/Tetedecraft/watch-history.json','/Users/mathieu/python/Historique Youtube/mattraque2000/watch-history.json'] results_dir = '/Users/mathieu/python/Historique Youtube/Resultds' # Load only videos after this date min_date = "28/08/17" # DD/MM/YY # List of allowed results files results = {'average': True, 'clean_history': True, 'channels': True, 'topics': True, 'tags': True, 'country': True, 'category': True, 'days': True, 'months': True, 'years': True} # Average percentage of total video watched watch_percentage = 60 # % service = build("youtube", "v3",developerKey=API_key) Day = ['Monday','Tuesday','Friday','Wednesday','Thursday','Sunday','Saturday'] PT_format = re.compile(r'PT((?P<hours>\d+?)H)?((?P<minutes>\d+?)M)?((?P<seconds>\d+?)S)?') min_date = datetime.strptime(min_date, '%d/%m/%y') video_part = 'snippet,contentDetails,topicDetails,recordingDetails' date_sorter = videos = average_li = channel_hm = {} topic_hm = {} tag_hm = {} country_hm = {} category_hm = {} categories_hm = {} duration_hm = {} day_name_hm = {} day_hm = {} month_hm = {} year_hm = {} # remove keyword arguments that are not set def remove_empty_kwargs(**kwargs): good_kwargs = {} if kwargs is not None: for key, value in kwargs.iteritems(): if value: good_kwargs[key] = value return good_kwargs # sample python code for videos.list def videos_list_by_id(service, **kwargs): kwargs = remove_empty_kwargs(**kwargs) return service.videos().list(**kwargs).execute() # sample python code for channels.list def channels_list_by_id(service, **kwargs): kwargs = remove_empty_kwargs(**kwargs) return service.channels().list(**kwargs).execute() # sample python code for videoCategories.list def video_categories_list(service, **kwargs): kwargs = remove_empty_kwargs(**kwargs) return service.videoCategories().list(**kwargs).execute() # sample python code for search.list def search_list_by_keyword(service, **kwargs): kwargs = remove_empty_kwargs(**kwargs) return service.search().list(**kwargs).execute() def get_new_videoId(video_title): results = search_list_by_keyword(service,maxResults=3,part='snippet',q=video_title,type='video') for video_result in results['items']: if video_result['snippet']['title'] == video_title: return video_result['id']['videoId'] return None def parse_time(time_str): parts = PT_format.match(time_str) if not parts: return parts = parts.groupdict() time_params = {} for (name, param) in parts.iteritems(): if param: time_params[name] = int(param) return timedelta(**time_params) def clear_videos_vars(): video_title = video_description = video_duration = video_categoryId = video_categoryName = video_thumbnails = video_location = video_topics = video_tags = video_date = channel_id = channel_name = channel_country = None def count_data(hm_name,key): if key in hm_name.keys(): hm_name[key] += 1 else: hm_name[key] = 1 def average(dict_name): nb = 0 for key in dict_name: nb += dict_name[key] return float(nb)/len(dict_name) def create_file(name,data): with open(os.path.join(results_dir,name + ".json"), 'w') as outfile: json.dump(data, outfile,indent=4) # save watching date of videos for f in files: for x in json.load(open(f)): date = datetime.strptime(x['snippet']['publishedAt'], '%Y-%m-%dT%H:%M:%S.000Z') delta_date = date - min_date if delta_date.total_seconds() > 0: date_sorter.append(date) if date_sorter == : raise Exception('No videos detected') print(str(len(date_sorter)) + " videos detected" ) # sort videos per watching date date_sorter.sort(reverse=True) for f in files: for x in json.load(open(f)): date = datetime.strptime(x['snippet']['publishedAt'], '%Y-%m-%dT%H:%M:%S.000Z') delta_date = date - min_date if delta_date.total_seconds() < 0: continue # define video position pos = 0 for sorted_date in date_sorter: if sorted_date == date: break else: pos += 1 # access to video data if x['snippet']['title'] != "Deleted video" and x['snippet']['title'] != "Private video": try: video_data = videos_list_by_id(service,part=video_part,id=x['contentDetails']['videoId']) if len(video_data['items']) == 0: new_videoId = get_new_videoId(x['snippet']['title']) if new_videoId != None: video_data = videos_list_by_id(service,part=video_part,id=new_videoId) else: raise IndexError video_title = video_data['items'][0]['snippet']['title'] video_description = video_data['items'][0]['snippet']['description'] video_duration = parse_time(video_data['items'][0]['contentDetails']['duration']) if pos > 0 and pos < len(date_sorter): next_date = date_sorter[pos-1] if video_duration > next_date - date: video_duration = next_date - date else: video_duration = video_duration * watch_percentage / 100 video_thumbnails = video_data['items'][0]['snippet']['thumbnails']['default']['url'] video_categoryId = video_data['items'][0]['snippet']['categoryId'] video_date = datetime.strptime(video_data['items'][0]['snippet']['publishedAt'],'%Y-%m-%dT%H:%M:%S.000Z').strftime('%d/%m/%y %H:%M') channel_id = video_data['items'][0]['snippet']['channelId'] channel_name = video_data['items'][0]['snippet']['channelTitle'] if video_data['items'][0].get('recordingDetails') != None: video_location = video_data['items'][0]['recordingDetails'].get('location') else: video_location = None if video_data['items'][0].get('topicDetails') != None: video_topics = video_data['items'][0]['topicDetails'].get('topicCategories',) else: video_topics = video_tags = video_data['items'][0]['snippet'].get('tags',) # try to get country code channel_data = channels_list_by_id(service,part='snippet',id=channel_id) channel_country = channel_data['items'][0]['snippet'].get('country',video_data['items'][0]['snippet'].get('defaultLanguage')) video_categoryName = None if channel_country != None: channel_country = channel_country.lower() count_data(country_hm,channel_country) # try to get category name try: if channel_country in categories_hm.keys(): categories_list = categories_hm[channel_country] else: if channel_country == 'en': channel_country = 'us' # or "gb" elif channel_country.split('-')[0] == 'en': channel_country = channel_country.split('-')[1] categories_list = video_categories_list(service,part='snippet',regionCode=channel_country)['items'] categories_hm[channel_country] = categories_list for categoryId in categories_list: if categoryId['id'] == video_categoryId: video_categoryName = categoryId['snippet']['title'] count_data(category_hm,video_categoryName) except: pass # save video data in dicts count_data(duration_hm,video_duration) count_data(channel_hm,channel_name) for topic in video_topics: count_data(topic_hm,topic) for tag in video_tags: count_data(tag_hm,tag.lower()) except (KeyError,IndexError): print("missing information for video '" + x['snippet']['title'] + "'") clear_videos_vars() except Exception as e: print(str(e) + " for video '" + x['snippet']['title'] + "'") clear_videos_vars() else: print("unable to access a " + x['snippet']['title']) clear_videos_vars() # save date in dicts count_data(year_hm,str(date.year)) count_data(month_hm,str(date.year) + "/" + str(format(date.month, '02'))) count_data(day_hm,str(date.year) + "/" + str(format(date.month, '02')) + "/" + str(format(date.day, '02'))) count_data(day_name_hm,date.strftime("%A")) # add all data to 'clean_history' videos.append({'pos':pos,'date':date.strftime('%d/%m/%y %H:%M'),'video':{'title':video_title,'description':video_description,'duration':str(video_duration),'category':video_categoryName,'thumbnails':video_thumbnails,'date':video_date,'location':video_location,'topics':video_topics,'tags':video_tags,'channel':{'title':channel_name,'id':channel_id,'country':channel_country}}}) print("account " + str(json.load(open(f))[0]['snippet']['channelTitle']) + " done") # create new path if necessary if os.path.isdir(results_dir) == False: os.makedirs(results_dir) # sort and save dicts in results files for list_result, value in results.iteritems(): if value: if list_result == 'average': if day_hm != {}: for day in Day: if day in day_name_hm.keys(): day_name_hm[day] = day_name_hm[day] / (len(day_hm) / 7.0) else: day_name_hm[day] = 0 average_li.append({'videos per':{'day of the week':{Day[0]:day_name_hm[Day[0]],Day[1]:day_name_hm[Day[1]],Day[2]:day_name_hm[Day[2]],Day[3]:day_name_hm[Day[3]],Day[4]:day_name_hm[Day[4]],Day[5]:day_name_hm[Day[5]],Day[6]:day_name_hm[Day[6]]},'year':average(year_hm),'month':average(month_hm),'day':average(day_hm),'channel':average(channel_hm)}}) else: average_li.append({'videos per':None}) if duration_hm != {}: nb = dur_sum = 0 for key in duration_hm: dur_sum += duration_hm[key] * key.total_seconds() nb += duration_hm[key] average_li.append({'video length (min)':dur_sum/nb/60}) else: average_li.append({'video length (min)':None}) create_file(list_result,average_li) elif list_result == 'clean_history': create_file(list_result,videos) elif list_result == 'channels': create_file(list_result,sorted(channel_hm.items(), key=operator.itemgetter(1),reverse=True)) elif list_result == 'topics': create_file(list_result,sorted(topic_hm.items(), key=operator.itemgetter(1),reverse=True)) elif list_result == 'tags': create_file(list_result,sorted(tag_hm.items(), key=operator.itemgetter(1),reverse=True)) elif list_result == 'country': create_file(list_result,sorted(country_hm.items(), key=operator.itemgetter(1),reverse=True)) elif list_result == 'category': create_file(list_result,sorted(category_hm.items(), key=operator.itemgetter(1),reverse=True)) elif list_result == 'days': create_file(list_result,sorted(day_hm.items(), key=operator.itemgetter(1),reverse=True)) elif list_result == 'months': create_file(list_result,sorted(month_hm.items(), key=operator.itemgetter(1),reverse=True)) elif list_result == 'years': create_file(list_result,year_hm(topic_hm.items(), key=operator.itemgetter(1),reverse=True))

[Redoudou]

Hello,

this is very usefull ! I can wait to be able to analyze my history. Thank you :)

ligne 29

date_sorter = videos = average_li =

should it be ?

date_sorter = {} videos = {} average_li = {}

Also would you have a text version of this code because I have an intending error and I think there might be a problem when copy paste.


Thank you !

[metulburr]

@[Redoudou]
Its an issue with our editor. its suppose to be empty lists
second bullet here
https://python-forum.io/misc.php?action=help&hid=39

[Redoudou]

Thank you @[metulburr],

I've had done the correction.

However I keep facing and intentation error in the follow function.

# remove keyword arguments that are not set def remove_empty_kwargs(**kwargs): good_kwargs = {} if kwargs is not None: for key, value in kwargs.iteritems(): if value: good_kwargs[key] = value return good_kwargs

Error:
line 47 good_kwargs = {} ^ IndentationError: expected an indented block

For what I know from my research the syntax is good. It edited the file in nqq and check good_kwargs(singlespace)=(singlespace){}
but the error stays.

@[Aerosmite] Any idea where I should oriented my research ?

Thank you :)