aboutsummaryrefslogtreecommitdiffstats
path: root/youtube/subscriptions.py
diff options
context:
space:
mode:
authorLibravatarLibravatar Biswakalyan Bhuyan <biswa@surgot.in> 2024-09-19 15:33:11 +0530
committerLibravatarLibravatar Biswakalyan Bhuyan <biswa@surgot.in> 2024-09-19 15:33:11 +0530
commita4e01da27c08e43a67b2618ad1e71c1f8f86d5cd (patch)
tree5b8f407dbb7e9d1ab2106ac0cc8564897e7a2098 /youtube/subscriptions.py
downloadyt-local-master.tar.gz
yt-local-master.tar.bz2
yt-local-master.zip
youtube fronendHEADmaster
Diffstat (limited to 'youtube/subscriptions.py')
-rw-r--r--youtube/subscriptions.py1107
1 files changed, 1107 insertions, 0 deletions
diff --git a/youtube/subscriptions.py b/youtube/subscriptions.py
new file mode 100644
index 0000000..04d3c5a
--- /dev/null
+++ b/youtube/subscriptions.py
@@ -0,0 +1,1107 @@
+from youtube import util, yt_data_extract, channel, local_playlist, playlist
+from youtube import yt_app
+import settings
+
+import sqlite3
+import os
+import time
+import gevent
+import json
+import traceback
+import contextlib
+import defusedxml.ElementTree
+import urllib
+import math
+import secrets
+import collections
+import calendar # bullshit! https://bugs.python.org/issue6280
+import csv
+import re
+
+import flask
+from flask import request
+
+
+thumbnails_directory = os.path.join(settings.data_dir, "subscription_thumbnails")
+
+# https://stackabuse.com/a-sqlite-tutorial-with-python/
+
+database_path = os.path.join(settings.data_dir, "subscriptions.sqlite")
+
+
+def open_database():
+ if not os.path.exists(settings.data_dir):
+ os.makedirs(settings.data_dir)
+ connection = sqlite3.connect(database_path, check_same_thread=False)
+
+ try:
+ cursor = connection.cursor()
+ cursor.execute('''PRAGMA foreign_keys = 1''')
+ # Create tables if they don't exist
+ cursor.execute('''CREATE TABLE IF NOT EXISTS subscribed_channels (
+ id integer PRIMARY KEY,
+ yt_channel_id text UNIQUE NOT NULL,
+ channel_name text NOT NULL,
+ time_last_checked integer DEFAULT 0,
+ next_check_time integer DEFAULT 0,
+ muted integer DEFAULT 0
+ )''')
+ cursor.execute('''CREATE TABLE IF NOT EXISTS videos (
+ id integer PRIMARY KEY,
+ sql_channel_id integer NOT NULL REFERENCES subscribed_channels(id) ON UPDATE CASCADE ON DELETE CASCADE,
+ video_id text UNIQUE NOT NULL,
+ title text NOT NULL,
+ duration text,
+ time_published integer NOT NULL,
+ is_time_published_exact integer DEFAULT 0,
+ time_noticed integer NOT NULL,
+ description text,
+ watched integer default 0
+ )''')
+ cursor.execute('''CREATE TABLE IF NOT EXISTS tag_associations (
+ id integer PRIMARY KEY,
+ tag text NOT NULL,
+ sql_channel_id integer NOT NULL REFERENCES subscribed_channels(id) ON UPDATE CASCADE ON DELETE CASCADE,
+ UNIQUE(tag, sql_channel_id)
+ )''')
+ cursor.execute('''CREATE TABLE IF NOT EXISTS db_info (
+ version integer DEFAULT 1
+ )''')
+
+ connection.commit()
+ except:
+ connection.rollback()
+ connection.close()
+ raise
+
+ # https://stackoverflow.com/questions/19522505/using-sqlite3-in-python-with-with-keyword
+ return contextlib.closing(connection)
+
+
+def with_open_db(function, *args, **kwargs):
+ with open_database() as connection:
+ with connection as cursor:
+ return function(cursor, *args, **kwargs)
+
+
+def _is_subscribed(cursor, channel_id):
+ result = cursor.execute('''SELECT EXISTS(
+ SELECT 1
+ FROM subscribed_channels
+ WHERE yt_channel_id=?
+ LIMIT 1
+ )''', [channel_id]).fetchone()
+ return bool(result[0])
+
+
+def is_subscribed(channel_id):
+ if not os.path.exists(database_path):
+ return False
+
+ return with_open_db(_is_subscribed, channel_id)
+
+
+def _subscribe(channels):
+ ''' channels is a list of (channel_id, channel_name) '''
+ channels = list(channels)
+ with open_database() as connection:
+ with connection as cursor:
+ channel_ids_to_check = [channel[0] for channel in channels if not _is_subscribed(cursor, channel[0])]
+
+ rows = ((channel_id, channel_name, 0, 0) for channel_id, channel_name in channels)
+ cursor.executemany('''INSERT OR IGNORE INTO subscribed_channels (yt_channel_id, channel_name, time_last_checked, next_check_time)
+ VALUES (?, ?, ?, ?)''', rows)
+
+ if settings.autocheck_subscriptions:
+ # important that this is after the changes have been committed to database
+ # otherwise the autochecker (other thread) tries checking the channel before it's in the database
+ channel_names.update(channels)
+ check_channels_if_necessary(channel_ids_to_check)
+
+
+def delete_thumbnails(to_delete):
+ for thumbnail in to_delete:
+ try:
+ video_id = thumbnail[0:-4]
+ if video_id in existing_thumbnails:
+ os.remove(os.path.join(thumbnails_directory, thumbnail))
+ existing_thumbnails.remove(video_id)
+ except Exception:
+ print('Failed to delete thumbnail: ' + thumbnail)
+ traceback.print_exc()
+
+
+def _unsubscribe(cursor, channel_ids):
+ ''' channel_ids is a list of channel_ids '''
+ to_delete = []
+ for channel_id in channel_ids:
+ rows = cursor.execute('''SELECT video_id
+ FROM videos
+ WHERE sql_channel_id = (
+ SELECT id
+ FROM subscribed_channels
+ WHERE yt_channel_id=?
+ )''', (channel_id,)).fetchall()
+ to_delete += [row[0] + '.jpg' for row in rows]
+
+ gevent.spawn(delete_thumbnails, to_delete)
+ cursor.executemany("DELETE FROM subscribed_channels WHERE yt_channel_id=?", ((channel_id, ) for channel_id in channel_ids))
+
+
+def _get_videos(cursor, number_per_page, offset, tag=None):
+ '''Returns a full page of videos with an offset, and a value good enough to be used as the total number of videos'''
+ # We ask for the next 9 pages from the database
+ # Then the actual length of the results tell us if there are more than 9 pages left, and if not, how many there actually are
+ # This is done since there are only 9 page buttons on display at a time
+ # If there are more than 9 pages left, we give a fake value in place of the real number of results if the entire database was queried without limit
+ # This fake value is sufficient to get the page button generation macro to display 9 page buttons
+ # If we wish to display more buttons this logic must change
+ # We cannot use tricks with the sql id for the video since we frequently have filters and other restrictions in place on the results anyway
+ # TODO: This is probably not the ideal solution
+ if tag is not None:
+ db_videos = cursor.execute('''SELECT video_id, title, duration, time_published, is_time_published_exact, channel_name, yt_channel_id
+ FROM videos
+ INNER JOIN subscribed_channels on videos.sql_channel_id = subscribed_channels.id
+ INNER JOIN tag_associations on videos.sql_channel_id = tag_associations.sql_channel_id
+ WHERE tag = ? AND muted = 0
+ ORDER BY time_noticed DESC, time_published DESC
+ LIMIT ? OFFSET ?''', (tag, number_per_page*9, offset)).fetchall()
+ else:
+ db_videos = cursor.execute('''SELECT video_id, title, duration, time_published, is_time_published_exact, channel_name, yt_channel_id
+ FROM videos
+ INNER JOIN subscribed_channels on videos.sql_channel_id = subscribed_channels.id
+ WHERE muted = 0
+ ORDER BY time_noticed DESC, time_published DESC
+ LIMIT ? OFFSET ?''', (number_per_page*9, offset)).fetchall()
+
+ pseudo_number_of_videos = offset + len(db_videos)
+
+ videos = []
+ for db_video in db_videos[0:number_per_page]:
+ videos.append({
+ 'id': db_video[0],
+ 'title': db_video[1],
+ 'duration': db_video[2],
+ 'time_published': exact_timestamp(db_video[3]) if db_video[4] else posix_to_dumbed_down(db_video[3]),
+ 'author': db_video[5],
+ 'author_id': db_video[6],
+ 'author_url': '/https://www.youtube.com/channel/' + db_video[6],
+ })
+
+ return videos, pseudo_number_of_videos
+
+
+def _get_subscribed_channels(cursor):
+ for item in cursor.execute('''SELECT channel_name, yt_channel_id, muted
+ FROM subscribed_channels
+ ORDER BY channel_name COLLATE NOCASE'''):
+ yield item
+
+
+def _add_tags(cursor, channel_ids, tags):
+ pairs = [(tag, yt_channel_id) for tag in tags for yt_channel_id in channel_ids]
+ cursor.executemany('''INSERT OR IGNORE INTO tag_associations (tag, sql_channel_id)
+ SELECT ?, id FROM subscribed_channels WHERE yt_channel_id = ? ''', pairs)
+
+
+def _remove_tags(cursor, channel_ids, tags):
+ pairs = [(tag, yt_channel_id) for tag in tags for yt_channel_id in channel_ids]
+ cursor.executemany('''DELETE FROM tag_associations
+ WHERE tag = ? AND sql_channel_id = (
+ SELECT id FROM subscribed_channels WHERE yt_channel_id = ?
+ )''', pairs)
+
+
+def _get_tags(cursor, channel_id):
+ return [row[0] for row in cursor.execute('''SELECT tag
+ FROM tag_associations
+ WHERE sql_channel_id = (
+ SELECT id FROM subscribed_channels WHERE yt_channel_id = ?
+ )''', (channel_id,))]
+
+
+def _get_all_tags(cursor):
+ return [row[0] for row in cursor.execute('''SELECT DISTINCT tag FROM tag_associations''')]
+
+
+def _get_channel_names(cursor, channel_ids):
+ ''' returns list of (channel_id, channel_name) '''
+ result = []
+ for channel_id in channel_ids:
+ row = cursor.execute('''SELECT channel_name
+ FROM subscribed_channels
+ WHERE yt_channel_id = ?''', (channel_id,)).fetchone()
+ result.append((channel_id, row[0]))
+ return result
+
+
+def _channels_with_tag(cursor, tag, order=False, exclude_muted=False, include_muted_status=False):
+ ''' returns list of (channel_id, channel_name) '''
+
+ statement = '''SELECT yt_channel_id, channel_name'''
+
+ if include_muted_status:
+ statement += ''', muted'''
+
+ statement += '''
+ FROM subscribed_channels
+ WHERE subscribed_channels.id IN (
+ SELECT tag_associations.sql_channel_id FROM tag_associations WHERE tag=?
+ )
+ '''
+ if exclude_muted:
+ statement += '''AND muted != 1\n'''
+ if order:
+ statement += '''ORDER BY channel_name COLLATE NOCASE'''
+
+ return cursor.execute(statement, [tag]).fetchall()
+
+
+def _schedule_checking(cursor, channel_id, next_check_time):
+ cursor.execute('''UPDATE subscribed_channels SET next_check_time = ? WHERE yt_channel_id = ?''', [int(next_check_time), channel_id])
+
+
+def _is_muted(cursor, channel_id):
+ return bool(cursor.execute('''SELECT muted FROM subscribed_channels WHERE yt_channel_id=?''', [channel_id]).fetchone()[0])
+
+
+units = collections.OrderedDict([
+ ('year', 31536000), # 365*24*3600
+ ('month', 2592000), # 30*24*3600
+ ('week', 604800), # 7*24*3600
+ ('day', 86400), # 24*3600
+ ('hour', 3600),
+ ('minute', 60),
+ ('second', 1),
+])
+
+
+def youtube_timestamp_to_posix(dumb_timestamp):
+ ''' Given a dumbed down timestamp such as 1 year ago, 3 hours ago,
+ approximates the unix time (seconds since 1/1/1970) '''
+ dumb_timestamp = dumb_timestamp.lower()
+ now = time.time()
+ if dumb_timestamp == "just now":
+ return now
+ split = dumb_timestamp.split(' ')
+ quantifier, unit = int(split[0]), split[1]
+ if quantifier > 1:
+ unit = unit[:-1] # remove s from end
+ return now - quantifier*units[unit]
+
+
+def posix_to_dumbed_down(posix_time):
+ '''Inverse of youtube_timestamp_to_posix.'''
+ delta = int(time.time() - posix_time)
+ assert delta >= 0
+
+ if delta == 0:
+ return '0 seconds ago'
+
+ for unit_name, unit_time in units.items():
+ if delta >= unit_time:
+ quantifier = round(delta/unit_time)
+ if quantifier == 1:
+ return '1 ' + unit_name + ' ago'
+ else:
+ return str(quantifier) + ' ' + unit_name + 's ago'
+ else:
+ raise Exception()
+
+
+def exact_timestamp(posix_time):
+ result = time.strftime('%I:%M %p %m/%d/%y', time.localtime(posix_time))
+ if result[0] == '0': # remove 0 infront of hour (like 01:00 PM)
+ return result[1:]
+ return result
+
+
+try:
+ existing_thumbnails = set(os.path.splitext(name)[0] for name in os.listdir(thumbnails_directory))
+except FileNotFoundError:
+ existing_thumbnails = set()
+
+
+# --- Manual checking system. Rate limited in order to support very large numbers of channels to be checked ---
+# Auto checking system plugs into this for convenience, though it doesn't really need the rate limiting
+
+check_channels_queue = util.RateLimitedQueue()
+checking_channels = set()
+
+# Just to use for printing channel checking status to console without opening database
+channel_names = dict()
+
+
+def check_channel_worker():
+ while True:
+ channel_id = check_channels_queue.get()
+ try:
+ _get_upstream_videos(channel_id)
+ except Exception:
+ traceback.print_exc()
+ finally:
+ checking_channels.remove(channel_id)
+
+
+for i in range(0, 5):
+ gevent.spawn(check_channel_worker)
+# ----------------------------
+
+
+# --- Auto checking system - Spaghetti code ---
+def autocheck_dispatcher():
+ '''Scans the auto_check_list. Sleeps until the earliest job is due, then adds that channel to the checking queue above. Can be sent a new job through autocheck_job_application'''
+ while True:
+ if len(autocheck_jobs) == 0:
+ new_job = autocheck_job_application.get()
+ autocheck_jobs.append(new_job)
+ else:
+ earliest_job_index = min(range(0, len(autocheck_jobs)), key=lambda index: autocheck_jobs[index]['next_check_time']) # https://stackoverflow.com/a/11825864
+ earliest_job = autocheck_jobs[earliest_job_index]
+ time_until_earliest_job = earliest_job['next_check_time'] - time.time()
+
+ if time_until_earliest_job <= -5: # should not happen unless we're running extremely slow
+ print('ERROR: autocheck_dispatcher got job scheduled in the past, skipping and rescheduling: ' + earliest_job['channel_id'] + ', ' + earliest_job['channel_name'] + ', ' + str(earliest_job['next_check_time']))
+ next_check_time = time.time() + 3600*secrets.randbelow(60)/60
+ with_open_db(_schedule_checking, earliest_job['channel_id'], next_check_time)
+ autocheck_jobs[earliest_job_index]['next_check_time'] = next_check_time
+ continue
+
+ # make sure it's not muted
+ if with_open_db(_is_muted, earliest_job['channel_id']):
+ del autocheck_jobs[earliest_job_index]
+ continue
+
+ if time_until_earliest_job > 0: # it can become less than zero (in the past) when it's set to go off while the dispatcher is doing something else at that moment
+ try:
+ new_job = autocheck_job_application.get(timeout=time_until_earliest_job) # sleep for time_until_earliest_job time, but allow to be interrupted by new jobs
+ except gevent.queue.Empty: # no new jobs
+ pass
+ else: # new job, add it to the list
+ autocheck_jobs.append(new_job)
+ continue
+
+ # no new jobs, time to execute the earliest job
+ channel_names[earliest_job['channel_id']] = earliest_job['channel_name']
+ checking_channels.add(earliest_job['channel_id'])
+ check_channels_queue.put(earliest_job['channel_id'])
+ del autocheck_jobs[earliest_job_index]
+
+
+dispatcher_greenlet = None
+
+
+def start_autocheck_system():
+ global autocheck_job_application
+ global autocheck_jobs
+ global dispatcher_greenlet
+
+ # job application format: dict with keys (channel_id, channel_name, next_check_time)
+ autocheck_job_application = gevent.queue.Queue() # only really meant to hold 1 item, just reusing gevent's wait and timeout machinery
+
+ autocheck_jobs = [] # list of dicts with the keys (channel_id, channel_name, next_check_time). Stores all the channels that need to be autochecked and when to check them
+ with open_database() as connection:
+ with connection as cursor:
+ now = time.time()
+ for row in cursor.execute('''SELECT yt_channel_id, channel_name, next_check_time FROM subscribed_channels WHERE muted != 1''').fetchall():
+
+ if row[2] is None:
+ next_check_time = 0
+ else:
+ next_check_time = row[2]
+
+ # expired, check randomly within the next hour
+ # note: even if it isn't scheduled in the past right now, it might end up being if it's due soon and we dont start dispatching by then, see below where time_until_earliest_job is negative
+ if next_check_time < now:
+ next_check_time = now + 3600*secrets.randbelow(60)/60
+ row = (row[0], row[1], next_check_time)
+ _schedule_checking(cursor, row[0], next_check_time)
+ autocheck_jobs.append({'channel_id': row[0], 'channel_name': row[1], 'next_check_time': next_check_time})
+ dispatcher_greenlet = gevent.spawn(autocheck_dispatcher)
+
+
+def stop_autocheck_system():
+ if dispatcher_greenlet is not None:
+ dispatcher_greenlet.kill()
+
+
+def autocheck_setting_changed(old_value, new_value):
+ if new_value:
+ start_autocheck_system()
+ else:
+ stop_autocheck_system()
+
+
+settings.add_setting_changed_hook(
+ 'autocheck_subscriptions',
+ autocheck_setting_changed
+)
+if settings.autocheck_subscriptions:
+ start_autocheck_system()
+# ----------------------------
+
+
+def check_channels_if_necessary(channel_ids):
+ for channel_id in channel_ids:
+ if channel_id not in checking_channels:
+ checking_channels.add(channel_id)
+ check_channels_queue.put(channel_id)
+
+
+def _get_atoma_feed(channel_id):
+ url = 'https://www.youtube.com/feeds/videos.xml?channel_id=' + channel_id
+ try:
+ return util.fetch_url(url).decode('utf-8')
+ except util.FetchError as e:
+ # 404 is expected for terminated channels
+ if e.code in ('404', '429'):
+ return ''
+ if e.code == '502':
+ return str(e)
+ raise
+
+
+def _get_channel_videos_first_page(channel_id, channel_status_name):
+ try:
+ # First try the playlist method
+ pl_json = playlist.get_videos(
+ 'UU' + channel_id[2:],
+ 1,
+ include_shorts=settings.include_shorts_in_subscriptions,
+ report_text=None
+ )
+ pl_info = yt_data_extract.extract_playlist_info(pl_json)
+ if pl_info.get('items'):
+ pl_info['items'] = pl_info['items'][0:30]
+ return pl_info
+
+ # Try the channel api method
+ channel_json = channel.get_channel_first_page(channel_id=channel_id)
+ channel_info = yt_data_extract.extract_channel_info(
+ json.loads(channel_json), 'videos'
+ )
+ return channel_info
+ except util.FetchError as e:
+ if e.code == '429' and settings.route_tor:
+ error_message = ('Error checking channel ' + channel_status_name
+ + ': YouTube blocked the request because the'
+ + ' Tor exit node is overutilized. Try getting a new exit node'
+ + ' by using the New Identity button in the Tor Browser.')
+ if e.ip:
+ error_message += ' Exit node IP address: ' + e.ip
+ print(error_message)
+ return None
+ elif e.code == '502':
+ print('Error checking channel', channel_status_name + ':', str(e))
+ return None
+ raise
+
+
+def _get_upstream_videos(channel_id):
+ try:
+ channel_status_name = channel_names[channel_id]
+ except KeyError:
+ channel_status_name = channel_id
+
+ print("Checking channel: " + channel_status_name)
+
+ tasks = (
+ # channel page, need for video duration
+ gevent.spawn(_get_channel_videos_first_page, channel_id,
+ channel_status_name),
+ # need atoma feed for exact published time
+ gevent.spawn(_get_atoma_feed, channel_id)
+ )
+ gevent.joinall(tasks)
+
+ channel_info, feed = tasks[0].value, tasks[1].value
+
+ # extract published times from atoma feed
+ times_published = {}
+ try:
+ def remove_bullshit(tag):
+ '''Remove XML namespace bullshit from tagname. https://bugs.python.org/issue18304'''
+ if '}' in tag:
+ return tag[tag.rfind('}')+1:]
+ return tag
+
+ def find_element(base, tag_name):
+ for element in base:
+ if remove_bullshit(element.tag) == tag_name:
+ return element
+ return None
+
+ root = defusedxml.ElementTree.fromstring(feed)
+ assert remove_bullshit(root.tag) == 'feed'
+ for entry in root:
+ if (remove_bullshit(entry.tag) != 'entry'):
+ continue
+
+ # it's yt:videoId in the xml but the yt: is turned into a namespace which is removed by remove_bullshit
+ video_id_element = find_element(entry, 'videoId')
+ time_published_element = find_element(entry, 'published')
+ assert video_id_element is not None
+ assert time_published_element is not None
+
+ time_published = int(calendar.timegm(time.strptime(time_published_element.text, '%Y-%m-%dT%H:%M:%S+00:00')))
+ times_published[video_id_element.text] = time_published
+
+ except AssertionError:
+ print('Failed to read atoma feed for ' + channel_status_name)
+ traceback.print_exc()
+ except defusedxml.ElementTree.ParseError:
+ print('Failed to read atoma feed for ' + channel_status_name)
+
+ if channel_info is None: # there was an error
+ return
+ if channel_info['error']:
+ print('Error checking channel ' + channel_status_name + ': ' + channel_info['error'])
+ return
+
+ videos = channel_info['items']
+ for i, video_item in enumerate(videos):
+ if not video_item.get('description'):
+ video_item['description'] = ''
+ else:
+ video_item['description'] = ''.join(run.get('text', '') for run in video_item['description'])
+
+ if video_item['id'] in times_published:
+ video_item['time_published'] = times_published[video_item['id']]
+ video_item['is_time_published_exact'] = True
+ elif video_item.get('time_published'):
+ video_item['is_time_published_exact'] = False
+ try:
+ video_item['time_published'] = youtube_timestamp_to_posix(video_item['time_published']) - i # subtract a few seconds off the videos so they will be in the right order
+ except Exception:
+ print(video_item)
+ else:
+ video_item['is_time_published_exact'] = False
+ video_item['time_published'] = None
+ video_item['channel_id'] = channel_id
+ if len(videos) > 1:
+ # Go back and fill in any videos that don't have a time published
+ # using the time published of the surrounding ones
+ for i in range(len(videos)-1):
+ if (videos[i+1]['time_published'] is None
+ and videos[i]['time_published'] is not None
+ ):
+ videos[i+1]['time_published'] = videos[i]['time_published'] - 1
+ for i in reversed(range(1,len(videos))):
+ if (videos[i-1]['time_published'] is None
+ and videos[i]['time_published'] is not None
+ ):
+ videos[i-1]['time_published'] = videos[i]['time_published'] + 1
+ # Special case: none of the videos have a time published.
+ # In this case, make something up
+ if videos and videos[0]['time_published'] is None:
+ assert all(v['time_published'] is None for v in videos)
+ now = time.time()
+ for i in range(len(videos)):
+ # 1 month between videos
+ videos[i]['time_published'] = now - i*3600*24*30
+
+
+ if len(videos) == 0:
+ average_upload_period = 4*7*24*3600 # assume 1 month for channel with no videos
+ elif len(videos) < 5:
+ average_upload_period = int((time.time() - videos[len(videos)-1]['time_published'])/len(videos))
+ else:
+ average_upload_period = int((time.time() - videos[4]['time_published'])/5) # equivalent to averaging the time between videos for the last 5 videos
+
+ # calculate when to check next for auto checking
+ # add some quantization and randomness to make pattern analysis by YouTube slightly harder
+ quantized_upload_period = average_upload_period - (average_upload_period % (4*3600)) + 4*3600 # round up to nearest 4 hours
+ randomized_upload_period = quantized_upload_period*(1 + secrets.randbelow(50)/50*0.5) # randomly between 1x and 1.5x
+ next_check_delay = randomized_upload_period/10 # check at 10x the channel posting rate. might want to fine tune this number
+ next_check_time = int(time.time() + next_check_delay)
+
+ with open_database() as connection:
+ with connection as cursor:
+
+ # Get video ids and duration of existing vids so we
+ # can see how many new ones there are and update
+ # livestreams/premiers
+ existing_vids = list(cursor.execute(
+ '''SELECT video_id, duration
+ FROM videos
+ INNER JOIN subscribed_channels
+ ON videos.sql_channel_id = subscribed_channels.id
+ WHERE yt_channel_id=?
+ ORDER BY time_published DESC
+ LIMIT 30''', [channel_id]).fetchall())
+ existing_vid_ids = set(row[0] for row in existing_vids)
+ existing_durs = dict(existing_vids)
+
+ # new videos the channel has uploaded since last time we checked
+ number_of_new_videos = 0
+ for video in videos:
+ if video['id'] in existing_vid_ids:
+ break
+ number_of_new_videos += 1
+
+ is_first_check = cursor.execute('''SELECT time_last_checked FROM subscribed_channels WHERE yt_channel_id=?''', [channel_id]).fetchone()[0] in (None, 0)
+ time_videos_retrieved = int(time.time())
+ rows = []
+ update_rows = []
+ for i, video_item in enumerate(videos):
+ if (is_first_check
+ or number_of_new_videos > 6
+ or i >= number_of_new_videos):
+ # don't want a crazy ordering on first check or check in a long time, since we're ordering by time_noticed
+ # Last condition is for when the channel deleting videos
+ # causes new videos to appear at the end of the backlog.
+ # For instance, if we have 30 vids in the DB, and 1 vid
+ # that we previously saw has since been deleted,
+ # then a video we haven't seen before will appear as the
+ # 30th. Don't want this to be considered a newly noticed
+ # vid which would appear at top of subscriptions feed
+ time_noticed = video_item['time_published']
+ else:
+ time_noticed = time_videos_retrieved
+
+ # videos which need durations updated
+ non_durations = ('upcoming', 'none', 'live', '')
+ v_id = video_item['id']
+ if (existing_durs.get(v_id) is not None
+ and existing_durs[v_id].lower() in non_durations
+ and video_item['duration'] not in non_durations
+ ):
+ update_rows.append((
+ video_item['title'],
+ video_item['duration'],
+ video_item['time_published'],
+ video_item['is_time_published_exact'],
+ video_item['description'],
+ video_item['id'],
+ ))
+ # all other videos
+ else:
+ rows.append((
+ video_item['channel_id'],
+ video_item['id'],
+ video_item['title'],
+ video_item['duration'],
+ video_item['time_published'],
+ video_item['is_time_published_exact'],
+ time_noticed,
+ video_item['description'],
+ ))
+
+ cursor.executemany('''INSERT OR IGNORE INTO videos (
+ sql_channel_id,
+ video_id,
+ title,
+ duration,
+ time_published,
+ is_time_published_exact,
+ time_noticed,
+ description
+ )
+ VALUES ((SELECT id FROM subscribed_channels WHERE yt_channel_id=?), ?, ?, ?, ?, ?, ?, ?)''', rows)
+ cursor.executemany('''UPDATE videos SET
+ title=?,
+ duration=?,
+ time_published=?,
+ is_time_published_exact=?,
+ description=?
+ WHERE video_id=?''', update_rows)
+ cursor.execute('''UPDATE subscribed_channels
+ SET time_last_checked = ?, next_check_time = ?
+ WHERE yt_channel_id=?''', [int(time.time()), next_check_time, channel_id])
+
+ if settings.autocheck_subscriptions:
+ if not _is_muted(cursor, channel_id):
+ autocheck_job_application.put({'channel_id': channel_id, 'channel_name': channel_names[channel_id], 'next_check_time': next_check_time})
+
+ if number_of_new_videos == 0:
+ print('No new videos from ' + channel_status_name)
+ elif number_of_new_videos == 1:
+ print('1 new video from ' + channel_status_name)
+ else:
+ print(str(number_of_new_videos) + ' new videos from ' + channel_status_name)
+
+
+def check_all_channels():
+ with open_database() as connection:
+ with connection as cursor:
+ channel_id_name_list = cursor.execute('''SELECT yt_channel_id, channel_name
+ FROM subscribed_channels
+ WHERE muted != 1''').fetchall()
+
+ channel_names.update(channel_id_name_list)
+ check_channels_if_necessary([item[0] for item in channel_id_name_list])
+
+
+def check_tags(tags):
+ channel_id_name_list = []
+ with open_database() as connection:
+ with connection as cursor:
+ for tag in tags:
+ channel_id_name_list += _channels_with_tag(cursor, tag, exclude_muted=True)
+
+ channel_names.update(channel_id_name_list)
+ check_channels_if_necessary([item[0] for item in channel_id_name_list])
+
+
+def check_specific_channels(channel_ids):
+ with open_database() as connection:
+ with connection as cursor:
+ channel_id_name_list = []
+ for channel_id in channel_ids:
+ channel_id_name_list += cursor.execute('''SELECT yt_channel_id, channel_name
+ FROM subscribed_channels
+ WHERE yt_channel_id=?''', [channel_id]).fetchall()
+ channel_names.update(channel_id_name_list)
+ check_channels_if_necessary(channel_ids)
+
+CHANNEL_ID_RE = re.compile(r'UC[-_\w]{22}')
+@yt_app.route('/import_subscriptions', methods=['POST'])
+def import_subscriptions():
+
+ # check if the post request has the file part
+ if 'subscriptions_file' not in request.files:
+ # flash('No file part')
+ return flask.redirect(util.URL_ORIGIN + request.full_path)
+ file = request.files['subscriptions_file']
+ # if user does not select file, browser also
+ # submit an empty part without filename
+ if file.filename == '':
+ # flash('No selected file')
+ return flask.redirect(util.URL_ORIGIN + request.full_path)
+
+ mime_type = file.mimetype
+
+ if mime_type == 'application/json':
+ info = file.read().decode('utf-8')
+ if info == '':
+ return '400 Bad Request: File is empty', 400
+ try:
+ info = json.loads(info)
+ except json.decoder.JSONDecodeError:
+ traceback.print_exc()
+ return '400 Bad Request: Invalid json file', 400
+
+ channels = []
+ try:
+ if 'app_version_int' in info: # NewPipe Format
+ for item in info['subscriptions']:
+ # Other service, such as SoundCloud
+ if item.get('service_id', 0) != 0:
+ continue
+ channel_url = item['url']
+ channel_id_match = CHANNEL_ID_RE.search(channel_url)
+ if channel_id_match:
+ channel_id = channel_id_match.group(0)
+ else:
+ print('WARNING: Could not find channel id in url',
+ channel_url)
+ continue
+ channels.append((channel_id, item['name']))
+ else: # Old Google Takeout format
+ for item in info:
+ snippet = item['snippet']
+ channel_id = snippet['resourceId']['channelId']
+ channels.append((channel_id, snippet['title']))
+ except (KeyError, IndexError):
+ traceback.print_exc()
+ return '400 Bad Request: Unknown json structure', 400
+ elif mime_type in ('application/xml', 'text/xml', 'text/x-opml'):
+ file = file.read().decode('utf-8')
+ try:
+ root = defusedxml.ElementTree.fromstring(file)
+ assert root.tag == 'opml'
+ channels = []
+ for outline_element in root[0][0]:
+ if (outline_element.tag != 'outline') or ('xmlUrl' not in outline_element.attrib):
+ continue
+
+ channel_name = outline_element.attrib['text']
+ channel_rss_url = outline_element.attrib['xmlUrl']
+ channel_id = channel_rss_url[channel_rss_url.find('channel_id=')+11:].strip()
+ channels.append((channel_id, channel_name))
+
+ except (AssertionError, IndexError, defusedxml.ElementTree.ParseError) as e:
+ return '400 Bad Request: Unable to read opml xml file, or the file is not the expected format', 400
+ elif mime_type in ('text/csv', 'application/vnd.ms-excel'):
+ content = file.read().decode('utf-8')
+ reader = csv.reader(content.splitlines())
+ channels = []
+ for row in reader:
+ if not row or row[0].lower().strip() == 'channel id':
+ continue
+ elif len(row) > 1 and CHANNEL_ID_RE.fullmatch(row[0].strip()):
+ channels.append( (row[0], row[-1]) )
+ else:
+ print('WARNING: Unknown row format:', row)
+ else:
+ error = 'Unsupported file format: ' + mime_type
+ error += (' . Only subscription.json, subscriptions.csv files'
+ ' (from Google Takeouts)'
+ ' and XML OPML files exported from YouTube\'s'
+ ' subscription manager page are supported')
+ return (flask.render_template('error.html', error_message=error),
+ 400)
+
+ _subscribe(channels)
+
+ return flask.redirect(util.URL_ORIGIN + '/subscription_manager', 303)
+
+
+@yt_app.route('/export_subscriptions', methods=['POST'])
+def export_subscriptions():
+ include_muted = request.values.get('include_muted') == 'on'
+ with open_database() as connection:
+ with connection as cursor:
+ sub_list = []
+ for channel_name, channel_id, muted in (
+ _get_subscribed_channels(cursor)):
+ if muted and not include_muted:
+ continue
+ if request.values['export_format'] == 'json_google_takeout':
+ sub_list.append({
+ 'kind': 'youtube#subscription',
+ 'snippet': {
+ 'muted': bool(muted),
+ 'resourceId': {
+ 'channelId': channel_id,
+ 'kind': 'youtube#channel',
+ },
+ 'tags': _get_tags(cursor, channel_id),
+ 'title': channel_name,
+ },
+ })
+ elif request.values['export_format'] == 'json_newpipe':
+ sub_list.append({
+ 'service_id': 0,
+ 'url': 'https://www.youtube.com/channel/' + channel_id,
+ 'name': channel_name,
+ })
+ elif request.values['export_format'] == 'opml':
+ sub_list.append({
+ 'channel_name': channel_name,
+ 'channel_id': channel_id,
+ })
+ date_time = time.strftime('%Y%m%d%H%M', time.localtime())
+ if request.values['export_format'] == 'json_google_takeout':
+ r = flask.Response(json.dumps(sub_list), mimetype='text/json')
+ cd = 'attachment; filename="subscriptions_%s.json"' % date_time
+ r.headers['Content-Disposition'] = cd
+ return r
+ elif request.values['export_format'] == 'json_newpipe':
+ r = flask.Response(json.dumps({
+ 'app_version': '0.21.9',
+ 'app_version_int': 975,
+ 'subscriptions': sub_list,
+ }), mimetype='text/json')
+ file_name = 'newpipe_subscriptions_%s_youtube-local.json' % date_time
+ cd = 'attachment; filename="%s"' % file_name
+ r.headers['Content-Disposition'] = cd
+ return r
+ elif request.values['export_format'] == 'opml':
+ r = flask.Response(
+ flask.render_template('subscriptions.xml', sub_list=sub_list),
+ mimetype='text/xml')
+ cd = 'attachment; filename="subscriptions_%s.xml"' % date_time
+ r.headers['Content-Disposition'] = cd
+ return r
+ else:
+ return '400 Bad Request', 400
+
+
+@yt_app.route('/subscription_manager', methods=['GET'])
+def get_subscription_manager_page():
+ group_by_tags = request.args.get('group_by_tags', '0') == '1'
+ with open_database() as connection:
+ with connection as cursor:
+ if group_by_tags:
+ tag_groups = []
+
+ for tag in _get_all_tags(cursor):
+ sub_list = []
+ for channel_id, channel_name, muted in _channels_with_tag(cursor, tag, order=True, include_muted_status=True):
+ sub_list.append({
+ 'channel_url': util.URL_ORIGIN + '/channel/' + channel_id,
+ 'channel_name': channel_name,
+ 'channel_id': channel_id,
+ 'muted': muted,
+ 'tags': [t for t in _get_tags(cursor, channel_id) if t != tag],
+ })
+
+ tag_groups.append((tag, sub_list))
+
+ # Channels with no tags
+ channel_list = cursor.execute('''SELECT yt_channel_id, channel_name, muted
+ FROM subscribed_channels
+ WHERE id NOT IN (
+ SELECT sql_channel_id FROM tag_associations
+ )
+ ORDER BY channel_name COLLATE NOCASE''').fetchall()
+ if channel_list:
+ sub_list = []
+ for channel_id, channel_name, muted in channel_list:
+ sub_list.append({
+ 'channel_url': util.URL_ORIGIN + '/channel/' + channel_id,
+ 'channel_name': channel_name,
+ 'channel_id': channel_id,
+ 'muted': muted,
+ 'tags': [],
+ })
+
+ tag_groups.append(('No tags', sub_list))
+ else:
+ sub_list = []
+ for channel_name, channel_id, muted in _get_subscribed_channels(cursor):
+ sub_list.append({
+ 'channel_url': util.URL_ORIGIN + '/channel/' + channel_id,
+ 'channel_name': channel_name,
+ 'channel_id': channel_id,
+ 'muted': muted,
+ 'tags': _get_tags(cursor, channel_id),
+ })
+
+ if group_by_tags:
+ return flask.render_template(
+ 'subscription_manager.html',
+ group_by_tags=True,
+ tag_groups=tag_groups,
+ )
+ else:
+ return flask.render_template(
+ 'subscription_manager.html',
+ group_by_tags=False,
+ sub_list=sub_list,
+ )
+
+
+def list_from_comma_separated_tags(string):
+ return [tag.strip() for tag in string.split(',') if tag.strip()]
+
+
+@yt_app.route('/subscription_manager', methods=['POST'])
+def post_subscription_manager_page():
+ action = request.values['action']
+
+ with open_database() as connection:
+ with connection as cursor:
+ if action == 'add_tags':
+ _add_tags(cursor, request.values.getlist('channel_ids'), [tag.lower() for tag in list_from_comma_separated_tags(request.values['tags'])])
+ elif action == 'remove_tags':
+ _remove_tags(cursor, request.values.getlist('channel_ids'), [tag.lower() for tag in list_from_comma_separated_tags(request.values['tags'])])
+ elif action == 'unsubscribe':
+ _unsubscribe(cursor, request.values.getlist('channel_ids'))
+ elif action == 'unsubscribe_verify':
+ unsubscribe_list = _get_channel_names(cursor, request.values.getlist('channel_ids'))
+ return flask.render_template('unsubscribe_verify.html', unsubscribe_list=unsubscribe_list)
+
+ elif action == 'mute':
+ cursor.executemany('''UPDATE subscribed_channels
+ SET muted = 1
+ WHERE yt_channel_id = ?''', [(ci,) for ci in request.values.getlist('channel_ids')])
+ elif action == 'unmute':
+ cursor.executemany('''UPDATE subscribed_channels
+ SET muted = 0
+ WHERE yt_channel_id = ?''', [(ci,) for ci in request.values.getlist('channel_ids')])
+ else:
+ flask.abort(400)
+
+ return flask.redirect(util.URL_ORIGIN + request.full_path, 303)
+
+
+@yt_app.route('/subscriptions', methods=['GET'])
+@yt_app.route('/feed/subscriptions', methods=['GET'])
+def get_subscriptions_page():
+ page = int(request.args.get('page', 1))
+ with open_database() as connection:
+ with connection as cursor:
+ tag = request.args.get('tag', None)
+ videos, number_of_videos_in_db = _get_videos(cursor, 60, (page - 1)*60, tag)
+ for video in videos:
+ video['thumbnail'] = util.URL_ORIGIN + '/data/subscription_thumbnails/' + video['id'] + '.jpg'
+ video['type'] = 'video'
+ video['item_size'] = 'small'
+ util.add_extra_html_info(video)
+
+ tags = _get_all_tags(cursor)
+
+ subscription_list = []
+ for channel_name, channel_id, muted in _get_subscribed_channels(cursor):
+ subscription_list.append({
+ 'channel_url': util.URL_ORIGIN + '/channel/' + channel_id,
+ 'channel_name': channel_name,
+ 'channel_id': channel_id,
+ 'muted': muted,
+ })
+
+ return flask.render_template(
+ 'subscriptions.html',
+ header_playlist_names=local_playlist.get_playlist_names(),
+ videos=videos,
+ num_pages=math.ceil(number_of_videos_in_db/60),
+ parameters_dictionary=request.args,
+ tags=tags,
+ current_tag=tag,
+ subscription_list=subscription_list,
+ )
+
+
+@yt_app.route('/subscriptions', methods=['POST'])
+@yt_app.route('/feed/subscriptions', methods=['POST'])
+def post_subscriptions_page():
+ action = request.values['action']
+ if action == 'subscribe':
+ if len(request.values.getlist('channel_id')) != len(request.values.getlist('channel_name')):
+ return '400 Bad Request, length of channel_id != length of channel_name', 400
+ _subscribe(zip(request.values.getlist('channel_id'), request.values.getlist('channel_name')))
+
+ elif action == 'unsubscribe':
+ with_open_db(_unsubscribe, request.values.getlist('channel_id'))
+
+ elif action == 'refresh':
+ type = request.values['type']
+ if type == 'all':
+ check_all_channels()
+ elif type == 'tag':
+ check_tags(request.values.getlist('tag_name'))
+ elif type == 'channel':
+ check_specific_channels(request.values.getlist('channel_id'))
+ else:
+ flask.abort(400)
+ else:
+ flask.abort(400)
+
+ return '', 204
+
+
+@yt_app.route('/data/subscription_thumbnails/<thumbnail>')
+def serve_subscription_thumbnail(thumbnail):
+ '''Serves thumbnail from disk if it's been saved already. If not, downloads the thumbnail, saves to disk, and serves it.'''
+ assert thumbnail[-4:] == '.jpg'
+ video_id = thumbnail[0:-4]
+ thumbnail_path = os.path.join(thumbnails_directory, thumbnail)
+
+ if video_id in existing_thumbnails:
+ try:
+ f = open(thumbnail_path, 'rb')
+ except FileNotFoundError:
+ existing_thumbnails.remove(video_id)
+ else:
+ image = f.read()
+ f.close()
+ return flask.Response(image, mimetype='image/jpeg')
+
+ url = f"https://i.ytimg.com/vi/{video_id}/hqdefault.jpg"
+ try:
+ image = util.fetch_url(url, report_text="Saved thumbnail: " + video_id)
+ except urllib.error.HTTPError as e:
+ print("Failed to download thumbnail for " + video_id + ": " + str(e))
+ abort(e.code)
+ try:
+ f = open(thumbnail_path, 'wb')
+ except FileNotFoundError:
+ os.makedirs(thumbnails_directory, exist_ok=True)
+ f = open(thumbnail_path, 'wb')
+ f.write(image)
+ f.close()
+ existing_thumbnails.add(video_id)
+
+ return flask.Response(image, mimetype='image/jpeg')