peter
b04b51c909
A related list's cursor cost nothing, and a cursor is a free string: a replayed one read and enriched without end. Every request now takes a slot of the user's daily cap — taken under a lock before the work (`quota.claim_action`), so requests sent together cannot all pass the last free one, and given back when the request fails outright (`quota.refund_action`). The live search's scrape source takes its slots the same way. A limit of 0 turns the feature off: a distinct 409 `related_disabled` / `search_disabled`, and `/api/me` says `related_enabled` / `search_enabled` so the UI can leave it out. A videos.list that fails no longer reads as an empty page: `ingest_candidates` raises, and the search and the related list answer the refusal (the shared budget spent says "quota") instead of "no results". The two routes' page loops are one (`_gather`). `clear` deletes only what the caller found: a related list's rows carry no find, and any user could name them.
245 lines
11 KiB
Python
245 lines
11 KiB
Python
"""Live YouTube search, the catalog's side: what a page of results leaves in the shared catalog.
|
|
|
|
A result becomes a catalog row so it renders with the normal feed cards and the in-app player and
|
|
gains per-user state. The ingest runs in steps with a YouTube call between each, and each step's
|
|
rows are committed before the next call: an enrich that fails rolls back to the stubs, which stay
|
|
(the FK targets for what follows, backfilled later). Results are flagged (`videos.via_search`,
|
|
`channels.from_search`) so the Library can hide this search-discovered noise; Shorts and
|
|
live/upcoming videos are never kept. The HTTP layer (`routes/search.py`) runs the paging, the caps
|
|
and the refusals.
|
|
"""
|
|
|
|
from collections.abc import Sequence
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import delete, func, select, update
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
from sqlalchemy.engine import CursorResult
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import logs, quota, sysconfig
|
|
from app.catalog import queries
|
|
from app.catalog.models import LIVE_OR_UPCOMING, Channel, SearchFind, Subscription, Video
|
|
from app.models import User
|
|
from app.reader import queries as reader_queries
|
|
from app.sync.subscriptions import apply_channel_details
|
|
from app.sync.videos import apply_video_details, insert_stubs, parse_dt
|
|
from app.userscope import owned_by
|
|
from app.utils import like_escape
|
|
from app.youtube.client import YouTubeClient, YouTubeError
|
|
from app.youtube.shorts import make_client, probe_is_short
|
|
|
|
log = logs.get_logger(__name__)
|
|
|
|
|
|
def classify_shorts(db: Session, videos: Sequence[Video]) -> None:
|
|
"""Confirm/deny Shorts for the freshly-ingested videos so none slip into the feed.
|
|
|
|
Mirrors the scheduler's run_shorts_classification, but scoped to just these rows and run
|
|
synchronously (the search is an explicit, awaited action). Non-candidates (too long, no
|
|
duration, or live) are finalised without a probe; the rest hit youtube.com/shorts (no
|
|
quota). Undetermined probes are left for the scheduler to retry."""
|
|
probe_max = sysconfig.get_int(db, "shorts_probe_max_seconds")
|
|
to_probe: list[Video] = []
|
|
for v in videos:
|
|
if v.shorts_probed:
|
|
continue
|
|
if v.enriched_at is None:
|
|
# Enrichment failed (proxy blip) or YouTube omitted the row — duration is unknown,
|
|
# so DON'T finalise shorts_probed off a null duration; leave it for the scheduler
|
|
# (which only classifies enriched rows) to retry. Otherwise a real Short would leak.
|
|
continue
|
|
if v.live_status != "none" or v.duration_seconds is None or v.duration_seconds > probe_max:
|
|
v.is_short = False
|
|
v.shorts_probed = True
|
|
else:
|
|
to_probe.append(v)
|
|
if to_probe:
|
|
with make_client() as client:
|
|
|
|
def work(v: Video):
|
|
return v, probe_is_short(client, v.id)
|
|
|
|
with ThreadPoolExecutor(max_workers=16) as pool:
|
|
for v, result in pool.map(work, to_probe):
|
|
if result is None:
|
|
continue
|
|
v.is_short = result
|
|
v.shorts_probed = True
|
|
db.commit()
|
|
|
|
|
|
def ingest_candidates(
|
|
db: Session, yt: YouTubeClient, user: User, candidates: list[dict]
|
|
) -> list[str]:
|
|
"""Materialise one page of search candidates into the catalog (channel stubs + enrich,
|
|
video stubs flagged via_search, enrich, Shorts classification) and return the KEEPABLE
|
|
video ids — non-Short, non-live — in the input relevance order. A videos.list that fails
|
|
raises its YouTubeError: with nothing enriched, nothing could be kept."""
|
|
ids = [it["id"] for it in candidates]
|
|
if not ids:
|
|
return []
|
|
|
|
# 1) Channels are the FK target — create stubs for unknown ones (flagged from_search),
|
|
# then enrich the new ones (channels.list, 1 unit). Stubs survive an enrich failure.
|
|
channel_ids = list({it["channel_id"] for it in candidates})
|
|
existing = set(db.scalars(select(Channel.id).where(Channel.id.in_(channel_ids))).all())
|
|
new_channel_ids = [c for c in channel_ids if c not in existing]
|
|
if new_channel_ids:
|
|
titles = {it["channel_id"]: it["channel_title"] for it in candidates}
|
|
for cid in new_channel_ids:
|
|
db.add(Channel(id=cid, title=titles.get(cid), from_search=True))
|
|
db.commit()
|
|
try:
|
|
with quota.attribute(user.id, quota.QuotaAction.CHANNELS_DISCOVER):
|
|
details = yt.get_channels(new_channel_ids)
|
|
apply_channel_details(db, details, new_channel_ids)
|
|
db.commit()
|
|
except YouTubeError:
|
|
# The stubs remain and the details can be backfilled later — but until they are, those
|
|
# channels sit in the results with no thumbnail and no subscriber count, looking like
|
|
# channels that have neither.
|
|
log.warning(
|
|
"the channels found by a search could not be enriched",
|
|
extra=logs.event("search.channel_enrich_failed", channels=len(new_channel_ids)),
|
|
)
|
|
db.rollback() # details can be backfilled later; the stubs remain
|
|
|
|
# 2) Video stubs, flagged via_search. Existing rows keep their flags (ON CONFLICT DO
|
|
# NOTHING), so an organically-synced video stays organic.
|
|
# published_at is NOT NULL (migration 0068); the scrape source exposes no publish date
|
|
# (only a relative "3 years ago"), so fall back to now() to satisfy the constraint — the
|
|
# enrich in step 3 overwrites it with the video's real snippet date.
|
|
now = datetime.now(timezone.utc)
|
|
insert_stubs(
|
|
db,
|
|
[
|
|
{
|
|
"id": it["id"],
|
|
"channel_id": it["channel_id"],
|
|
"title": it["title"],
|
|
"published_at": parse_dt(it["published_at"]) or now,
|
|
"thumbnail_url": it["thumbnail_url"],
|
|
"via_search": True,
|
|
}
|
|
for it in candidates
|
|
],
|
|
)
|
|
|
|
# 3) Enrich (videos.list, 1 unit/50) so cards have duration/stats and we can judge live.
|
|
videos = db.scalars(select(Video).where(Video.id.in_(ids))).all()
|
|
try:
|
|
with quota.attribute(user.id, quota.QuotaAction.VIDEOS_ENRICH):
|
|
detail_by_id = {it["id"]: it for it in yt.get_videos(ids)}
|
|
except YouTubeError:
|
|
# An unenriched video cannot be judged live or Short, so none of this page could be kept:
|
|
# swallowed, the failure answered as a page with nothing on it — "no results" for a search
|
|
# that never ran (the shared budget spent, no API key). The stubs stay (`insert_stubs`
|
|
# committed them), as the channels' do above; the caller answers the failure (its refusal
|
|
# says which).
|
|
log.warning(
|
|
"the videos found by a search could not be enriched",
|
|
extra=logs.event("search.video_enrich_failed", candidates=len(ids)),
|
|
)
|
|
raise
|
|
enriched_at = datetime.now(
|
|
timezone.utc
|
|
) # when THIS enrich ran (distinct from the stub placeholder)
|
|
for v in videos:
|
|
item = detail_by_id.get(v.id)
|
|
if item is None:
|
|
continue
|
|
apply_video_details(v, item)
|
|
v.enriched_at = enriched_at
|
|
db.commit()
|
|
|
|
# 4) Confirm/deny Shorts (no quota) so none enter via search.
|
|
classify_shorts(db, videos)
|
|
|
|
keep = {
|
|
v.id
|
|
for v in videos
|
|
if v.enriched_at is not None and not v.is_short and v.live_status not in LIVE_OR_UPCOMING
|
|
}
|
|
return [vid for vid in ids if vid in keep]
|
|
|
|
|
|
def _fold_term(db: Session, video_ids: list[str], term: str) -> None:
|
|
"""Append `term` to each video's space-joined `search_terms`, unless it is already one of them.
|
|
|
|
"Already one of them" is a whole-term match, as a literal: a substring test skipped `cat` on a
|
|
video already found for `category`, and an unescaped one read the `_` in `my_channel` as a
|
|
wildcard — either way the term was never folded in, and local search lost that relevance."""
|
|
padded = func.concat(" ", func.coalesce(Video.search_terms, ""), " ")
|
|
db.execute(
|
|
update(Video)
|
|
.where(Video.id.in_(video_ids))
|
|
.where(padded.notilike(f"% {like_escape(term)} %"))
|
|
.values(
|
|
search_terms=func.trim(
|
|
func.coalesce(Video.search_terms, "").op("||")(" ").op("||")(term)
|
|
)
|
|
)
|
|
)
|
|
|
|
|
|
def new_first(db: Session, user: User, ids: list[str]) -> list[str]:
|
|
"""Reorder results so genuinely-new discoveries come first and things the user already has
|
|
(a subscribed channel, or a video they've watched / in-progress / saved / playlisted) sink
|
|
to the back — keeping the relevance order within each group."""
|
|
if not ids:
|
|
return ids
|
|
have = set(
|
|
db.scalars(
|
|
select(Video.id).where(
|
|
Video.id.in_(ids), Video.channel_id.in_(queries.subscribed_channels(user))
|
|
)
|
|
)
|
|
) | reader_queries.held_by(db, user, ids)
|
|
return [i for i in ids if i not in have] + [i for i in ids if i in have]
|
|
|
|
|
|
def record_finds(db: Session, user: User, video_ids: list[str], term: str) -> None:
|
|
"""These results as THIS user's search finds (the Mine feed, the Source filter), and the term
|
|
folded into each one's `search_terms` so local full-text search inherits YouTube's relevance.
|
|
Idempotent."""
|
|
db.execute(
|
|
pg_insert(SearchFind)
|
|
.values([{"user_id": user.id, "video_id": vid} for vid in video_ids])
|
|
.on_conflict_do_nothing(index_elements=["user_id", "video_id"])
|
|
)
|
|
_fold_term(db, video_ids, term)
|
|
|
|
|
|
def clear(db: Session, user: User, video_ids: list[str]) -> int:
|
|
"""Discard these results "as if never added": drop this user's finds of them, then delete the
|
|
ones nobody kept — a `via_search` video no reader holds (`reader.queries.kept_by_a_reader`),
|
|
whose channel nobody subscribes to, and that no other user still has as a find. Returns how
|
|
many videos went.
|
|
|
|
Only the ones this user found: a video in the catalog with no find at all (a related list's,
|
|
which records none) is in someone's list right now, and an id any user could name would let
|
|
them delete it from under that list."""
|
|
mine = list(
|
|
db.scalars(
|
|
delete(SearchFind)
|
|
.where(owned_by(SearchFind.user_id, user), SearchFind.video_id.in_(video_ids))
|
|
.returning(SearchFind.video_id)
|
|
)
|
|
)
|
|
if not mine:
|
|
return 0
|
|
subscribed = select(Subscription.id).where(Subscription.channel_id == Video.channel_id).exists()
|
|
other_find = select(SearchFind.id).where(SearchFind.video_id == Video.id).exists()
|
|
result = db.execute(
|
|
delete(Video).where(
|
|
Video.id.in_(mine),
|
|
Video.via_search.is_(True),
|
|
~reader_queries.kept_by_a_reader(Video.id),
|
|
~subscribed,
|
|
~other_find,
|
|
)
|
|
)
|
|
return result.rowcount if isinstance(result, CursorResult) else 0
|