peter
dc221338be
The E22 redactor is a safety net over addresses and credential params, and every one of these call sites was leaning on it to do the whole job. What it does not cover went in whole, and what it does cover came out as a constant. Not covered, so a real leak: - routes/search.py logged the raw search term at ERROR. The privacy policy shipped in 0.88.0 says a search string is kept only at Teljes/TRACE, with the user's own consent. The ERROR line now carries the shape of the failure (source, candidate and collected counts) and the term follows on a TRACE line, which nothing emits unless that consent is in force. - downloads/backfill.py logged the full source URL at INFO — a row of the user's download history. Now the asset id and the host. - youtube/client.py logged the upstream error body, which is prose that quotes the request back. Every failing YouTube call now goes through one _upstream_failure() that logs the API's `reason` code, keeps the body for the same consented TRACE line, and raises a YouTubeError carrying `reason` as a field — callers put that error in log lines of their own (routes/channels.py) and in a 422 answered to the user, so the body must not travel in it either. Covered, so merely useless: routes/auth.py named the account by an address that reached both sinks as `[email]`, and app/email.py printed recipient lists that read as `['[email]']`. Both now carry the uid (or a count) and a reason code, which is also the first half of E22.5 S4's failed-sign-in coverage. routes/channels.py branched on the text of that error message to recognise an already-existing subscription. It now branches on `exc.reason`, with the substring check kept beside it for a body that carried no reason at all. tests/test_log_pii.py pins the call sites and the family: a source-level contract test walks every log call in app/ and fails on an argument naming user-typed content unless the line says why it is safe (`# log-pii-ok: <why>`). Eight sites carry that marker today, each with its reason. Gate 8/8 green; all five cases mutation-checked.
403 lines
17 KiB
Python
403 lines
17 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)."""
|
|
|
|
import logging
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
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 logs, 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.feedquery import feed_columns, serialize
|
|
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
|
|
|
|
log = logging.getLogger("siftlode.search")
|
|
|
|
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.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 _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.
|
|
# 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:
|
|
detail_by_id = {}
|
|
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 _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")
|
|
|
|
# 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
|
|
# 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 burns a second quota-costed search. Upstream failure
|
|
# on a JSON endpoint = 422 + message; reserve 5xx for the raw byte-proxies. (B-core-1)
|
|
raise HTTPException(
|
|
status_code=422, detail=f"YouTube search failed: {exc}"
|
|
) from 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
|
|
]
|
|
# Ingest (DB writes + enrich) runs outside the page-fetch try above; guard it too so a
|
|
# write/shape failure degrades to a 422 with a message on this JSON endpoint instead of a
|
|
# bare 500 hidden behind the FE's generic "search failed" (same rationale as B-core-1).
|
|
try:
|
|
collected.extend(_ingest_candidates(db, yt, user, candidates))
|
|
except Exception as exc: # noqa: BLE001 — any ingest failure, not just YouTubeError
|
|
db.rollback()
|
|
# 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, where the
|
|
# consent that allows it is the only thing that emits the record at all.
|
|
log.exception(
|
|
"live search ingest failed",
|
|
extra=logs.event(
|
|
"search.live.ingest_failed",
|
|
source=source,
|
|
candidates=len(candidates),
|
|
collected=len(collected),
|
|
),
|
|
)
|
|
log.log(
|
|
logs.TRACE,
|
|
"the term the ingest failed on",
|
|
# log-pii-ok: TRACE is the consented level, and this is what it is for
|
|
extra=logs.event("search.live.term", term=term),
|
|
)
|
|
if collected:
|
|
break # keep what earlier pages already gathered
|
|
raise HTTPException(
|
|
status_code=422, detail=f"YouTube search failed: {exc}"
|
|
) from exc
|
|
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(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,
|
|
}
|
|
|
|
|
|
class ClearSearchIn(BaseModel):
|
|
video_ids: list[str] = []
|
|
|
|
|
|
@router.post("/clear")
|
|
def clear_search(
|
|
body: ClearSearchIn,
|
|
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 = body.video_ids
|
|
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}
|