siftlode/backend/app/routes/search.py

205 lines
8.4 KiB
Python
Raw Normal View History

"""Live YouTube search.
Searches YouTube directly (search.list) and materialises the results into the shared catalog
so they render with the normal feed cards + in-app player and gain per-user state (watch /
save / resume). search.list costs **100 quota units per call**, so it's the most expensive
read by far: it's gated behind a per-user daily cap + a budget pre-check, and only real users
(never the shared demo account) may call it.
Ingested results are flagged (videos.via_search / channels.from_search) so the Library can hide
this search-discovered "noise" by default. Shorts and currently-live/upcoming videos are never
ingested this way (Shorts are confirmed by the youtube.com/shorts probe; live/upcoming are
dropped from the search response up front)."""
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import and_, select
from sqlalchemy.orm import Session, aliased
from app import quota, sysconfig
from app.auth import require_human
from app.db import get_db
from app.models import Channel, User, Video, VideoState
from app.routes.feed import _serialize, feed_columns
from app.sync.subscriptions import apply_channel_details
from app.sync.videos import _insert_stubs, apply_video_details, parse_dt
from app.youtube.client import YouTubeClient, YouTubeError
from app.youtube.shorts import make_client, probe_is_short
router = APIRouter(prefix="/api/search", tags=["search"])
# search.list costs this many quota units per page.
SEARCH_COST = 100
# live_status values we never surface from a live search (was_live VODs are real content).
_LIVE_HIDDEN = ("live", "upcoming")
def _classify_shorts(db: Session, videos: list[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.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 _serialize_ids(db: Session, user: User, ids: list[str]) -> list[dict]:
"""Serialize the given video ids with the shared feed projection (joined channel + this
user's watch state + saved), preserving the input (search relevance) order."""
if not ids:
return []
state = aliased(VideoState)
rows = db.execute(
select(*feed_columns(user, state))
.join(Channel, Channel.id == Video.channel_id)
.outerjoin(state, and_(state.video_id == Video.id, state.user_id == user.id))
.where(Video.id.in_(ids))
).all()
by_id = {r.id: _serialize(r) for r in rows}
return [by_id[i] for i in ids if i in by_id]
@router.get("/youtube")
def search_youtube(
q: str | None = None,
cursor: str | None = None,
user: User = Depends(require_human),
db: Session = Depends(get_db),
) -> dict:
"""One page of live YouTube search results, materialised into the catalog. `cursor` is the
YouTube nextPageToken (each page is another 100-unit search)."""
term = (q or "").strip()
if not term:
raise HTTPException(status_code=400, detail="A search term is required.")
# Guard the shared quota BEFORE spending: per-user daily cap, then today's budget.
daily_limit = sysconfig.get_int(db, "search_daily_limit_per_user")
used = quota.actions_today(db, user.id, quota.QuotaAction.VIDEOS_SEARCH)
if used >= daily_limit:
raise HTTPException(
status_code=429,
detail=f"You've reached today's YouTube search limit ({daily_limit}). Try again tomorrow.",
)
if not quota.can_spend(db, SEARCH_COST):
raise HTTPException(
status_code=429,
detail="The shared YouTube quota for today is used up. Try again tomorrow.",
)
with YouTubeClient(db, user) as yt:
try:
with quota.attribute(user.id, quota.QuotaAction.VIDEOS_SEARCH):
page = yt.search_videos(term, page_token=cursor)
except YouTubeError as exc:
raise HTTPException(status_code=502, detail=f"YouTube search failed: {exc}")
# Drop currently-live/upcoming results (search can't filter them and we never ingest
# them this way) and anything missing a channel id.
candidates = [
it
for it in page["items"]
if it["live_broadcast"] == "none" and it["channel_id"]
]
ids = [it["id"] for it in candidates]
if not ids:
return {
"items": [],
"next_cursor": page["next_page_token"],
"has_more": bool(page["next_page_token"]),
}
# 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)
db.commit()
except YouTubeError:
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.
_insert_stubs(
db,
[
{
"id": it["id"],
"channel_id": it["channel_id"],
"title": it["title"],
"published_at": parse_dt(it["published_at"]),
"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:
detail_by_id = {}
now = datetime.now(timezone.utc)
for v in videos:
item = detail_by_id.get(v.id)
if item is None:
continue
apply_video_details(v, item)
v.enriched_at = now
db.commit()
# 4) Confirm/deny Shorts (no quota) so none enter via search.
_classify_shorts(db, videos)
# 5) Exclude Shorts and still-live results; keep the search's relevance order.
keep = {
v.id for v in videos if not v.is_short and v.live_status not in _LIVE_HIDDEN
}
ordered = [vid for vid in ids if vid in keep]
return {
"items": _serialize_ids(db, user, ordered),
"next_cursor": page["next_page_token"],
"has_more": bool(page["next_page_token"]),
}