peter
2e1934689f
The slot was kept as soon as search.list was about to be called, so a search refused before it went out (no connection, no token) cost the user a slot for nothing. It is kept now only once the client recorded the call's units, or the request went out and its answer never came (YouTube may have counted it): YouTubeError says whether a transport failure was sent, and the client counts what it spent. The process lock around the budget check and the call is waited for at most ten seconds; past that the search is refused as busy (search_busy, a retry can help) and gets its slot back, instead of parking a request worker behind every slow call before it. The lock test now pins one hold around both steps, and the busy test fails on a wait without a bound.
411 lines
18 KiB
Python
411 lines
18 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. 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).
|
|
|
|
A video's related videos (`/related/{video_id}`, the list YouTube shows beside it) come in the same
|
|
way: read from the website (`related_scrape`), ingested as search finds are, under their own daily
|
|
cap — with what this user has already seen left out.
|
|
|
|
This is the HTTP side — the paging, the caps and the refusals; what a page of results leaves in the
|
|
catalog is `catalog/search`."""
|
|
|
|
import threading
|
|
from collections.abc import Callable
|
|
from typing import Literal
|
|
|
|
from fastapi import APIRouter, Depends, Path
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import logs, quota, sysconfig
|
|
from app.auth import current_user
|
|
from app.catalog import queries
|
|
from app.catalog import search as catalog_search
|
|
from app.db import get_db
|
|
from app.models import User
|
|
from app.ratelimit import RateLimiter
|
|
from app.reader import queries as reader_queries
|
|
from app.youtube import related_scrape, search_scrape
|
|
from app.youtube.innertube import ScrapeError
|
|
from app.youtube.client import YouTubeClient, YouTubeError
|
|
from app.refusal import refused
|
|
from app.routes._common import youtube_refusal
|
|
|
|
log = logs.get_logger(__name__)
|
|
router = APIRouter(prefix="/api/search", tags=["search"])
|
|
|
|
# search.list costs this many quota units per page.
|
|
SEARCH_COST = 100
|
|
# The scrape source reads at most this many pages for one search.
|
|
SEARCH_MAX_PAGES = 12
|
|
|
|
# A related list is filled to this many unseen videos, reading at most this many pages (about 20
|
|
# videos each, measured) — the ones the user has seen are left out, so a page alone can fall short.
|
|
RELATED_TARGET = 20
|
|
RELATED_MAX_PAGES = 4
|
|
# The locale YouTube fills a related list in: on the video's topic, or with what trends in Hungary.
|
|
RELATED_REGIONS = {"topical": ("en", "US"), "hu": ("hu", "HU")}
|
|
|
|
# A request that fails outright gives its slot back, but its shape is the user's (a cursor is a
|
|
# free string): one built to fail would read YouTube for nothing, without end. So only this many
|
|
# refunds an hour per user and list; past them a failure keeps its slot, as a success does.
|
|
REFUNDS_PER_HOUR = 5
|
|
_refunds = RateLimiter(max_events=REFUNDS_PER_HOUR, window_seconds=3600)
|
|
# The API source's budget check and the search.list call it pays for run one at a time: checked
|
|
# apart, searches sent together at the last 100 units all passed, and all spent. One process
|
|
# serves the API (a single uvicorn worker, as `ratelimit` says), so a process lock holds them all.
|
|
# A search waits this long for it, then is refused as busy: each waiter holds a request worker, and
|
|
# a slow YouTube (30 s a call at worst) would otherwise park them all behind one another.
|
|
_api_search = threading.Lock()
|
|
API_SEARCH_WAIT_SECONDS = 10.0
|
|
|
|
|
|
def _gather(
|
|
db: Session,
|
|
yt: YouTubeClient,
|
|
user: User,
|
|
*,
|
|
cursor: str | None,
|
|
max_pages: int,
|
|
fetch: Callable[[str | None], dict],
|
|
candidates_of: Callable[[list[dict], list[str]], list[dict]],
|
|
enough: Callable[[list[str]], bool],
|
|
scrape_failed: Callable[[Exception], None],
|
|
ingest_failed: Callable[[int, int], None],
|
|
) -> tuple[list[str], str | None]:
|
|
"""Page through a YouTube list into the catalog: fetch a page, keep its `candidates_of`, ingest
|
|
them, and go on from its continuation until `enough` or `max_pages`. Returns the kept ids and
|
|
the cursor to read further from. A page that fails after earlier ones gathered something ends
|
|
the list with what it has; a first page that fails raises the 422 refusal.
|
|
|
|
422, not 502: this is a JSON endpoint, and the FE treats 502/503/504 as a gateway dropout
|
|
(auto-retry + "connection lost" banner — lib/api/core RETRIABLE_GATEWAY), which hides the real
|
|
message and repeats the work. A Data API failure is logged where it happened; the scraper's is
|
|
not, and its text does not ride the answer — so `scrape_failed` says it. An ingest (DB writes +
|
|
enrich) fails the same way, guarded for any exception so a write or shape failure is a 422 with
|
|
a message too, not a bare 500; `ingest_failed` logs it inside the `except`, with the trace."""
|
|
collected: list[str] = []
|
|
next_cursor = cursor
|
|
for _ in range(max_pages):
|
|
try:
|
|
page = fetch(next_cursor)
|
|
except (YouTubeError, ScrapeError) as exc:
|
|
if collected:
|
|
break # keep what earlier pages already gathered
|
|
if isinstance(exc, ScrapeError):
|
|
scrape_failed(exc)
|
|
raise youtube_refusal(exc) from exc
|
|
candidates = candidates_of(page["items"], collected)
|
|
try:
|
|
collected.extend(catalog_search.ingest_candidates(db, yt, user, candidates))
|
|
except Exception as exc: # any ingest failure, not just YouTubeError
|
|
db.rollback()
|
|
ingest_failed(len(candidates), len(collected))
|
|
if collected:
|
|
break
|
|
raise youtube_refusal(exc) from exc
|
|
next_cursor = page["next_page_token"]
|
|
if enough(collected) or not next_cursor:
|
|
break
|
|
return collected, next_cursor
|
|
|
|
|
|
def _give_back(claimed: int, feature: str, user_id: int) -> None:
|
|
"""Refund the slot of a request that failed outright, while this user has refunds left."""
|
|
if _refunds.allow(f"{feature}:{user_id}"):
|
|
quota.refund_action(claimed)
|
|
return
|
|
log.info(
|
|
"a failed request kept its slot: this hour's refunds are spent",
|
|
extra=logs.event(f"{feature}.refund_refused", per_hour=REFUNDS_PER_HOUR),
|
|
)
|
|
|
|
|
|
def _budget_spent() -> None:
|
|
"""Say that a live search was refused because the shared YouTube budget cannot pay for it."""
|
|
log.info(
|
|
"a live search was refused: the shared YouTube budget is spent",
|
|
extra=logs.event("search.refused", reason="shared_quota", cost=SEARCH_COST),
|
|
)
|
|
|
|
|
|
def _busy() -> None:
|
|
"""Say that a live search was refused because other API searches held the line too long."""
|
|
log.info(
|
|
"a live search was refused: other searches held the line",
|
|
extra=logs.event("search.refused", reason="busy", waited=API_SEARCH_WAIT_SECONDS),
|
|
)
|
|
|
|
|
|
def _cap_spent(feature: str, limit: int) -> None:
|
|
"""Say that a request was refused for the per-user daily cap. The 429 is in the request line
|
|
either way; which of the caps produced it is not, and they are answered by different things —
|
|
this one by tomorrow, the shared budget by the admin."""
|
|
log.info(
|
|
"a request was refused: the per-user daily cap is spent",
|
|
extra=logs.event(f"{feature}.refused", reason="daily_cap", limit=limit),
|
|
)
|
|
|
|
|
|
@router.get("/youtube")
|
|
def search_youtube(
|
|
q: str | None = None,
|
|
cursor: str | None = None,
|
|
limit: int = 20,
|
|
user: User = Depends(current_user),
|
|
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 refused(400, "search_term_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")
|
|
|
|
# Every request takes a slot of the per-user daily cap, a cursor's too (anti-abuse); with
|
|
# scrape it's the only guard since the search itself spends no quota. A limit of 0 turns live
|
|
# search off on this instance — a refusal tomorrow will not change.
|
|
daily_limit = sysconfig.get_int(db, "search_daily_limit_per_user")
|
|
if daily_limit <= 0:
|
|
raise refused(409, "search_disabled")
|
|
# The 100-unit budget pre-check only applies to the API source. It is asked again under the
|
|
# lock, right before the call (`fetch`); this one refuses before a slot is taken.
|
|
if source == "api" and not quota.can_spend(db, SEARCH_COST):
|
|
_budget_spent()
|
|
raise refused(429, "quota_spent")
|
|
# The check and the take are one step (`quota.claim_action`), whichever the source: checked
|
|
# first and logged after, requests sent together at the last free slot all passed — and with
|
|
# the API source each of them spent 100 units. A search that fails outright gives it back below.
|
|
claimed = quota.claim_action(user.id, quota.QuotaAction.VIDEOS_SEARCH, daily_limit)
|
|
if claimed is None:
|
|
_cap_spent("search", daily_limit)
|
|
raise refused(429, "search_daily_limit", limit=daily_limit)
|
|
|
|
# Channels this user has blocked are dropped before ingestion — never shown, never stored.
|
|
blocked = queries.blocked_ids(db, user)
|
|
|
|
# Whether search.list was paid for: YouTube charges a call it refuses too, so once the client
|
|
# recorded its units — or the request went out and no answer came back — the search has spent
|
|
# the shared budget. One that never reached YouTube (no connection, no token) has not.
|
|
charged = False
|
|
|
|
def fetch(token: str | None) -> dict:
|
|
if source != "api":
|
|
return search_scrape.search(term, continuation=token)
|
|
if not _api_search.acquire(timeout=API_SEARCH_WAIT_SECONDS):
|
|
_busy()
|
|
raise refused(429, "search_busy")
|
|
try:
|
|
return paid_search(token)
|
|
finally:
|
|
_api_search.release()
|
|
|
|
def paid_search(token: str | None) -> dict:
|
|
nonlocal charged
|
|
with quota.attribute(user.id, quota.QuotaAction.VIDEOS_SEARCH):
|
|
if not quota.can_spend(db, SEARCH_COST):
|
|
_budget_spent()
|
|
raise refused(429, "quota_spent")
|
|
spent = yt.spent
|
|
try:
|
|
page = yt.search_videos(term, page_token=token)
|
|
except YouTubeError as exc:
|
|
charged = yt.spent > spent or exc.sent
|
|
raise
|
|
charged = True
|
|
return page
|
|
|
|
def candidates_of(items: list[dict], _collected: list[str]) -> list[dict]:
|
|
# Drop currently-live/upcoming, channel-less, and blocked-channel results.
|
|
return [
|
|
it
|
|
for it in items
|
|
if it["live_broadcast"] == "none"
|
|
and it["channel_id"]
|
|
and it["channel_id"] not in blocked
|
|
]
|
|
|
|
def scrape_failed(exc: Exception) -> None:
|
|
log.warning(
|
|
"the YouTube search scrape failed",
|
|
extra=logs.event("search.scrape_failed", error=str(exc)),
|
|
)
|
|
|
|
def ingest_failed(candidates: int, collected: int) -> None:
|
|
# The term is NOT in this line. A search string is exactly what the privacy policy keeps
|
|
# out of the log unless the user turned Teljes/TRACE on for themselves, and `redact_text`
|
|
# does not cover it — it scrubs addresses and credential params, not what someone typed
|
|
# into the search box. What the ERROR line carries instead is the shape of the failure,
|
|
# which is what tells an ingest bug from an upstream one; the term follows at TRACE, and
|
|
# only under this user's own consent — not whenever the process runs at TRACE.
|
|
log.exception(
|
|
"live search ingest failed",
|
|
extra=logs.event(
|
|
"search.live.ingest_failed",
|
|
source=source,
|
|
candidates=candidates,
|
|
collected=collected,
|
|
),
|
|
)
|
|
if logs.consented_trace():
|
|
log.log(
|
|
logs.TRACE,
|
|
"the term the ingest failed on",
|
|
# log-pii-ok: only under this user's own Teljes consent
|
|
extra=logs.event("search.live.term", term=term),
|
|
)
|
|
|
|
try:
|
|
with YouTubeClient(db, user) as yt:
|
|
collected, next_cursor = _gather(
|
|
db,
|
|
yt,
|
|
user,
|
|
cursor=cursor,
|
|
# The API source pages cost 100 units each — never auto-page; one page per request.
|
|
max_pages=1 if source == "api" else SEARCH_MAX_PAGES,
|
|
fetch=fetch,
|
|
candidates_of=candidates_of,
|
|
enough=lambda got: len(got) >= limit,
|
|
scrape_failed=scrape_failed,
|
|
ingest_failed=ingest_failed,
|
|
)
|
|
except Exception:
|
|
# A search that showed the user nothing does not count against them, unless search.list
|
|
# was already paid for: then the slot is what stops a failing search (a bogus cursor gets
|
|
# a 400 every time) from spending the shared budget without end. A busy refusal paid
|
|
# nothing and gives it back.
|
|
if not charged:
|
|
_give_back(claimed, "search", user.id)
|
|
raise
|
|
|
|
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.
|
|
catalog_search.record_finds(db, user, ordered, term)
|
|
db.commit()
|
|
|
|
ordered = catalog_search.new_first(db, user, ordered)
|
|
return {
|
|
"items": reader_queries.feed_rows_by_ids(db, user, ordered),
|
|
"next_cursor": next_cursor,
|
|
"has_more": bool(next_cursor),
|
|
"source": source,
|
|
}
|
|
|
|
|
|
class ClearSearchIn(BaseModel):
|
|
video_ids: list[str] = []
|
|
|
|
|
|
@router.post("/clear")
|
|
def clear_search(
|
|
body: ClearSearchIn,
|
|
user: User = Depends(current_user),
|
|
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 = body.video_ids
|
|
if not ids:
|
|
return {"ok": True, "removed": 0}
|
|
removed = catalog_search.clear(db, user, ids)
|
|
db.commit()
|
|
return {"ok": True, "removed": removed}
|
|
|
|
|
|
@router.get("/related/{video_id}")
|
|
def related_videos(
|
|
video_id: str = Path(max_length=16),
|
|
cursor: str | None = None,
|
|
region: Literal["topical", "hu"] = "topical",
|
|
user: User = Depends(current_user),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""The videos YouTube shows beside this one, materialised into the catalog like search results,
|
|
without the ones this user has watched or hidden here or in their Takeout history, and without
|
|
blocked channels. Pages on (`cursor`) until `RELATED_TARGET` are gathered; `next_cursor` reads
|
|
further. `region` picks the locale YouTube fills the list in (`RELATED_REGIONS`)."""
|
|
# Every request takes a slot, a cursor's too: each one scrapes pages and enriches what they
|
|
# hold, and a cursor is a free string — counting only first pages let a replayed cursor read
|
|
# without end. The check and the take are one step (`quota.claim_action`); a request that fails
|
|
# outright gives its slot back, a few an hour (`_give_back`). A limit of 0 turns related videos
|
|
# off on this instance.
|
|
daily_limit = sysconfig.get_int(db, "related_daily_limit_per_user")
|
|
if daily_limit <= 0:
|
|
raise refused(409, "related_disabled")
|
|
claimed = quota.claim_action(user.id, quota.QuotaAction.VIDEOS_RELATED, daily_limit)
|
|
if claimed is None:
|
|
_cap_spent("related", daily_limit)
|
|
raise refused(429, "related_daily_limit", limit=daily_limit)
|
|
|
|
hl, gl = RELATED_REGIONS[region]
|
|
blocked = queries.blocked_ids(db, user)
|
|
|
|
def candidates_of(items: list[dict], collected: list[str]) -> list[dict]:
|
|
ids = [it["id"] for it in items]
|
|
drop = reader_queries.seen_by(db, user, ids) | set(collected) | {video_id}
|
|
return [it for it in items if it["id"] not in drop and it["channel_id"] not in blocked]
|
|
|
|
def scrape_failed(exc: Exception) -> None:
|
|
log.warning(
|
|
"the related-videos scrape failed",
|
|
extra=logs.event("related.scrape_failed", error=str(exc)),
|
|
)
|
|
|
|
def ingest_failed(candidates: int, collected: int) -> None:
|
|
log.exception(
|
|
"a related list's ingest failed",
|
|
extra=logs.event("related.ingest_failed", candidates=candidates, collected=collected),
|
|
)
|
|
|
|
try:
|
|
with YouTubeClient(db, user) as yt:
|
|
collected, next_cursor = _gather(
|
|
db,
|
|
yt,
|
|
user,
|
|
cursor=cursor,
|
|
max_pages=RELATED_MAX_PAGES,
|
|
fetch=lambda token: related_scrape.related(
|
|
video_id, continuation=token, hl=hl, gl=gl
|
|
),
|
|
candidates_of=candidates_of,
|
|
enough=lambda got: len(got) >= RELATED_TARGET,
|
|
scrape_failed=scrape_failed,
|
|
ingest_failed=ingest_failed,
|
|
)
|
|
except Exception:
|
|
_give_back(claimed, "related", user.id)
|
|
raise
|
|
|
|
rows = reader_queries.feed_rows_by_ids(db, user, collected)
|
|
# A related list mixes channels the user follows with ones they never met: each row says which.
|
|
followed = queries.followed_among(db, user, {r["channel_id"] for r in rows})
|
|
return {
|
|
"items": [{**r, "followed": r["channel_id"] in followed} for r in rows],
|
|
"next_cursor": next_cursor,
|
|
"has_more": bool(next_cursor),
|
|
}
|