feat(search): ephemeral results, count selector, blocklist, new-first ordering (backend)
- Live-search results stay ephemeral: the discovery-cleanup job now also reclaims un-kept via_search videos (no watch state / not playlisted / channel not subscribed) after a grace period (search_grace_days), and POST /api/search/clear discards a given result set 'as if never added' (drops the user's search-finds + deletes the now-orphaned, un-kept videos). Admin POST /api/admin/purge-discovery runs it on demand (grace 0). - Count selector: GET /api/search/youtube gains a 'limit' — the free scrape source pages through continuations until that many results are gathered (no manual load-more); the API source stays one ≤50 page (cost). - Per-user channel blocklist (migration 0034 blocked_channels): block/unblock/list endpoints; blocked channels' videos are dropped from live search before ingest and hidden from the feed / Library / explore / channel page; explore refuses a blocked channel. - New-first ordering: results you already have (subscribed channel, watched/in-progress/saved/ playlisted) sink below genuinely-new discoveries, preserving relevance within each group.
This commit is contained in:
parent
77707a3cbe
commit
f27f31b6db
10 changed files with 396 additions and 102 deletions
46
backend/alembic/versions/0034_blocked_channels.py
Normal file
46
backend/alembic/versions/0034_blocked_channels.py
Normal file
|
|
@ -0,0 +1,46 @@
|
||||||
|
"""per-user blocked channels (YouTube search / feed / explore)
|
||||||
|
|
||||||
|
Revision ID: 0034_blocked_channels
|
||||||
|
Revises: 0033_channel_explore
|
||||||
|
Create Date: 2026-07-01
|
||||||
|
|
||||||
|
Adds the `blocked_channels` table: a per-user blocklist. Videos from a blocked channel are
|
||||||
|
never shown in that user's live YouTube search (and not ingested), and are hidden from their
|
||||||
|
feed / explore views. Per-user, so one user's block doesn't affect anyone else.
|
||||||
|
"""
|
||||||
|
from typing import Sequence, Union
|
||||||
|
|
||||||
|
import sqlalchemy as sa
|
||||||
|
from alembic import op
|
||||||
|
|
||||||
|
revision: str = "0034_blocked_channels"
|
||||||
|
down_revision: Union[str, None] = "0033_channel_explore"
|
||||||
|
branch_labels: Union[str, Sequence[str], None] = None
|
||||||
|
depends_on: Union[str, Sequence[str], None] = None
|
||||||
|
|
||||||
|
|
||||||
|
def upgrade() -> None:
|
||||||
|
op.create_table(
|
||||||
|
"blocked_channels",
|
||||||
|
sa.Column("id", sa.Integer(), nullable=False),
|
||||||
|
sa.Column("user_id", sa.Integer(), nullable=False),
|
||||||
|
sa.Column("channel_id", sa.String(length=32), nullable=False),
|
||||||
|
sa.Column(
|
||||||
|
"created_at",
|
||||||
|
sa.DateTime(timezone=True),
|
||||||
|
server_default=sa.text("now()"),
|
||||||
|
nullable=False,
|
||||||
|
),
|
||||||
|
sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"),
|
||||||
|
sa.ForeignKeyConstraint(["channel_id"], ["channels.id"], ondelete="CASCADE"),
|
||||||
|
sa.PrimaryKeyConstraint("id"),
|
||||||
|
sa.UniqueConstraint("user_id", "channel_id", name="uq_user_blocked_channel"),
|
||||||
|
)
|
||||||
|
op.create_index("ix_blocked_channels_user_id", "blocked_channels", ["user_id"])
|
||||||
|
op.create_index("ix_blocked_channels_channel_id", "blocked_channels", ["channel_id"])
|
||||||
|
|
||||||
|
|
||||||
|
def downgrade() -> None:
|
||||||
|
op.drop_index("ix_blocked_channels_channel_id", table_name="blocked_channels")
|
||||||
|
op.drop_index("ix_blocked_channels_user_id", table_name="blocked_channels")
|
||||||
|
op.drop_table("blocked_channels")
|
||||||
|
|
@ -124,6 +124,9 @@ class Settings(BaseSettings):
|
||||||
# days after it was last explored — the grace clock is ExploredChannel.created_at.
|
# days after it was last explored — the grace clock is ExploredChannel.created_at.
|
||||||
explore_cleanup_minutes: int = 1440
|
explore_cleanup_minutes: int = 1440
|
||||||
explore_grace_days: int = 7
|
explore_grace_days: int = 7
|
||||||
|
# How long an un-kept live-search result (a via_search video nobody watched / saved /
|
||||||
|
# subscribed to) is kept before the same cleanup job removes it — "as if never added".
|
||||||
|
search_grace_days: int = 7
|
||||||
# live_status values hidden from the feed by default. Completed-stream VODs
|
# live_status values hidden from the feed by default. Completed-stream VODs
|
||||||
# ("was_live") are real watchable content and stay visible.
|
# ("was_live") are real watchable content and stay visible.
|
||||||
feed_default_hidden_live: str = "live,upcoming"
|
feed_default_hidden_live: str = "live,upcoming"
|
||||||
|
|
|
||||||
|
|
@ -435,6 +435,25 @@ class ExploredChannel(Base, TimestampMixin):
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class BlockedChannel(Base, TimestampMixin):
|
||||||
|
"""Per-user channel blocklist. Videos from a blocked channel are excluded from this user's
|
||||||
|
live YouTube search (and not ingested on their behalf) and hidden from their feed / explore
|
||||||
|
views — a personal "never show me this channel again" that doesn't affect other users."""
|
||||||
|
|
||||||
|
__tablename__ = "blocked_channels"
|
||||||
|
__table_args__ = (
|
||||||
|
UniqueConstraint("user_id", "channel_id", name="uq_user_blocked_channel"),
|
||||||
|
)
|
||||||
|
|
||||||
|
id: Mapped[int] = mapped_column(primary_key=True)
|
||||||
|
user_id: Mapped[int] = mapped_column(
|
||||||
|
ForeignKey("users.id", ondelete="CASCADE"), index=True
|
||||||
|
)
|
||||||
|
channel_id: Mapped[str] = mapped_column(
|
||||||
|
ForeignKey("channels.id", ondelete="CASCADE"), index=True
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class ApiQuotaUsage(Base):
|
class ApiQuotaUsage(Base):
|
||||||
"""Tracks YouTube Data API units spent per Pacific-time day (the quota resets at
|
"""Tracks YouTube Data API units spent per Pacific-time day (the quota resets at
|
||||||
midnight Pacific). The whole app shares one daily budget."""
|
midnight Pacific). The whole app shares one daily budget."""
|
||||||
|
|
|
||||||
|
|
@ -376,3 +376,15 @@ def reset_demo(
|
||||||
same automatically (admin-tunable interval); this is the admin's on-demand button."""
|
same automatically (admin-tunable interval); this is the admin's on-demand button."""
|
||||||
demo = get_or_create_demo_user(db)
|
demo = get_or_create_demo_user(db)
|
||||||
return {"reset": True, "playlists_seeded": reset_demo_state(db, demo)}
|
return {"reset": True, "playlists_seeded": reset_demo_state(db, demo)}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/purge-discovery")
|
||||||
|
def purge_discovery(
|
||||||
|
_: User = Depends(admin_user), db: Session = Depends(get_db)
|
||||||
|
) -> dict:
|
||||||
|
"""Reclaim all un-kept discovery content NOW (ignoring the grace period): live-search result
|
||||||
|
videos nobody watched/saved/subscribed to, plus explored-but-unsubscribed channels. The
|
||||||
|
scheduled discovery-cleanup job does the same on its interval; this is the on-demand button."""
|
||||||
|
from app.sync.explore import purge_unkept_explored, purge_unkept_search
|
||||||
|
|
||||||
|
return {**purge_unkept_search(db, grace_days=0), **purge_unkept_explored(db)}
|
||||||
|
|
|
||||||
|
|
@ -12,6 +12,7 @@ from app import quota, sysconfig
|
||||||
from app.auth import current_user, has_write_scope, require_human
|
from app.auth import current_user, has_write_scope, require_human
|
||||||
from app.db import get_db
|
from app.db import get_db
|
||||||
from app.models import (
|
from app.models import (
|
||||||
|
BlockedChannel,
|
||||||
Channel,
|
Channel,
|
||||||
ChannelTag,
|
ChannelTag,
|
||||||
ExploredChannel,
|
ExploredChannel,
|
||||||
|
|
@ -287,6 +288,12 @@ def explore_channel(
|
||||||
channel = db.get(Channel, channel_id)
|
channel = db.get(Channel, channel_id)
|
||||||
if channel is None:
|
if channel is None:
|
||||||
raise HTTPException(status_code=404, detail="Unknown channel")
|
raise HTTPException(status_code=404, detail="Unknown channel")
|
||||||
|
if db.execute(
|
||||||
|
select(BlockedChannel.id).where(
|
||||||
|
BlockedChannel.user_id == user.id, BlockedChannel.channel_id == channel_id
|
||||||
|
)
|
||||||
|
).first() is not None:
|
||||||
|
raise HTTPException(status_code=403, detail="You've blocked this channel.")
|
||||||
if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
|
if quota.remaining_today(db) <= sysconfig.get_int(db, "backfill_quota_reserve"):
|
||||||
raise HTTPException(
|
raise HTTPException(
|
||||||
status_code=429,
|
status_code=429,
|
||||||
|
|
@ -504,6 +511,68 @@ def unsubscribe(
|
||||||
return {"unsubscribed": channel_id}
|
return {"unsubscribed": channel_id}
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/blocked/list")
|
||||||
|
def list_blocked(
|
||||||
|
user: User = Depends(current_user), db: Session = Depends(get_db)
|
||||||
|
) -> list[dict]:
|
||||||
|
"""The user's blocked channels (videos from these are excluded from their search/feed/explore)."""
|
||||||
|
rows = db.execute(
|
||||||
|
select(Channel)
|
||||||
|
.join(BlockedChannel, BlockedChannel.channel_id == Channel.id)
|
||||||
|
.where(BlockedChannel.user_id == user.id)
|
||||||
|
.order_by(func.lower(Channel.title))
|
||||||
|
).scalars().all()
|
||||||
|
return [
|
||||||
|
{"id": c.id, "title": c.title, "handle": c.handle, "thumbnail_url": c.thumbnail_url}
|
||||||
|
for c in rows
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/{channel_id}/block")
|
||||||
|
def block_channel(
|
||||||
|
channel_id: str,
|
||||||
|
user: User = Depends(current_user),
|
||||||
|
db: Session = Depends(get_db),
|
||||||
|
) -> dict:
|
||||||
|
"""Block a channel for this user: its videos won't appear in their live search (and won't be
|
||||||
|
ingested for them) and are hidden from their feed / explore. Idempotent. The now-hidden
|
||||||
|
un-kept search/explore videos are reclaimed by the discovery-cleanup job in due course."""
|
||||||
|
if db.get(Channel, channel_id) is None:
|
||||||
|
raise HTTPException(status_code=404, detail="Unknown channel")
|
||||||
|
exists = db.execute(
|
||||||
|
select(BlockedChannel.id).where(
|
||||||
|
BlockedChannel.user_id == user.id, BlockedChannel.channel_id == channel_id
|
||||||
|
)
|
||||||
|
).first()
|
||||||
|
if exists is None:
|
||||||
|
db.add(BlockedChannel(user_id=user.id, channel_id=channel_id))
|
||||||
|
# Stop any active exploration of it so it can be cleaned up.
|
||||||
|
db.execute(
|
||||||
|
ExploredChannel.__table__.delete().where(
|
||||||
|
(ExploredChannel.user_id == user.id)
|
||||||
|
& (ExploredChannel.channel_id == channel_id)
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db.commit()
|
||||||
|
return {"blocked": channel_id}
|
||||||
|
|
||||||
|
|
||||||
|
@router.delete("/{channel_id}/block")
|
||||||
|
def unblock_channel(
|
||||||
|
channel_id: str,
|
||||||
|
user: User = Depends(current_user),
|
||||||
|
db: Session = Depends(get_db),
|
||||||
|
) -> dict:
|
||||||
|
db.execute(
|
||||||
|
BlockedChannel.__table__.delete().where(
|
||||||
|
(BlockedChannel.user_id == user.id)
|
||||||
|
& (BlockedChannel.channel_id == channel_id)
|
||||||
|
)
|
||||||
|
)
|
||||||
|
db.commit()
|
||||||
|
return {"unblocked": channel_id}
|
||||||
|
|
||||||
|
|
||||||
@router.post("/{channel_id}/tags")
|
@router.post("/{channel_id}/tags")
|
||||||
def attach_tag(
|
def attach_tag(
|
||||||
channel_id: str,
|
channel_id: str,
|
||||||
|
|
|
||||||
|
|
@ -14,6 +14,7 @@ from app import quota
|
||||||
from app.auth import current_user
|
from app.auth import current_user
|
||||||
from app.db import get_db
|
from app.db import get_db
|
||||||
from app.models import (
|
from app.models import (
|
||||||
|
BlockedChannel,
|
||||||
Channel,
|
Channel,
|
||||||
ChannelTag,
|
ChannelTag,
|
||||||
ExploredChannel,
|
ExploredChannel,
|
||||||
|
|
@ -198,6 +199,14 @@ def _filtered_query(
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Per-user channel blocklist: never show this user videos from a channel they blocked
|
||||||
|
# (applies to feed, Library, explore, and the channel page alike).
|
||||||
|
query = query.where(
|
||||||
|
Video.channel_id.notin_(
|
||||||
|
select(BlockedChannel.channel_id).where(BlockedChannel.user_id == user.id)
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
# Source filter. In the shared Library ("all" scope) it uses the GLOBAL via_search flag
|
# Source filter. In the shared Library ("all" scope) it uses the GLOBAL via_search flag
|
||||||
# (search-discovered by anyone). In "my" scope it's the user's OWN provenance:
|
# (search-discovered by anyone). In "my" scope it's the user's OWN provenance:
|
||||||
# "organic" = your subscriptions, "search" = videos you found via your own search,
|
# "organic" = your subscriptions, "search" = videos you found via your own search,
|
||||||
|
|
|
||||||
|
|
@ -16,14 +16,24 @@ from concurrent.futures import ThreadPoolExecutor
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
from fastapi import APIRouter, Depends, HTTPException
|
from fastapi import APIRouter, Depends, HTTPException
|
||||||
from sqlalchemy import and_, func, select, update
|
from sqlalchemy import and_, delete, func, or_, select, update
|
||||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||||
from sqlalchemy.orm import Session, aliased
|
from sqlalchemy.orm import Session, aliased
|
||||||
|
|
||||||
from app import quota, sysconfig
|
from app import quota, sysconfig
|
||||||
from app.auth import require_human
|
from app.auth import require_human
|
||||||
from app.db import get_db
|
from app.db import get_db
|
||||||
from app.models import Channel, SearchFind, User, Video, VideoState
|
from app.models import (
|
||||||
|
BlockedChannel,
|
||||||
|
Channel,
|
||||||
|
Playlist,
|
||||||
|
PlaylistItem,
|
||||||
|
SearchFind,
|
||||||
|
Subscription,
|
||||||
|
User,
|
||||||
|
Video,
|
||||||
|
VideoState,
|
||||||
|
)
|
||||||
from app.routes.feed import _serialize, feed_columns
|
from app.routes.feed import _serialize, feed_columns
|
||||||
from app.sync.subscriptions import apply_channel_details
|
from app.sync.subscriptions import apply_channel_details
|
||||||
from app.sync.videos import _insert_stubs, apply_video_details, parse_dt
|
from app.sync.videos import _insert_stubs, apply_video_details, parse_dt
|
||||||
|
|
@ -91,18 +101,119 @@ def _serialize_ids(db: Session, user: User, ids: list[str]) -> list[dict]:
|
||||||
return [by_id[i] for i in ids if i in by_id]
|
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_HIDDEN}
|
||||||
|
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")
|
@router.get("/youtube")
|
||||||
def search_youtube(
|
def search_youtube(
|
||||||
q: str | None = None,
|
q: str | None = None,
|
||||||
cursor: str | None = None,
|
cursor: str | None = None,
|
||||||
|
limit: int = 20,
|
||||||
user: User = Depends(require_human),
|
user: User = Depends(require_human),
|
||||||
db: Session = Depends(get_db),
|
db: Session = Depends(get_db),
|
||||||
) -> dict:
|
) -> dict:
|
||||||
"""One page of live YouTube search results, materialised into the catalog. `cursor` is the
|
"""Live YouTube search results materialised into the catalog. `limit` is how many results to
|
||||||
YouTube nextPageToken (each page is another 100-unit search)."""
|
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()
|
term = (q or "").strip()
|
||||||
if not term:
|
if not term:
|
||||||
raise HTTPException(status_code=400, detail="A search term is required.")
|
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: "scrape" (InnerTube, no API quota) or "api" (search.list, 100 units/page).
|
||||||
source = (sysconfig.get_str(db, "search_source") or "scrape").lower()
|
source = (sysconfig.get_str(db, "search_source") or "scrape").lower()
|
||||||
|
|
@ -125,108 +236,56 @@ def search_youtube(
|
||||||
detail="The shared YouTube quota for today is used up. Try again tomorrow.",
|
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:
|
with YouTubeClient(db, user) as yt:
|
||||||
try:
|
for _ in range(MAX_PAGES):
|
||||||
if source == "api":
|
|
||||||
with quota.attribute(user.id, quota.QuotaAction.VIDEOS_SEARCH):
|
|
||||||
page = yt.search_videos(term, page_token=cursor)
|
|
||||||
else:
|
|
||||||
page = search_scrape.search(term, continuation=cursor)
|
|
||||||
# Zero-cost, but logged so the per-user daily cap above keeps counting it.
|
|
||||||
quota.log_action(db, user.id, quota.QuotaAction.VIDEOS_SEARCH)
|
|
||||||
except (YouTubeError, search_scrape.ScrapeError) 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"]),
|
|
||||||
"source": source,
|
|
||||||
}
|
|
||||||
|
|
||||||
# 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:
|
try:
|
||||||
with quota.attribute(user.id, quota.QuotaAction.CHANNELS_DISCOVER):
|
if source == "api":
|
||||||
details = yt.get_channels(new_channel_ids)
|
with quota.attribute(user.id, quota.QuotaAction.VIDEOS_SEARCH):
|
||||||
apply_channel_details(db, details)
|
page = yt.search_videos(term, page_token=next_cursor)
|
||||||
db.commit()
|
else:
|
||||||
except YouTubeError:
|
page = search_scrape.search(term, continuation=next_cursor)
|
||||||
db.rollback() # details can be backfilled later; the stubs remain
|
# Zero-cost, but logged so the per-user daily cap above keeps counting it.
|
||||||
|
quota.log_action(db, user.id, quota.QuotaAction.VIDEOS_SEARCH)
|
||||||
|
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}")
|
||||||
|
|
||||||
# 2) Video stubs, flagged via_search. Existing rows keep their flags (ON CONFLICT DO
|
# Drop currently-live/upcoming, channel-less, and blocked-channel results.
|
||||||
# NOTHING), so an organically-synced video stays organic.
|
candidates = [
|
||||||
_insert_stubs(
|
it
|
||||||
db,
|
for it in page["items"]
|
||||||
[
|
if it["live_broadcast"] == "none"
|
||||||
{
|
and it["channel_id"]
|
||||||
"id": it["id"],
|
and it["channel_id"] not in blocked
|
||||||
"channel_id": it["channel_id"],
|
]
|
||||||
"title": it["title"],
|
collected.extend(_ingest_candidates(db, yt, user, candidates))
|
||||||
"published_at": parse_dt(it["published_at"]),
|
next_cursor = page["next_page_token"]
|
||||||
"thumbnail_url": it["thumbnail_url"],
|
# The API source pages cost 100 units each — never auto-page; one page per request.
|
||||||
"via_search": True,
|
if source == "api" or len(collected) >= limit or not next_cursor:
|
||||||
}
|
break
|
||||||
for it in candidates
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
# 3) Enrich (videos.list, 1 unit/50) so cards have duration/stats and we can judge live.
|
ordered = collected[:limit]
|
||||||
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]
|
|
||||||
|
|
||||||
# 6) Record these as THIS user's search finds so they surface in their own Mine feed
|
|
||||||
# (and the Mine Source filter), independent of the global via_search flag. Idempotent.
|
|
||||||
if ordered:
|
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(
|
db.execute(
|
||||||
pg_insert(SearchFind)
|
pg_insert(SearchFind)
|
||||||
.values([{"user_id": user.id, "video_id": vid} for vid in ordered])
|
.values([{"user_id": user.id, "video_id": vid} for vid in ordered])
|
||||||
.on_conflict_do_nothing(index_elements=["user_id", "video_id"])
|
.on_conflict_do_nothing(index_elements=["user_id", "video_id"])
|
||||||
)
|
)
|
||||||
# 7) Fold the query into each result's search_terms (shared across users) so local
|
|
||||||
# full-text search inherits YouTube's relevance — a video YT returned for this query
|
|
||||||
# becomes findable by it even when its title doesn't contain the words. Skip rows
|
|
||||||
# that already include the term; the generated search_vector updates automatically.
|
|
||||||
db.execute(
|
db.execute(
|
||||||
update(Video)
|
update(Video)
|
||||||
.where(Video.id.in_(ordered))
|
.where(Video.id.in_(ordered))
|
||||||
|
|
@ -239,9 +298,49 @@ def search_youtube(
|
||||||
)
|
)
|
||||||
db.commit()
|
db.commit()
|
||||||
|
|
||||||
|
ordered = _new_first(db, user, ordered)
|
||||||
return {
|
return {
|
||||||
"items": _serialize_ids(db, user, ordered),
|
"items": _serialize_ids(db, user, ordered),
|
||||||
"next_cursor": page["next_page_token"],
|
"next_cursor": next_cursor,
|
||||||
"has_more": bool(page["next_page_token"]),
|
"has_more": bool(next_cursor),
|
||||||
"source": source,
|
"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}
|
||||||
|
|
|
||||||
|
|
@ -16,7 +16,7 @@ from app.models import SchedulerSetting
|
||||||
from app.notifications import create_notification
|
from app.notifications import create_notification
|
||||||
from app.state import is_sync_paused
|
from app.state import is_sync_paused
|
||||||
from app.sync.autotag import run_autotag_all
|
from app.sync.autotag import run_autotag_all
|
||||||
from app.sync.explore import purge_unkept_explored
|
from app.sync.explore import purge_ephemeral
|
||||||
from app.sync.maintenance import run_maintenance
|
from app.sync.maintenance import run_maintenance
|
||||||
from app.sync.playlists import sync_all_playlists
|
from app.sync.playlists import sync_all_playlists
|
||||||
from app.sync.runner import (
|
from app.sync.runner import (
|
||||||
|
|
@ -271,9 +271,9 @@ def _demo_reset_job() -> None:
|
||||||
|
|
||||||
|
|
||||||
def _explore_cleanup_job() -> None:
|
def _explore_cleanup_job() -> None:
|
||||||
# Hard-delete explored-but-never-kept channels and their untouched ephemeral videos after
|
# Reclaim un-kept ephemeral discovery content (explored channels + live-search results) after
|
||||||
# the grace period. 0 quota (pure DB), so no quota attribution.
|
# their grace periods. 0 quota (pure DB), so no quota attribution.
|
||||||
_job("explore_cleanup", purge_unkept_explored)
|
_job("explore_cleanup", purge_ephemeral)
|
||||||
|
|
||||||
|
|
||||||
# job_id -> wrapper. The single source of truth for which jobs exist and how to run one,
|
# job_id -> wrapper. The single source of truth for which jobs exist and how to run one,
|
||||||
|
|
|
||||||
|
|
@ -143,3 +143,38 @@ def purge_unkept_explored(db: Session) -> dict:
|
||||||
"videos_deleted": videos_deleted,
|
"videos_deleted": videos_deleted,
|
||||||
"channels_deleted": channels_deleted,
|
"channels_deleted": channels_deleted,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def purge_unkept_search(db: Session, grace_days: int | None = None) -> dict:
|
||||||
|
"""Remove live-search result videos nobody kept, after a grace period — "as if never added".
|
||||||
|
|
||||||
|
A `via_search` video is KEPT (and spared) when anyone has watched / in-progress / hidden it
|
||||||
|
(a VideoState), saved/playlisted it (a PlaylistItem), or subscribes to its channel (it's then
|
||||||
|
organically owned). A bare SearchFind does NOT count — it's the ephemeral "shown in search"
|
||||||
|
marker itself — so a result you only glanced at is reclaimed. Deleting the video cascades its
|
||||||
|
SearchFinds. `grace_days=0` purges immediately (the admin "purge now" path); None uses config.
|
||||||
|
0 quota."""
|
||||||
|
grace = sysconfig.get_int(db, "search_grace_days") if grace_days is None else grace_days
|
||||||
|
cutoff = _now() - timedelta(days=grace)
|
||||||
|
state_exists = select(VideoState.id).where(VideoState.video_id == Video.id).exists()
|
||||||
|
pl_exists = select(PlaylistItem.id).where(PlaylistItem.video_id == Video.id).exists()
|
||||||
|
sub_exists = (
|
||||||
|
select(Subscription.id).where(Subscription.channel_id == Video.channel_id).exists()
|
||||||
|
)
|
||||||
|
deleted = db.execute(
|
||||||
|
delete(Video).where(
|
||||||
|
Video.via_search.is_(True),
|
||||||
|
Video.created_at < cutoff,
|
||||||
|
~state_exists,
|
||||||
|
~pl_exists,
|
||||||
|
~sub_exists,
|
||||||
|
)
|
||||||
|
).rowcount
|
||||||
|
db.commit()
|
||||||
|
return {"search_videos_deleted": deleted}
|
||||||
|
|
||||||
|
|
||||||
|
def purge_ephemeral(db: Session) -> dict:
|
||||||
|
"""The scheduler's discovery-cleanup pass: reclaim un-kept explored channels AND un-kept
|
||||||
|
live-search result videos in one run."""
|
||||||
|
return {**purge_unkept_explored(db), **purge_unkept_search(db)}
|
||||||
|
|
|
||||||
|
|
@ -51,9 +51,11 @@ SPECS: tuple[ConfigSpec, ...] = (
|
||||||
# --- Batch sizes ---
|
# --- Batch sizes ---
|
||||||
ConfigSpec("enrich_batch_size", "int", "batch", "enrich_batch_size", min=1, max=50),
|
ConfigSpec("enrich_batch_size", "int", "batch", "enrich_batch_size", min=1, max=50),
|
||||||
ConfigSpec("autotag_title_sample", "int", "batch", "autotag_title_sample", min=1, max=1_000),
|
ConfigSpec("autotag_title_sample", "int", "batch", "autotag_title_sample", min=1, max=1_000),
|
||||||
# --- Channel explore ---
|
# --- Discovery cleanup (search + explore ephemeral content) ---
|
||||||
# Days an explored-but-unsubscribed channel is kept before the cleanup job purges it.
|
# Days an explored-but-unsubscribed channel is kept before the cleanup job purges it.
|
||||||
ConfigSpec("explore_grace_days", "int", "explore", "explore_grace_days", min=0, max=3_650),
|
ConfigSpec("explore_grace_days", "int", "explore", "explore_grace_days", min=0, max=3_650),
|
||||||
|
# Days an un-kept live-search result video is kept before the cleanup job removes it.
|
||||||
|
ConfigSpec("search_grace_days", "int", "explore", "search_grace_days", min=0, max=3_650),
|
||||||
# --- YouTube Data API key (secret; optional — public reads prefer it over OAuth) ---
|
# --- YouTube Data API key (secret; optional — public reads prefer it over OAuth) ---
|
||||||
ConfigSpec("youtube_api_key", "str", "youtube", "youtube_api_key", secret=True),
|
ConfigSpec("youtube_api_key", "str", "youtube", "youtube_api_key", secret=True),
|
||||||
# Optional egress proxy for YouTube API calls (fixed-IP host for an IP-restricted key).
|
# Optional egress proxy for YouTube API calls (fixed-IP host for an IP-restricted key).
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue