app/browse/__init__.py

   1 import requests
   2 from flask import Blueprint, render_template, request, flash, g, url_for, redirect
   3 from flask_login import current_user
   4 from werkzeug.exceptions import BadRequest, NotFound
   5
   6 from ..common.common import *
   7 from ..common.innertube import *
   8 from .lib import *
   9 from .protobuf import make_sp, make_channel_params, make_playlist_params, Filters
  10
  11 frontend = Blueprint('browse', __name__,
  12     template_folder='templates',
  13     static_folder='static',
  14     static_url_path='/static/ys')
  15
  16 @frontend.route('/results')
  17 @frontend.route('/search')
  18 def search():
  19     #token = getattr(current_user, 'token', 'guest')
  20     q = request.args.get('q') or request.args.get('search_query')
  21     continuation = request.args.get('continuation')
  22
  23     sp = make_sp(**{
  24         k:v for k,v in request.args.items()
  25         if k in ['sort','date','type','len']
  26     }, features=[
  27         f for f in request.args.getlist('feature')
  28         if f in Filters.__dataclass_fields__.keys()
  29     ], extras=[
  30         e for e in request.args.getlist('feature')
  31         if e in ['verbatim']
  32     ])
  33
  34     if continuation or q:
  35         yt_results = fetch_ajax("search", **(
  36             {'continuation': continuation} if continuation else {'query': q, 'params': sp}
  37         ))
  38
  39         results, extras, continuation = prepare_searchresults(yt_results)
  40
  41         for extra in extras:
  42             flash(extra, 'info')
  43     else:
  44         results = None
  45
  46     return render_template('search.html.j2', rows=results, query=q, continuation=continuation)
  47
  48 @frontend.route('/channel/<channel_id>/')
  49 @frontend.route('/channel/<channel_id>/<subpage>')
  50 def channel(channel_id, subpage="videos"):
  51     token = getattr(current_user, 'token', 'guest')
  52     if subpage == "videos":
  53         sort_by = request.args.get('sort') or "newest"
  54         query = None
  55     elif subpage == "playlists":
  56         sort_by = request.args.get('sort', "modified")
  57         query = None
  58     elif subpage == "search":
  59         query = request.args.get('q')
  60         sort_by = None
  61     else: # we don't support /home, /about, ..., so redirect to /videos.
  62         return redirect(url_for('.channel', channel_id=channel_id))
  63
  64     # best effort; if it fails, it fails in the redirect.
  65     if not re.match(r"(UC[A-Za-z0-9_-]{22})", channel_id):
  66         return redirect(url_for('.channel_redirect', user=channel_id))
  67
  68     # if we don't have a continuation, we create parameters for page 1 manually:
  69     continuation = request.args.get('continuation') or \
  70         make_channel_params(channel_id, subpage, 1, sort_by, query, v3=(subpage != "search"))
  71     result = fetch_ajax("browse", continuation=continuation)
  72     error = find_and_parse_error(result)
  73
  74     if result is None: # if fetching from innertube failed, fall back to xmlfeed:
  75         flash("unable to fetch results from ajax; displaying fallback results (15 newest)", "error")
  76         return fallback_route(channel_id, subpage)
  77
  78     if error:
  79         return error, 400 # todo: ugly
  80
  81     title, descr, thumb, rows, more = prepare_channel(result, channel_id)
  82     if not rows: # overran end of list, or is special channel (e.g. music topic)
  83         flash("ajax returned nothing; displaying fallback results (15 newest)", "error")
  84         return fallback_route(channel_id, subpage)
  85
  86     # set pin/hide stati of retrieved videos:
  87     video_ids = [card['content']['video_id'] for card in rows]
  88     pinned, hidden = fetch_video_flags(token, video_ids)
  89     rows = sorted([
  90         {'type':v['type'], 'content':{**v['content'], 'pinned': v['content']['video_id'] in pinned}}
  91         for v in rows
  92         if v['content']['video_id'] not in hidden
  93     ], key=lambda v:v['content']['pinned'], reverse=True)
  94
  95     with sqlite3.connect(cf['global']['database']) as conn:
  96         c = conn.cursor()
  97         c.execute("""
  98             SELECT COUNT(*)
  99               FROM subscriptions
 100              WHERE channel_id = ? AND user = ?
 101         """, (channel_id, token))
 102         (is_subscribed,) = c.fetchone()
 103
 104     return render_template('channel.html.j2',
 105         title=title,
 106         subpage=subpage,
 107         rows=rows,
 108         channel_id=channel_id,
 109         channel_img=thumb,
 110         channel_desc=descr,
 111         is_subscribed=is_subscribed,
 112         has_more=more)
 113
 114 @frontend.route('/user/<user>/')
 115 @frontend.route('/user/<user>/<subpage>')
 116 @frontend.route('/c/<user>/')
 117 @frontend.route('/c/<user>/<subpage>')
 118 def channel_redirect(user, subpage=None):
 119     """
 120     The browse_ajax 'API' needs the UCID.
 121     """
 122
 123     # inverse of the test in /channel/:
 124     if re.match(r"(UC[A-Za-z0-9_-]{22})", user):
 125         return redirect(url_for('.channel', channel_id=user))
 126
 127     channel_id = canonicalize_channel(user)
 128     if not channel_id:
 129         raise NotFound("channel appears to not exist")
 130     return redirect(
 131         url_for('.channel', channel_id=channel_id, subpage=subpage), 308
 132     )
 133
 134 @frontend.route('/playlist')
 135 def playlist():
 136     playlist_id = request.args.get('list')
 137     if not playlist_id:
 138         raise BadRequest("No playlist ID")
 139
 140     xmlfeed = fetch_xml("playlist_id", playlist_id)
 141     if not xmlfeed:
 142         raise NotFound("Unable to fetch playlist")
 143     title, author, _, channel_id, _ = parse_xml(xmlfeed)
 144
 145     # if we don't have a continuation, we create parameters for page 1 manually:
 146     continuation = request.args.get('continuation') or \
 147         make_playlist_params(playlist_id, 0)
 148     result = fetch_ajax("browse", continuation=continuation)
 149     error = find_and_parse_error(result)
 150
 151     if result is None:
 152         flash(f"1 {error}. Loading fallback.", 'error')
 153         return fallback_route()
 154
 155     if not 'continuationContents' in result:
 156         flash(f"2 {error}. Loading fallback.", 'error')
 157         return fallback_route()
 158
 159     rows, more = prepare_playlist(result)
 160
 161     return render_template('playlist.html.j2',
 162         title=title,
 163         author=author,
 164         channel_id=channel_id,
 165         rows=rows,
 166         has_more=more)
 167
 168 @frontend.route('/<something>', strict_slashes=False)
 169 def plain_user_or_video(something):
 170     # this is a near-copy of the same route in app/youtube, but using a
 171     # different, more reliable endpoint to determine whether a channel exists.
 172     if '.' in something:
 173         # prevent a lot of false-positives (and reduce youtube api calls)
 174         raise NotFound
 175
 176     channel_id = canonicalize_channel(something)
 177     if channel_id:
 178         return redirect(url_for('.channel', channel_id=channel_id))
 179     elif re.match(r"^[-_0-9A-Za-z]{11}$", something): # looks like a video id
 180         return redirect(url_for('youtube.watch', v=something, t=request.args.get('t')))
 181     else: # ¯\_(ツ)_/¯
 182         raise NotFound("Note: some usernames not recognized; try searching it")
 183
 184 @frontend.before_app_request
 185 def inject_button():
 186     if not 'header_items' in g:
 187         g.header_items = []
 188     g.header_items.append({
 189         'name': 'search',
 190         'url': url_for('browse.search'),
 191         'parent': frontend.name,
 192         'priority': 15,
 193     })