rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 19:01:39 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions
wrolpi/modules/videos/video/lib.py
lrnselfreliance 97a6566e3d
Fix API stalls: non-blocking switch worker, threaded searches, indexed Videos page (#678) 
* Fix API stalls: non-blocking switch worker, threaded searches, indexed Videos page

Three response-time fixes found by benchmarking the read-only API endpoints
hot and cold on real libraries:

* `switch_worker` runs on the Sanic worker's event loop every 0.1s and called
  the blocking `multiprocessing.Event.wait(timeout=1)`.  With no switch pending
  (the normal state) this froze the loop for a second at a time; about a third
  of all requests, including trivial ones, took ~1,000 ms.  It now polls
  `is_set()` and returns immediately.

* The search handlers (videos, files, archives, docs, zims) and `/api/statistics`
  ran synchronous SQLite queries on the event loop, so one slow search stalled
  every other request on that worker.  They now run via `asyncio.to_thread`.

* The unfiltered Videos page (`POST /api/videos/search` with only an order)
  used `mimetype LIKE 'video/%' OR ...` plus `COUNT(*) OVER()`, which visits
  every video's index entry before LIMIT applies: seconds cold on a large
  library on a spinning disk, and ~10 s hot for a deep page.  It now seeks each
  exact mimetype on the (mimetype, effective_datetime) index for `limit + offset`
  rows and merges them, with the total from a separate count cached for 60 s.
  Results are identical to the generic query.

Also adds `scripts/api_perf.py` (times every read-only endpoint against a host,
optionally restarting the API over SSH for cold numbers) and
`scripts/api_perf_compare.py` (merges per-host results).  Host profiles live in a
gitignored JSON file; an example is included.

Measured on an 8-core docker host with a 10 GB library: median hot request
1,007 ms -> 22 ms; Videos page 1 hot 287 -> 108 ms, page 50 1,345 -> 120 ms.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Serve the video description from its own endpoint instead of Video.__json__

`Video.__json__` computed `description` as `c_text or get_video_description()`,
where the fallback opens and parses the video's info json from disk.  The list
endpoints serialize every result with `__json__`, so any page containing videos
without an indexed description (common: many videos have none) read their info
json files on every request, cold, for a field the list never displays.  The
description is not part of search either: search runs against the FTS copy of
`c_text`, so the fallback never fed the index.

Drop `description` from `__json__` and serve it from
`GET /api/videos/<id>/description`, like comments and captions.  The endpoint
returns the indexed `c_text`, falling back to the info json only there, for a
row indexed before descriptions were stored.  The player page fetches it through
`useVideoExtras` alongside the comments.

Tests assert the list and detail responses never touch the info json, and cover
the endpoint's indexed, fallback, and not-found cases.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Address review: thread-local zim sessions, fast-path equivalence test, comments

* Zim search, single-zim search, and search estimates run libzim off the event
  loop, each opening its own session inside the worker thread rather than
  passing the request's SQLAlchemy Session across threads.
* The Videos fast-path test now runs the same requests through the generic
  query and asserts identical rows and paging for both date orders.  The one
  intentional difference is documented: past the end, the fast path still
  reports the true total.
* switch_worker returns before its try block so the empty path does not clear
  the event; its comment states the invariant only.  The copied "synchronous
  SQLite" comments are reduced to one line in one place.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-08-29 16:17:53 -06:00

418 lines
18 KiB
Python
Raw Blame History

import asyncio
import random
import time
from datetime import timedelta
from typing import Tuple, Optional, List, Dict, Callable
import yt_dlp
from sqlalchemy import or_
from sqlalchemy.orm import Session
from modules.videos.models import Video, Channel
from wrolpi import fts
from wrolpi.common import logger, limit_concurrent, wrol_mode_check
from wrolpi.dates import now
from wrolpi.db import get_db_session, get_db_curs
from wrolpi.downloader import download_manager
from wrolpi.files.lib import handle_file_group_search_results
from wrolpi.files.models import FileGroup
from wrolpi.tags import tag_append_sub_select_where
from wrolpi.vars import VIDEO_COMMENTS_FETCH_COUNT, YTDLP_CACHE_DIR, PYTEST
from ..cookies import cookies_unlocked, cookies_for_download
from ..lib import get_yt_dlp_http_headers, get_yt_dlp_sleep_opts
logger.getChild(__name__)
def get_video_for_app(file_group_id: int, skip_viewed: bool = False) -> Tuple[dict, Optional[dict], Optional[dict]]:
"""
Get a Video by its FileGroup ID, with its prev/next videos. Mark the Video as viewed.
"""
with get_db_session(commit=True) as session:
video = Video.find_by_file_group_id(session, file_group_id)
if not skip_viewed:
video.file_group.set_viewed()
previous_video, next_video = video.get_surrounding_videos()
video = video.__json__()
previous_video = previous_video.__json__() if previous_video and previous_video.file_group else None
next_video = next_video.__json__() if next_video and next_video.file_group else None
return video, previous_video, next_video
def get_video(file_group_id: int) -> Video:
"""
Get a Video by its FileGroup ID.
"""
with get_db_session() as session:
video = Video.find_by_file_group_id(session, file_group_id)
return video
VIDEO_ORDERS = {
# fg.id (the rowid) is the pagination tiebreaker: it is present in every index, so these
# orders stay index-only. LOWER(fg.primary_path) forced SQLite to read every (large) row.
# Sometimes we don't have a published_datetime. This is equivalent to COALESCE(fg.published_datetime, fg.download_datetime)
'published_datetime': 'fg.effective_datetime ASC, fg.id ASC',
'-published_datetime': 'fg.effective_datetime DESC NULLS LAST, fg.id ASC',
'rank': '2 DESC, fg.id DESC',
'-rank': '2 ASC, fg.id ASC',
'size': 'fg.size ASC, fg.id ASC',
'-size': 'fg.size DESC, fg.id DESC',
'length': 'fg.length ASC, fg.id ASC',
'-length': 'fg.length DESC, fg.id DESC',
'size_to_duration': 'CASE WHEN fg.length > 0 THEN CAST(fg.size AS REAL) / fg.length ELSE fg.size END ASC, fg.id ASC',
'-size_to_duration': 'CASE WHEN fg.length > 0 THEN CAST(fg.size AS REAL) / fg.length ELSE fg.size END DESC, fg.id DESC',
'viewed': 'fg.viewed ASC',
'-viewed': 'fg.viewed DESC',
'view_count': 'v.view_count ASC',
'-view_count': 'v.view_count DESC',
'download_datetime': 'fg.download_datetime ASC, fg.id ASC',
'-download_datetime': 'fg.download_datetime DESC NULLS LAST, fg.id ASC',
}
NO_NULL_ORDERS = {
'viewed': 'fg.viewed IS NOT NULL',
'-viewed': 'fg.viewed IS NOT NULL',
'length': 'fg.length IS NOT NULL',
'-length': 'fg.length IS NOT NULL',
'size': 'fg.size IS NOT NULL',
'-size': 'fg.size IS NOT NULL',
'size_to_duration': 'fg.size IS NOT NULL AND fg.length IS NOT NULL',
'-size_to_duration': 'fg.size IS NOT NULL AND fg.length IS NOT NULL',
'view_count': 'v.view_count IS NOT NULL',
'-view_count': 'v.view_count IS NOT NULL',
}
JOIN_ORDERS = ('view_count', '-view_count')
DEFAULT_VIDEO_ORDER = 'rank'
VIDEO_QUERY_LIMIT = 24
# Orders the unfiltered Videos page can serve straight from the (mimetype, effective_datetime) index.
INDEXED_DATE_ORDERS = ('published_datetime', '-published_datetime')
VIDEO_MIMETYPE_PREFIXES = ('video/', 'audio/')
# The unfiltered Videos page total, cached for a short time. Counting every video walks the whole
# mimetype index, which costs seconds cold on a large library on a spinning disk; pagination can
# tolerate a total that is a minute stale. Bypassed under pytest so tests see exact totals.
TOTALS_CACHE_SECONDS = 60
_TOTALS_CACHE: Dict[str, Tuple[float, int]] = dict()
def _cached_total(key: str, compute: Callable[[], int]) -> int:
if not PYTEST:
hit = _TOTALS_CACHE.get(key)
if hit and time.monotonic() - hit[0] < TOTALS_CACHE_SECONDS:
return hit[1]
total = compute()
_TOTALS_CACHE[key] = (time.monotonic(), total)
return total
def _video_mimetypes() -> List[str]:
"""Every distinct video/* and audio/* mimetype in the library (a handful; served by the mimetype index)."""
mimetypes = []
with get_db_curs() as curs:
for prefix in VIDEO_MIMETYPE_PREFIXES:
# '0' is the character after '/', so this is the half-open range of the prefix.
curs.execute('SELECT DISTINCT mimetype FROM file_group WHERE mimetype >= ? AND mimetype < ?',
(prefix, prefix[:-1] + '0'))
mimetypes.extend(i[0] for i in curs.fetchall())
return mimetypes
def _count_videos() -> int:
ranges = ' OR '.join(f'(mimetype >= :p{i} AND mimetype < :e{i})' for i in range(len(VIDEO_MIMETYPE_PREFIXES)))
params = dict()
for i, prefix in enumerate(VIDEO_MIMETYPE_PREFIXES):
params[f'p{i}'], params[f'e{i}'] = prefix, prefix[:-1] + '0'
with get_db_curs() as curs:
curs.execute(f'SELECT COUNT(*) FROM file_group WHERE {ranges}', params)
return curs.fetchone()[0]
def _search_videos_by_date(order: str, limit: int, offset: int) -> Tuple[List[dict], int]:
"""The unfiltered Videos page, ordered by date, without visiting every video.
The generic query (`mimetype LIKE 'video/%' OR ...` + `COUNT(*) OVER()`) must visit every
video's index entry before it can apply LIMIT, which scales with the library rather than the
page. Here each exact mimetype is a pure covering-index seek on (mimetype, effective_datetime)
that reads only `limit + offset` entries; those few rows are then merged. Results must be
identical to the generic query's (same ORDER BY, including the id tiebreaker and NULL placement).
The total is a separate, cached count."""
order_by = VIDEO_ORDERS[order]
mimetypes = _video_mimetypes()
if not mimetypes:
return [], 0
n = limit + offset
params = {f'mt{i}': mimetype for i, mimetype in enumerate(mimetypes)}
params['offset'] = offset
members = '\n UNION ALL\n'.join(
f'SELECT * FROM (SELECT fg.id, fg.effective_datetime FROM file_group fg'
f' WHERE fg.mimetype = :mt{i} ORDER BY {order_by} LIMIT {n})'
for i in range(len(mimetypes))
)
stmt = f'''
SELECT fg.id
FROM ({members}) fg
ORDER BY {order_by}
LIMIT {limit} OFFSET :offset
'''.strip()
logger.debug(f'{stmt} {params}')
total = _cached_total('videos', _count_videos)
return handle_file_group_search_results(stmt, params, total=total)
def search_videos(
search_str: str = None,
offset: int = None,
limit: int = VIDEO_QUERY_LIMIT,
channel_id: int = None,
order: str = None,
tag_names: List[str] = None,
headline: bool = False,
censored: bool = False,
deep: bool = False,
) -> Tuple[List[dict], int]:
tag_names = tag_names or []
# Search videos and audio-only files.
wheres = ["(fg.mimetype LIKE 'video/%' OR fg.mimetype LIKE 'audio/%')"]
if censored:
# Only return videos which are no longer available for download.
wheres.append('fg.censored = TRUE')
joins = list()
join_video = False
# Explicit JSON nulls bypass the schema defaults; SQLite rejects LIMIT/OFFSET NULL.
limit = int(limit) if limit else 20
offset = int(offset) if offset else 0
if not search_str and not tag_names and not channel_id and not censored and order in INDEXED_DATE_ORDERS:
# The Videos page with no filters, in date order.
return _search_videos_by_date(order, limit, offset)
params = dict(search_str=search_str, offset=offset)
if channel_id:
wheres.append('v.channel_id = :channel_id')
joins.append('LEFT JOIN channel c ON c.id = v.channel_id')
join_video = True
params['channel_id'] = channel_id
# `deep` searches d_text (captions) as well; the default abc search is much smaller and faster.
# The subquery join is required: snippet() cannot coexist with COUNT(*) OVER() in one SELECT.
fts_search = fts.file_group_search_join(search_str, deep=deep, headlines=bool(headline)) \
if search_str else None
if fts_search:
# A search_str was provided by the user, modify the query to filter by it.
select_columns = f'fg.id, {fts_search.rank_select}, COUNT(*) OVER() AS total'
joins.append(fts_search.join)
params.update(fts_search.params)
else:
# No (usable) search_str provided. Get id and total only.
select_columns = 'fg.id, COUNT(*) OVER() AS total'
wheres, params = tag_append_sub_select_where(wheres, params, tag_names)
if fts_search and headline:
# The title headline is computed in Python by `handle_file_group_search_results`.
headline_selects = ', fts.b_headline, fts.c_headline, fts.d_headline'
else:
headline_selects = ''
# Convert the user-friendly order by into a real order by, restrict what can be interpolated by using the
# whitelist.
order_by = VIDEO_ORDERS[DEFAULT_VIDEO_ORDER]
if order:
try:
order_by = VIDEO_ORDERS[order]
except KeyError:
raise
if order in NO_NULL_ORDERS:
wheres.append(NO_NULL_ORDERS[order])
if order in JOIN_ORDERS:
join_video = True
if join_video:
joins.insert(0, 'LEFT JOIN video v on v.file_group_id = fg.id')
wheres = '\n AND '.join(wheres)
where = f'WHERE\n{wheres}' if wheres else ''
join = '\n'.join(joins)
stmt = f'''
SELECT
{select_columns}
{headline_selects}
FROM file_group fg
{join}
{where}
ORDER BY {order_by}
LIMIT {int(limit)} OFFSET :offset
'''.strip()
logger.debug(f'{stmt} {params}')
results, total = handle_file_group_search_results(stmt, params)
return results, total
def download_video_info_json(url: str) -> dict:
"""Download video info JSON, using encrypted cookies for authentication if available."""
ydl_opts = dict(
getcomments=True,
skip_download=True,
extractor_args={'youtube': {'max_comments': ['all', '20', 'all', '10'], 'comment_sort': ['top']}},
cachedir=YTDLP_CACHE_DIR,
**get_yt_dlp_sleep_opts(),
)
ydl_logger = logger.getChild('youtube-dl')
if cookies_unlocked():
logger.info(f'Using encrypted cookies for video info: {url}')
with cookies_for_download() as cookies_path:
ydl_opts['cookiefile'] = str(cookies_path)
http_headers = get_yt_dlp_http_headers()
if http_headers:
ydl_opts['http_headers'] = http_headers
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
ydl.params['logger'] = ydl_logger
info = ydl.extract_info(url, download=False)
ydl.sanitize_info(info)
return info
logger.info(f'No cookies available for video info: {url}')
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
ydl.params['logger'] = ydl_logger
info = ydl.extract_info(url, download=False)
ydl.sanitize_info(info)
return info
@limit_concurrent(1)
@wrol_mode_check
async def get_missing_videos_comments(limit: int = VIDEO_COMMENTS_FETCH_COUNT):
"""
Fetches Video info json for videos without comments, if comments are found then the info json file is replaced.
Uses encrypted cookies for authentication if available.
"""
one_month_ago = now() - timedelta(days=30)
# Get any videos over a month old that do not have comments.
with get_db_session() as session:
videos = session.query(Video).filter(
# Have yet to get comments.
Video.have_comments != True,
Video.comments_failed != True,
# Needs to be downloadable.
FileGroup.url != None, # noqa
FileGroup.censored != True,
# We want old videos (time for comments to accumulate), or those which we don't know the published date.
or_(
one_month_ago > FileGroup.published_datetime,
FileGroup.published_datetime == None,
),
# Do not download a Videos with a Channel and Channel.download_missing_data is False.
or_(
Channel.download_missing_data == True,
Channel.id == None,
)
).join(FileGroup) \
.outerjoin(Channel) \
.order_by(FileGroup.published_datetime.nullsfirst()) \
.limit(limit)
video_urls = [i.file_group.url for i in videos]
logger.info(f'Found {len(video_urls)} videos missing comments')
def add_video_to_skip_list(video_url: str, censored: bool = False):
"""Do not attempt to get comments of the provided Video again."""
with get_db_session(commit=True) as session_:
for video_, fg in session_.query(Video, FileGroup).join(FileGroup).filter_by(url=video_url).all():
video_.comments_failed = True
fg.censored = censored
download_manager.add_to_skip_list(video_url)
logger.debug(f'add_video_to_skip_list: {video_url=} {censored=}')
for url in video_urls:
# Sleep to catch cancel.
await asyncio.sleep(0)
if download_manager.is_skipped(url):
# This video is skipped, do not try to get comments again.
add_video_to_skip_list(url)
if download_manager.is_disabled or download_manager.is_stopped:
logger.info('Downloads disabled, stop getting comments...')
return
with get_db_session(commit=True) as session:
have_comments = False
for video in session.query(Video).join(FileGroup).filter_by(url=url).all():
if (comments := video.get_comments()) and len(comments) >= 5:
# We already have enough comments. DB may have been wiped, we don't want to fetch comments again.
logger.info(f'Already have comments for Video: {url=}')
have_comments = video.have_comments = True
if have_comments:
download_manager.add_to_skip_list(url)
continue
logger.info(f'Getting comments for: {url}')
try:
# Get info json about the video.
info = download_video_info_json(url)
if not info or not isinstance(info.get('comments'), list):
logger.error(f'Unable to get comments for video: {url=}')
add_video_to_skip_list(url)
continue
except Exception as e:
if 'no longer available' in str(e):
logger.error(f'Giving up on downloading comments because video does not exist: {url=}', exc_info=e)
add_video_to_skip_list(url, censored=True)
elif 'Comments are turned off' in str(e):
logger.error(f'Giving up on downloading comments because comments are disabled: {url=}', exc_info=e)
add_video_to_skip_list(url)
elif 'Private video' in str(e):
logger.error(f'Giving up on downloading comments because video is private: {url=}', exc_info=e)
add_video_to_skip_list(url, censored=True)
elif 'Video unavailable' in str(e):
logger.error(f'Giving up on downloading comments because video is unavailable: {url=}', exc_info=e)
add_video_to_skip_list(url, censored=True)
elif 'video has been removed' in str(e):
logger.error(f'Giving up on downloading comments because video is removed: {url=}', exc_info=e)
add_video_to_skip_list(url, censored=True)
elif ' members' in str(e):
logger.error(f'Giving up on downloading comments because video is members-only: {url=}', exc_info=e)
add_video_to_skip_list(url, censored=True)
elif 'rumble.com' in url and "'bool' object has no attribute 'get'" in str(e):
logger.error(f'Rumble error in yt-dlp', exc_info=e)
add_video_to_skip_list(url, censored=False)
elif 'not a valid URL' in str(e):
logger.error(f'URL is not valid: {url=}', exc_info=e)
add_video_to_skip_list(url, censored=True)
elif ' Sign in ' in str(e):
logger.error(f'Video requires sign-in: {url=}', exc_info=e)
add_video_to_skip_list(url, censored=True)
else:
logger.error(f'Got error when attempting to download video comments: {url=}', exc_info=e)
continue
logger.info(f'Got {len(info["comments"])} comments for Video: {url=}')
with get_db_session(commit=True) as session:
videos = list(session.query(Video).join(FileGroup).filter_by(url=url).all())
for video in videos:
if video.video_path.is_file():
video.replace_info_json(info)
video.have_comments = True
video.comments_failed = False
else:
logger.error(f'Attempting to replace comments for non-existent video! {video}')
if len(videos) == 1:
download_manager.add_to_skip_list(url)
video.comments_failed = True
# Sleep a random amount of time, so we don't spam.
await asyncio.sleep(random.randint(2, 20))
Reference in a new issue View git blame Copy permalink