# -*- coding: utf-8 -*-
"""Untitled28.ipynb
Automatically generated by Colaboratory.
Original file is located at
https://colab.research.google.com/
"""
!pip install beautifulsoup4
# coding: UTF-8
import urllib.request
import urllib.parse
import re
import time
import json
from bs4 import BeautifulSoup
import collections
words = "ocaml"
keywords = urllib.parse.quote(words)
target_url = "https://www.youtube.com/results?search_query=" + keywords
html = urllib.request.urlopen(target_url).read()
html_strings = html.decode()
del(html)
soup = BeautifulSoup(html_strings, 'html.parser')
for ind,script_tag in enumerate(soup.find_all('script')):
if (ma := re.search(r"var\sytInitialData", str(script_tag))):
#if ind == 33:
json_strings = str(script_tag)[ma.end()+3:-10]
#json_strings = str(script_tag)[58:-10]
del(soup)
j_dict = json.loads(json_strings)
#print(json.dumps(j_dict, indent=1))
li_contents = []
#for k1,v1 in j_dict['contents'].items():
# if (k1 == 'twoColumnSearchResultsRenderer'):
# for k2,v2 in v1.items():
# if (k2 == 'primaryContents'):
# for k3,v3 in v2.items():
# if (k3 == 'sectionListRenderer'):
# for k4,v4 in v3.items():
# if (k4 == 'contents'):
# for ind,v5 in enumerate(v4):
# if (ind == 0):
# for k6,v6 in v5.items():
# if (k6 == 'itemSectionRenderer'):
# for k7,v7 in v6.items():
# if (k7 == 'contents'):
# li_contents.extend(v7)
v7 = j_dict['contents']['twoColumnSearchResultsRenderer']['primaryContents']['sectionListRenderer']['contents'][0]['itemSectionRenderer']['contents']
li_contents.extend(v7)
del j_dict
#print(len(li_contents))
uniq_videoid = []
uniq_title = []
def recur(v):
if type(v) == list:
# print('list',v)
# print()
for lv in v:
recur(lv)
elif type(v) == dict:
# print('dict',v)
for dk,dv in v.items():
recur2(dk,dv)
# print()
def recur2(k,v):
#print('{' + k + ' : ' + str(v)+ '}')
#print()
if(k=='videoId' or k=='title'):
if k== 'videoId':
# print(v)
if not(v in uniq_videoid):
uniq_videoid.append(v)
if k== 'title':
if (check := 'runs' in v):
# print(v['runs'][0]['text'])
if not(v['runs'][0]['text'] in uniq_title):
uniq_title.append(v['runs'][0]['text'])
# print()
if type(v) == list:
for ind,lv in enumerate(v):
# print('list',ind,'=>')
recur(lv)
elif type(v) == dict:
# print('dict',v)
# print()
for dk,dv in v.items():
recur2(dk,dv)
while len(li_contents) > 0 :
head = li_contents.pop(0)
recur(head)
zip_list = zip(uniq_videoid,uniq_title)
#for ind,x in enumerate(zip_list):
# print (ind,x)
# print ()
# check values
search_list_temporary = [x for x in zip_list if x != '' ]
del(zip_list)
uniq_videoid.clear()
uniq_title.clear()
######## ##### ###### ###### ###### ###### ###### ######
#
#for ind,p in enumerate(search_list_temporary):
# print(ind, p)
#print()
#
######## ##### ###### ###### ###### ###### ###### ######
# check keywords in video title
search_list = [ x for x in search_list_temporary if re.search(rf'(?i){words}',x[1]) ]
del(search_list_temporary)
#for ind,p in enumerate(search_list):
# print(ind, p)
#print()
#
######## ##### ###### ###### ###### ###### ###### ######
video_list_0 = []
already_list = []
total_list = []
id_cell_1 = {}
id_cell_2 = {}
#video_list_0.extend(search_list) # [0] videoid / [1] title
search_list2 = [x[0] for x in search_list] # [0] videoid / [1] title
video_list_0.extend(search_list2) # [0] videoid
def first_gether(words, target_url, total_list, id_cell_1, id_cell_2):
id_cell_seconds = {}
id_cell_title = {}
ommit = target_url.replace("https://www.youtube.com/watch?v=",'') # videoid
html = urllib.request.urlopen(target_url).read()
html_strings = html.decode()
del(html)
soup = BeautifulSoup(html_strings, 'html.parser')
for ind,script_tag in enumerate(soup.find_all('script')):
if (ma1 := re.search(r"var\sytInitialData",str(script_tag))):
#if ind == 43:
json_strings = str(script_tag)[ma1.end()+3:-10]
#json_strings = str(script_tag)[58:-10]
#page_title = soup.find('title')
#print(page_title)
del(soup)
print()
j_dict = json.loads(json_strings)
current_item_data = j_dict['contents']['twoColumnWatchNextResults']['results']
secondary_data = j_dict['contents']['twoColumnWatchNextResults']['secondaryResults']
del j_dict
current_title = [] ## only one size list
current_videoid = [] ## only one size list
current_description = [] ## only one size list
def f1(v):###:###:###:###:###:###
if type(v) == list:
# print('list',v)
# print()
for lv in v:
f1(lv)
elif type(v) == dict:
# print('dict',v)
for dk,dv in v.items():
f2(dk,dv)
# print()
def f2(k,v):
# print('{' + k + ' : ' + str(v)+ '}')
# print()
if(k=='dateText'):
# print(k,v['simpleText'])
pass
if(k=='description'):
# print(v['runs'][0]['text'])
current_description.append(v['runs'][0]['text'])
if(k=='videoId' or k=='title'):
if k== 'videoId':
# print(v)
if (empty := (len(current_videoid) < 1)):
if not(v in current_videoid):
current_videoid.append(v)
if k== 'title':
if (empty := (len(current_title) < 1)):
if (check := 'runs' in v):
if not(v['runs'][0]['text'] in current_title):
# print(v['runs'][0]['text'])
current_title.append(v['runs'][0]['text'])
# print()
if type(v) == list:
for ind,lv in enumerate(v):
# print('list',ind,'=>')
f1(lv)
elif type(v) == dict:
# print('dict',v)
# print()
for dk,dv in v.items():
f2(dk,dv)
###:###:###:###:###:###
f1(current_item_data) # recursive function
# print()
# print('current contents')
# print(current_videoid)
# print(current_title)
# print()
uniq_videoids = []
uniq_titles = []
def rr(v):
if type(v) == list:
# print('list',v)
# print()
for lv in v:
rr(lv)
elif type(v) == dict:
# print('dict',v)
for dk,dv in v.items():
rr2(dk,dv)
# print()
def rr2(k,v):
# print('{' + k + ' : ' + str(v)+ '}')
# print()
if(k=='videoId' or k=='title'):
if k== 'videoId':
# print(v)
if str(v) != ommit:
uniq_videoids.append(v)
else:
# print(v)
pass
if k== 'title':
if (check := 'accessibility' in v): # key check
if v['accessibility']['accessibilityData']['label'] != None:
if not(v['accessibility']['accessibilityData']['label'] in uniq_titles):
uniq_titles.append(v['accessibility']['accessibilityData']['label'])
# print()
if type(v) == list:
for ind,lv in enumerate(v):
# print('list',ind,'=>')
rr(lv)
elif type(v) == dict:
# print('dict',v)
# print()
for dk,dv in v.items():
rr2(dk,dv)
rr(secondary_data)
poplist = []
for i,x in enumerate(uniq_videoids):
k = uniq_videoids.count(x)
if k > 3:
pass
else:
poplist.append(i)
for x in poplist:
uniq_videoids.pop(x)
poplist.clear()
new_uniq_videoids = []
for x in uniq_videoids:
if (check := x in new_uniq_videoids):
pass
else:
new_uniq_videoids.append(x)
# check
# print('--')
# print(len(new_uniq_videoids))
# print(len(uniq_titles))
# print('--')
titles_videoids = zip(uniq_titles,new_uniq_videoids)
# check
# for z in titles_videoids:
# print('----------------------')
# print(*z)
# print('----------------------')
videoid_list = []
title_list = []
durations_list = []
channelid_list = []
titles_videoids2 = [ x for x in titles_videoids if re.search(rf'(?i){words}',x[0])]
video_list = []
id_data = [x[1] for x in titles_videoids2]
video_list.extend(id_data)
title = re.search(r"(?<=\<title\>).*?(?=\<\/title\>)",html_strings)
if title is None:
print("skip")
else:
title_strings = title.group()
kakawari = re.search(rf'(?i){words}',title_strings)
kakawari2 = re.search(rf'(?i){words}',str(current_description[0]))
if (kakawari is None) and (kakawari2 is None):
print()
print(title_strings)
print("skip (not relevant)")
print()
pass
else:
total_list.append(target_url)
idxxx = target_url.replace("https://www.youtube.com/watch\?v=","")
#id_cell_title[idxxx] = title_strings
id_cell_title[idxxx] = current_title[0]
print()
print('!!!',title_strings)
print('@@@',target_url)
keyword2 = re.search(r'"channelId" content=',html_strings)
if keyword2 is None:
pass
else:
channelid = keyword2.end(0)
print("channelId",html_strings[channelid:(channelid + 26)])
id_cell_1[idxxx] = html_strings[channelid:(channelid + 26)]
keyword3 = re.search(r'"datePublished" content=',html_strings)
if keyword3 is None:
pass
else:
published = keyword3.end(0)
print("datePublished",html_strings[published:(published + 12)])
keyword4 = re.search(r'"uploadDate" content=',html_strings)
if keyword4 is None:
pass
else:
uploaded = keyword4.end(0)
print("uploadDate",html_strings[uploaded:(uploaded + 12)])
id_cell_2[idxxx] = html_strings[uploaded:(uploaded + 12)]
keywords5 = re.search(r'"lengthSeconds":"(\d+)",',html_strings)
if keywords5 is None:
pass
else:
seconds = keywords5.group(1)
print("seconds",seconds)
id_cell_seconds[idxxx] = seconds
print(current_description[0])
# for ind,p in enumerate(video_list):
# print(ind,p)
return video_list, total_list, id_cell_1, id_cell_2, id_cell_seconds, id_cell_title
def second_gether(words, x, video_list_0, already_list, total_list, id_cell_1, id_cell_2):
url_list = [ "https://www.youtube.com/watch?v=" + str(x) for x in video_list_0 ]
# print(video_list_0)
video_list_0.clear()
if len(url_list) > 0:
sum_list = []
rem = []
for i,url in enumerate(url_list):
if not re.match(r'^http',url):
# if not re.match(r'^@',url):
continue
# url = url.replace('@ ',"https://www.youtube.com/watch\?v=")
videoid = url.replace("https://www.youtube.com/watch\?v=","")
if videoid in id_cell_1:
# print("----------------")
# print(x,';',(i + 1),';',"skip ")
# print("----------------")
continue
if url in already_list:
# print()
# print(x,';',(i + 1),';',"skip ")
# print(" (not relevant)")
rem.append(url)
continue
# print("----------------")
# print(x,';',(i + 1),';',url)
try:
video_list, total_list, id_cell_1, id_cell_2, id_cell_seconds, id_cell_title = first_gether(words, url, total_list, id_cell_1, id_cell_2)
except:
time.sleep(1)
continue
already_list.append(url)
# l o g
### ## ## ## ## ## ## ## ## ## ## ## ## ## ##
if videoid in id_cell_1 and videoid in id_cell_2:
with open('log.txt', 'a') as out:
try:
out.write( "vi_id: " + videoid + " up_id: " + id_cell_2[videoid] + " ch_id: " + id_cell_1[videoid] + " second: " + id_cell_seconds[videoid] + " title: " + id_cell_title[videoid] + "\n")
except:
print("file write error")
out.close()
del(id_cell_seconds, id_cell_title) # from first_gether function return values
### ## ## ## ## ## ## ## ## ## ## ## ## ## ##
if ( len(video_list) > 0 ):
# video_list = list(set(video_list))
sum_list.extend(video_list)
total_list = list(set(total_list))
if ( len(rem) > 0 ):
for remove in rem:
if remove in sum_list:
inum = sum_list.index(remove)
sum_list.pop(inum)
new_list = sorted(list(set(sum_list)),key=sum_list.index)
video_list_0.extend(new_list)
# for ind,p in enumerate(video_list_0):
# print(ind,p)
x = x + 1
if ( x < 4000 and len(new_list) > 0 ):
already_list, total_list, id_cell_1, id_cell_2 = second_gether(words, x, video_list_0, already_list, total_list, id_cell_1, id_cell_2)
return already_list,total_list,id_cell_1,id_cell_2
# top level
counter_x = 0
second_gether(words, counter_x, video_list_0, already_list, total_list, id_cell_1, id_cell_2)