94417ada72
BUG-1 (search.py, high): the scrape search source logged one VIDEOS_SEARCH quota event PER continuation page inside the paging loop, but actions_today counts events — so a single user search that pages N times consumed N against search_daily_limit_per_user (its docstring even warns it only works for once-per-action logging). Log exactly once per request after the loop instead; the API source already logs once via record_usage (it never auto-pages). BUG-2 (Feed.tsx, medium): the optimistic-override reset effect keyed on query.dataUpdatedAt also fired on fetchNextPage (infinite scroll bumps dataUpdatedAt while page 1 is unchanged), so a just-hidden card flashed back mid-scroll. Guard with a page-count ref: only reset on a real refetch, not an append. The two override-reset effects legitimately differ now (resolves the would-be C-F1 "identical effects" cleanup). tsc green, ruff clean, localdev boots healthy.
351 lines
14 KiB
Python
351 lines
14 KiB
Python
"""Live YouTube search.
|
|
|
|
Searches YouTube directly 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). The
|
|
source is admin-selectable (``search_source``): **scrape** (InnerTube, the website's own
|
|
endpoint, zero API quota — the default) or **api** (search.list, **100 quota units per page**).
|
|
Either way it's gated behind a per-user daily cap, the API source additionally behind a budget
|
|
pre-check, and only real users (never the shared demo account) may call it. The result ids are
|
|
enriched through the cheap shared videos.list path (1 unit / 50) regardless of source.
|
|
|
|
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_, delete, func, or_, select, update
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
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 (
|
|
LIVE_OR_UPCOMING,
|
|
BlockedChannel,
|
|
Channel,
|
|
Playlist,
|
|
PlaylistItem,
|
|
SearchFind,
|
|
Subscription,
|
|
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 import search_scrape
|
|
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
|
|
|
|
|
|
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]
|
|
|
|
|
|
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."""
|
|
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)
|
|
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)
|
|
|
|
keep = {v.id for v in videos if not v.is_short and v.live_status not in LIVE_OR_UPCOMING}
|
|
return [vid for vid in ids if vid in keep]
|
|
|
|
|
|
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
|
|
sub_ch = select(Subscription.channel_id).where(Subscription.user_id == user.id)
|
|
has_state = (
|
|
select(VideoState.id)
|
|
.where(VideoState.video_id == Video.id, VideoState.user_id == user.id)
|
|
.exists()
|
|
)
|
|
in_playlist = (
|
|
select(PlaylistItem.id)
|
|
.join(Playlist, Playlist.id == PlaylistItem.playlist_id)
|
|
.where(PlaylistItem.video_id == Video.id, Playlist.user_id == user.id)
|
|
.exists()
|
|
)
|
|
have = set(
|
|
db.scalars(
|
|
select(Video.id).where(
|
|
Video.id.in_(ids),
|
|
or_(Video.channel_id.in_(sub_ch), has_state, in_playlist),
|
|
)
|
|
).all()
|
|
)
|
|
return [i for i in ids if i not in have] + [i for i in ids if i in have]
|
|
|
|
|
|
@router.get("/youtube")
|
|
def search_youtube(
|
|
q: str | None = None,
|
|
cursor: str | None = None,
|
|
limit: int = 20,
|
|
user: User = Depends(require_human),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""Live YouTube search results materialised into the catalog. `limit` is how many results to
|
|
gather (the count selector); with the free scrape source we page through continuations until
|
|
that many are collected, so there's no manual "load more". The API source costs 100 units per
|
|
page, so it returns a single page (≤50) regardless. `cursor` lets the caller fetch further."""
|
|
term = (q or "").strip()
|
|
if not term:
|
|
raise HTTPException(status_code=400, detail="A search term is required.")
|
|
limit = max(1, min(limit, 100))
|
|
|
|
# Source: "scrape" (InnerTube, no API quota) or "api" (search.list, 100 units/page).
|
|
source = (sysconfig.get_str(db, "search_source") or "scrape").lower()
|
|
if source != "scrape":
|
|
source = "api"
|
|
|
|
# Per-user daily cap applies to both sources (anti-abuse); with scrape it's the only guard
|
|
# since the search itself spends no quota.
|
|
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.",
|
|
)
|
|
# The 100-unit budget pre-check only applies to the API source.
|
|
if source == "api" and 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.",
|
|
)
|
|
|
|
# Channels this user has blocked are dropped before ingestion — never shown, never stored.
|
|
blocked = set(
|
|
db.scalars(
|
|
select(BlockedChannel.channel_id).where(BlockedChannel.user_id == user.id)
|
|
).all()
|
|
)
|
|
|
|
collected: list[str] = []
|
|
next_cursor = cursor
|
|
MAX_PAGES = 12 # scrape safety bound
|
|
with YouTubeClient(db, user) as yt:
|
|
for _ in range(MAX_PAGES):
|
|
try:
|
|
if source == "api":
|
|
with quota.attribute(user.id, quota.QuotaAction.VIDEOS_SEARCH):
|
|
page = yt.search_videos(term, page_token=next_cursor)
|
|
else:
|
|
page = search_scrape.search(term, continuation=next_cursor)
|
|
except (YouTubeError, search_scrape.ScrapeError) as exc:
|
|
if collected:
|
|
break # keep what we already gathered
|
|
raise HTTPException(status_code=502, detail=f"YouTube search failed: {exc}")
|
|
|
|
# Drop currently-live/upcoming, channel-less, and blocked-channel results.
|
|
candidates = [
|
|
it
|
|
for it in page["items"]
|
|
if it["live_broadcast"] == "none"
|
|
and it["channel_id"]
|
|
and it["channel_id"] not in blocked
|
|
]
|
|
collected.extend(_ingest_candidates(db, yt, user, candidates))
|
|
next_cursor = page["next_page_token"]
|
|
# The API source pages cost 100 units each — never auto-page; one page per request.
|
|
if source == "api" or len(collected) >= limit or not next_cursor:
|
|
break
|
|
|
|
# Count this as ONE per-user search against the daily cap. The scrape source spends no
|
|
# quota, so it never logs an event on its own; the API source already logged exactly one
|
|
# (via record_usage — it never auto-pages). Logging once here, after a search that produced
|
|
# at least one page (a total failure raises 502 above and never reaches this), keeps the cap
|
|
# counting user searches instead of internal continuation pages.
|
|
if source != "api":
|
|
quota.log_action(db, user.id, quota.QuotaAction.VIDEOS_SEARCH)
|
|
|
|
ordered = collected[:limit]
|
|
|
|
if ordered:
|
|
# Record these as THIS user's search finds (Mine feed / Source filter), and fold the
|
|
# query into each result's search_terms so local full-text search inherits YouTube's
|
|
# relevance. Idempotent; the generated search_vector updates automatically.
|
|
db.execute(
|
|
pg_insert(SearchFind)
|
|
.values([{"user_id": user.id, "video_id": vid} for vid in ordered])
|
|
.on_conflict_do_nothing(index_elements=["user_id", "video_id"])
|
|
)
|
|
db.execute(
|
|
update(Video)
|
|
.where(Video.id.in_(ordered))
|
|
.where(func.coalesce(Video.search_terms, "").notilike(f"%{term}%"))
|
|
.values(
|
|
search_terms=func.trim(
|
|
func.coalesce(Video.search_terms, "").op("||")(" ").op("||")(term)
|
|
)
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
ordered = _new_first(db, user, ordered)
|
|
return {
|
|
"items": _serialize_ids(db, user, ordered),
|
|
"next_cursor": next_cursor,
|
|
"has_more": bool(next_cursor),
|
|
"source": source,
|
|
}
|
|
|
|
|
|
@router.post("/clear")
|
|
def clear_search(
|
|
payload: dict,
|
|
user: User = Depends(require_human),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""Discard the given live-search results "as if never added": drop this user's search-finds
|
|
for them, then hard-delete the now-orphaned ones — a via_search video nobody KEPT (no watch
|
|
state, not playlisted, channel not subscribed) and that no other user still has as a search
|
|
find. Kept videos and organically-owned ones are left untouched."""
|
|
ids = [str(v) for v in (payload.get("video_ids") or [])]
|
|
if not ids:
|
|
return {"removed": 0}
|
|
db.execute(
|
|
delete(SearchFind).where(
|
|
SearchFind.user_id == user.id, SearchFind.video_id.in_(ids)
|
|
)
|
|
)
|
|
db.commit()
|
|
has_state = select(VideoState.id).where(VideoState.video_id == Video.id).exists()
|
|
in_playlist = select(PlaylistItem.id).where(PlaylistItem.video_id == Video.id).exists()
|
|
subscribed = (
|
|
select(Subscription.id).where(Subscription.channel_id == Video.channel_id).exists()
|
|
)
|
|
other_find = select(SearchFind.id).where(SearchFind.video_id == Video.id).exists()
|
|
removed = db.execute(
|
|
delete(Video).where(
|
|
Video.id.in_(ids),
|
|
Video.via_search.is_(True),
|
|
~has_state,
|
|
~in_playlist,
|
|
~subscribed,
|
|
~other_find,
|
|
)
|
|
).rowcount
|
|
db.commit()
|
|
return {"removed": removed}
|