Compare commits

...
3 Commits
Author SHA1 Message Date
peter 4b6db47083 test(catalog): the channel manager's local writes, committed through HTTP 2026-10-19 02:36:19 +02:00
peter 5229fd96be refactor(catalog): the HTTP layer reads through queries and writes through the service
Channels, search and sync do no database work in their routes any more: reads go through
catalog/queries, writes through catalog/service, and a live search's ingest is catalog/search.
What a catalog route needs from Reader (a playlist's channels, a user's tag links, the feed
projection of results) it asks reader/queries and reader/service. The scheduler and the runs page
read channel titles and the work queue through catalog/queries. Catalog is fenced below its HTTP
layer: its owner filters go through userscope, it reads no other unit's tables, and its ingest
jobs commit only where they are named.
2026-10-19 02:32:20 +02:00
peter 238e6f8a61 refactor(catalog): the unit's tables move into catalog/models
Channel, Subscription, Video, SearchFind, ExploredChannel and BlockedChannel (and LIVE_OR_UPCOMING)
leave app/models.py for app/catalog/models.py; app.tables registers them. The User↔Subscription
relationship names its other side as a type only. No schema change.
2026-10-19 02:03:14 +02:00
66 changed files with 1643 additions and 1027 deletions
+1 -1
View File
@@ -63,7 +63,7 @@ how it gets built. Concretely:
## Architecture shape (don't fight it)
`routes/*` (HTTP only: validation, auth-dependency, serialization) → domain packages `sync/`, `plex/`,
`downloads/`, `reader/`, `discover/`, `youtube/`, `issues/` (shared by routes + scheduler + worker) → flat core (`auth`, `apitokens`, `quota`, `jobruns`,
`downloads/`, `reader/`, `discover/`, `catalog/`, `youtube/`, `issues/` (shared by routes + scheduler + worker) → flat core (`auth`, `apitokens`, `quota`, `jobruns`,
`sysconfig`, `state`, `audit`, `notifications`, `userscope`, `ratelimit`, `bodylimit`, `progress`, `collation`,
`security`, `logs`, `logsink`, `diagnostics`, `debugsession`) → `models/db/config`. **Dependency direction is one-way: core never imports routes.**
Frontend: `main.tsx` provider tree → `App.tsx` shell (a shared `/watch` link boots from its own
-86
View File
@@ -640,48 +640,6 @@
}
}
],
"./app/routes/channels.py": [
{
"code": "reportReturnType",
"range": {
"startColumn": 11,
"endColumn": 11,
"lineCount": 6
}
},
{
"code": "reportArgumentType",
"range": {
"startColumn": 8,
"endColumn": 15,
"lineCount": 1
}
},
{
"code": "reportArgumentType",
"range": {
"startColumn": 49,
"endColumn": 56,
"lineCount": 1
}
},
{
"code": "reportAttributeAccessIssue",
"range": {
"startColumn": 38,
"endColumn": 44,
"lineCount": 1
}
},
{
"code": "reportAttributeAccessIssue",
"range": {
"startColumn": 33,
"endColumn": 39,
"lineCount": 1
}
}
],
"./app/routes/discover.py": [
{
"code": "reportInvalidTypeForm",
@@ -967,14 +925,6 @@
}
],
"./app/routes/scheduler.py": [
{
"code": "reportArgumentType",
"range": {
"startColumn": 42,
"endColumn": 55,
"lineCount": 1
}
},
{
"code": "reportOptionalMemberAccess",
"range": {
@@ -992,42 +942,6 @@
}
}
],
"./app/routes/search.py": [
{
"code": "reportArgumentType",
"range": {
"startColumn": 25,
"endColumn": 31,
"lineCount": 1
}
},
{
"code": "reportAttributeAccessIssue",
"range": {
"startColumn": 6,
"endColumn": 14,
"lineCount": 1
}
}
],
"./app/routes/sync.py": [
{
"code": "reportArgumentType",
"range": {
"startColumn": 42,
"endColumn": 65,
"lineCount": 1
}
},
{
"code": "reportAttributeAccessIssue",
"range": {
"startColumn": 42,
"endColumn": 50,
"lineCount": 1
}
}
],
"./app/sync/explore.py": [
{
"code": "reportAttributeAccessIssue",
View File
+265
View File
@@ -0,0 +1,265 @@
"""Catalog's tables: the shared YouTube world every user reads — channels and their videos — and
each user's footprint in it: what they subscribe to, block, explored or found by search. Only this
unit writes them; another unit reads them along a declared edge, or asks this unit's service."""
from datetime import datetime
from typing import TYPE_CHECKING
from sqlalchemy import (
BigInteger,
Boolean,
Computed,
DateTime,
ForeignKey,
Integer,
JSON,
String,
Text,
UniqueConstraint,
func,
)
from sqlalchemy.dialects.postgresql import TSVECTOR
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.db import Base
from app.models import TimestampMixin
if TYPE_CHECKING:
from app.models import User
class Channel(Base, TimestampMixin):
"""A YouTube channel. Shared across all users (one channel's videos are the same
for everyone), so its expensive metadata is fetched and stored only once."""
__tablename__ = "channels"
id: Mapped[str] = mapped_column(String(32), primary_key=True) # "UC..." channel id
title: Mapped[str | None] = mapped_column(String(255))
handle: Mapped[str | None] = mapped_column(String(255))
description: Mapped[str | None] = mapped_column(Text)
thumbnail_url: Mapped[str | None] = mapped_column(String(1024))
uploads_playlist_id: Mapped[str | None] = mapped_column(String(64))
subscriber_count: Mapped[int | None] = mapped_column(BigInteger)
video_count: Mapped[int | None] = mapped_column(BigInteger)
# Channel "About" metadata (fetched with the same channels.list call, no extra quota):
# lifetime view count, the channel's join date, banner image, and external links.
total_view_count: Mapped[int | None] = mapped_column(BigInteger)
published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
banner_url: Mapped[str | None] = mapped_column(String(1024))
external_links: Mapped[list | None] = mapped_column(JSON)
topic_categories: Mapped[list | None] = mapped_column(JSON)
default_language: Mapped[str | None] = mapped_column(String(16))
country: Mapped[str | None] = mapped_column(String(8))
# Creator keyword tags (brandingSettings.channel.keywords) — a single space-separated string,
# multi-word tags quoted; the client parses it into chips for the About tab.
keywords: Mapped[str | None] = mapped_column(Text)
# Sync bookkeeping.
details_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
recent_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
last_rss_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
# When the newest uploads were last read through the Data API because the RSS feed would not
# answer (`sync/runner.py` `run_feed_fallback`): the once-per-`rss_fallback_hours` clock.
feed_fallback_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
backfill_cursor: Mapped[str | None] = mapped_column(String(128))
backfill_done: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
# True when this channel only entered the catalog because a live YouTube search
# surfaced one of its videos (nobody subscribes to it / it has no organic upload sync).
# Lets the Library hide search-only "noise" by default.
from_search: Mapped[bool] = mapped_column(
Boolean, default=False, server_default="false", index=True
)
# True when this channel exists only because a user EXPLORED it (opened its page without
# subscribing) — an ephemeral, per-explorer addition. The cleanup job hard-deletes such a
# channel (and its via_explore videos) once nobody keeps it (no subscription, no live
# ExploredChannel row, no per-user state) after a grace period. Cleared when someone
# subscribes (the "keep" signal promotes it to organic).
from_explore: Mapped[bool] = mapped_column(
Boolean, default=False, server_default="false", index=True
)
videos: Mapped[list["Video"]] = relationship(back_populates="channel")
subscriptions: Mapped[list["Subscription"]] = relationship(back_populates="channel")
class Subscription(Base, TimestampMixin):
"""Per-user link to a channel, with the user's own priority/visibility overrides."""
__tablename__ = "subscriptions"
__table_args__ = (UniqueConstraint("user_id", "channel_id", name="uq_user_channel"),)
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
channel_id: Mapped[str] = mapped_column(
ForeignKey("channels.id", ondelete="CASCADE"), index=True
)
# YouTube's subscription resource id, needed to unsubscribe via the API.
yt_subscription_id: Mapped[str | None] = mapped_column(String(128))
priority: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
hidden: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
# Per-user opt-in to full-history (deep) backfill of this channel. Default off so a
# new user's unique channels don't trigger a big shared-quota burst; the deep
# scheduler only picks up channels at least one user has requested.
deep_requested: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
subscribed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
user: Mapped["User"] = relationship(back_populates="subscriptions")
channel: Mapped["Channel"] = relationship(back_populates="subscriptions")
# Video.live_status values that mean "currently a live/upcoming broadcast" — transient states
# hidden from feeds/search and revisited by the refresh job until they settle into a VOD.
LIVE_OR_UPCOMING = ("live", "upcoming")
class Video(Base, TimestampMixin):
"""A YouTube video. Shared across users; per-user state lives elsewhere."""
__tablename__ = "videos"
id: Mapped[str] = mapped_column(String(16), primary_key=True) # video id
# No standalone channel_id index: the composite `ix_videos_channel_pub_cover`
# (channel_id, published_at DESC NULLS LAST, id) from migration 0095 leads with channel_id, so it
# serves every channel_id lookup (FK cascade, sync upsert, channel-detail) and the per-channel feed
# sort. Its INCLUDEd columns make the channel manager's sums and the feed sidebar's counts Index Only
# Scans (no heap): a column those queries start reading belongs in it (test_channel_list_index_only).
# Like the search_vector GIN, this composite is migration-defined, not an inline `index=True`.
channel_id: Mapped[str] = mapped_column(ForeignKey("channels.id", ondelete="CASCADE"))
title: Mapped[str | None] = mapped_column(Text) # normalized for display (see app.titles)
original_title: Mapped[str | None] = mapped_column(Text) # raw YouTube title, preserved
description: Mapped[str | None] = mapped_column(Text)
# NOT NULL (migration 0068): in reality every video carries a publish date, and a nullable key
# forced the feed's keyset predicate into a non-sargable `... OR published_at IS NULL` disjunction
# that blocked the ix_videos_published_at_desc range (deep pages scanned from the top and filtered —
# PERF-15). server_default now() is the safety net for the rare date-less ingest (the enrich fills
# the real date); the ix_videos_published_at index stays for the `oldest` sort.
published_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), index=True, nullable=False, server_default=func.now()
)
thumbnail_url: Mapped[str | None] = mapped_column(String(1024))
# The feed's views / duration / title sorts each ride a migration-defined index in their exact
# ORDER BY, one per direction (and per ICU collation for title): migration 0096, held to
# `feedquery.order_by` by test_feed_sort_indexes.
duration_seconds: Mapped[int | None] = mapped_column(Integer)
view_count: Mapped[int | None] = mapped_column(BigInteger)
like_count: Mapped[int | None] = mapped_column(BigInteger)
category_id: Mapped[int | None] = mapped_column(Integer)
topic_categories: Mapped[list | None] = mapped_column(JSON)
default_language: Mapped[str | None] = mapped_column(String(16))
detected_language: Mapped[str | None] = mapped_column(String(16))
# Creator-supplied keyword tags (snippet.tags joined by spaces) — fetched for free with the
# snippet we already request. Indexed into search_vector so the feed search matches the
# uploader's own keywords, not just words in the title.
keywords: Mapped[str | None] = mapped_column(Text)
# Accumulated distinct live-search queries that surfaced this video (across all users) —
# folds YouTube's own relevance judgement into the local search: a video YouTube returned
# for "ti amo magyarul" becomes findable by that query locally even if its title lacks it.
search_terms: Mapped[str | None] = mapped_column(Text)
# No btree here or on the flag columns below: migration 0069 (PERF-3) dropped the never-picked
# boolean btrees and replaced the useful ones with partial indexes over the target subset
# (via_search=true, NOT shorts_probed, unavailable_since IS NOT NULL) — migration-defined, like the
# search_vector GIN.
is_short: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
# Whether the youtube.com/shorts/<id> probe has run for this video.
shorts_probed: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
# True when this video first entered the catalog via a live YouTube search (not an
# organic channel/playlist sync). Used to hide search-discovered "noise" from the
# Library by default; left False by the organic ingest paths.
via_search: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
# True when this video entered the catalog because a user EXPLORED its (un-subscribed)
# channel. Like via_search it's hidden from the shared Library by default, but it's
# additionally PER-USER private: a via_explore video is only visible to users who
# explored or subscribe to its channel (a gate in the feed query), so one user's
# curiosity never floods everyone else's catalog. Left False by the organic ingest paths.
via_explore: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
# none | live | upcoming | premiere | was_live
live_status: Mapped[str] = mapped_column(
String(16), default="none", server_default="none", index=True
)
enriched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
# videos.list?part=status fields (populated during enrichment); used by the
# maintenance/validation job to decide whether a video is playable anywhere.
embeddable: Mapped[bool | None] = mapped_column(Boolean)
privacy_status: Mapped[str | None] = mapped_column(String(16)) # public|unlisted|private
upload_status: Mapped[str | None] = mapped_column(
String(16)
) # processed|failed|rejected|deleted
# Maintenance/validation lifecycle. last_checked_at drives the rolling re-validation
# (least-recently-checked first); unavailable_since marks a video as currently
# unplayable (hidden from the feed immediately, hard-deleted after a grace period);
# unavailable_reason records why (deleted|private|paywalled|abandoned|stuck_live).
# last_checked_at: no plain btree — migration 0069 (PERF-18) replaced it with the partial
# ix_videos_recheck_order (last_checked_at NULLS FIRST WHERE unavailable_since IS NULL) that the
# revalidate job's exact ORDER BY rides; unavailable_since is a partial WHERE IS NOT NULL (PERF-3).
last_checked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
unavailable_since: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
unavailable_reason: Mapped[str | None] = mapped_column(String(24))
# Weighted full-text search document (DB-generated, never written by the app): title (A) >
# creator keywords + search queries (B) > truncated description (C). ts_rank honours the
# weights so title matches outrank description matches. Backed by a GIN index (migration
# 0032). The accent-insensitive `unaccent_simple` config comes from migration 0031.
search_vector: Mapped[object | None] = mapped_column(
TSVECTOR,
Computed(
"setweight(to_tsvector('public.unaccent_simple', coalesce(title, '')), 'A') || "
"setweight(to_tsvector('public.unaccent_simple', coalesce(keywords, '') || ' ' || "
"coalesce(search_terms, '')), 'B') || "
"setweight(to_tsvector('public.unaccent_simple', left(coalesce(description, ''), 1000)), 'C')",
persisted=True,
),
)
channel: Mapped["Channel"] = relationship(back_populates="videos")
class SearchFind(Base, TimestampMixin):
"""Per-user record of a video the user surfaced via their own live YouTube search.
Lets "videos I searched for" appear in that user's own (Mine) feed and the Mine Source
filter — distinct from the GLOBAL `Video.via_search` provenance, which marks a video as
search-discovered by *anyone* and drives the shared Library's Source filter."""
__tablename__ = "search_finds"
__table_args__ = (UniqueConstraint("user_id", "video_id", name="uq_user_search_video"),)
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
video_id: Mapped[str] = mapped_column(ForeignKey("videos.id", ondelete="CASCADE"), index=True)
class ExploredChannel(Base, TimestampMixin):
"""Per-user record of a channel the user opened the page of WITHOUT subscribing.
Exploring an un-subscribed channel ingests its recent uploads (flagged via_explore) so the
user can browse them, but keeps them out of everyone else's catalog. This row (a) gates the
explorer's access to those ephemeral videos in the feed query and (b) drives the grace-period
purge: a from_explore channel with no live ExploredChannel row (and no subscription / no
per-user state on its videos) is hard-deleted by the cleanup job. Subscribing is the "keep"
signal — it promotes the channel to organic and these rows become irrelevant. `created_at`
(from TimestampMixin) is the grace clock; a re-visit refreshes it."""
__tablename__ = "explored_channels"
__table_args__ = (UniqueConstraint("user_id", "channel_id", name="uq_user_explored_channel"),)
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
channel_id: Mapped[str] = mapped_column(
ForeignKey("channels.id", ondelete="CASCADE"), index=True
)
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"))
channel_id: Mapped[str] = mapped_column(
ForeignKey("channels.id", ondelete="CASCADE"), index=True
)
+296
View File
@@ -0,0 +1,296 @@
"""Catalog's reads: channels and videos, and each user's footprint in them (subscriptions, blocks,
explorations), for the HTTP layer and for the core code that asks about them.
A user's row is read through `userscope`, so someone else's is the same answer as a missing one.
Catalog does not read another unit's tables: what it needs from Reader (a playlist's channels, a
user's tag links) it asks `reader.queries`.
"""
from datetime import datetime
from sqlalchemy import Select, and_, case, func, or_, select
from sqlalchemy.orm import Session
from app.catalog.models import (
LIVE_OR_UPCOMING,
BlockedChannel,
Channel,
ExploredChannel,
Subscription,
Video,
)
from app.collation import collated
from app.models import User
from app.sync.explore import followed_channel
from app.userscope import owned_by
# --- channels --------------------------------------------------------------------------------
def channel(db: Session, channel_id: str) -> Channel | None:
return db.get(Channel, channel_id)
def channel_title(db: Session, channel_id: str) -> str | None:
return db.scalar(select(Channel.title).where(Channel.id == channel_id))
def channel_titles(db: Session, ids: set[str]) -> dict[str, str]:
"""Each channel's title by id; a channel with no title yet is left out."""
if not ids:
return {}
rows = db.execute(select(Channel.id, Channel.title).where(Channel.id.in_(ids)))
return {cid: title for cid, title in rows.tuples() if title is not None}
def channel_aggregates(channel_ids: list[str]) -> Select:
"""Per-channel stored count, last upload and total duration, in one grouped pass. Every column it
reads is in `ix_videos_channel_pub_cover` (migration 0095), so Postgres answers it from the index
alone; a test holds that (test_channel_list_index_only)."""
return (
select(
Video.channel_id,
func.count(Video.id),
func.max(Video.published_at),
func.coalesce(func.sum(Video.duration_seconds), 0),
)
.where(Video.channel_id.in_(channel_ids))
.group_by(Video.channel_id)
)
def aggregates(db: Session, channel_ids: list[str]) -> list[tuple[str, int, datetime | None, int]]:
"""`channel_aggregates`, run: (channel id, stored videos, last upload, total seconds)."""
if not channel_ids:
return []
return list(db.execute(channel_aggregates(channel_ids)).tuples())
def video_counts(db: Session, channel_id: str) -> tuple[int, int, int]:
"""A channel's stored videos, its Shorts and its live/upcoming ones, in one grouped pass."""
n_all, n_short, n_live = db.execute(
select(
func.count(Video.id),
func.count(Video.id).filter(Video.is_short.is_(True)),
func.count(Video.id).filter(Video.live_status.in_(LIVE_OR_UPCOMING)),
).where(Video.channel_id == channel_id)
).one()
return int(n_all or 0), int(n_short or 0), int(n_live or 0)
def deep_queued(db: Session, channel_ids: list[str]) -> set[str]:
"""Which of these channels at least one user (any user) has asked full history for."""
if not channel_ids:
return set()
return set(
db.execute(
select(Subscription.channel_id)
.where(
Subscription.channel_id.in_(channel_ids),
Subscription.deep_requested.is_(True),
)
.distinct()
).scalars()
)
def followed(db: Session, channel_id: str) -> bool:
"""Whether anyone subscribes to the channel."""
return (
db.execute(
select(Subscription.id).where(Subscription.channel_id == channel_id).limit(1)
).first()
is not None
)
# --- a user's subscriptions, blocks and explorations ----------------------------------------
def subscriptions(db: Session, user: User, lang: str) -> list[tuple[Subscription, Channel]]:
"""The user's subscriptions with their channels, highest priority first, then by title in the
reader's alphabet."""
return list(
db.execute(
select(Subscription, Channel)
.join(Channel, Channel.id == Subscription.channel_id)
.where(owned_by(Subscription.user_id, user))
.order_by(Subscription.priority.desc(), collated(Channel.title, lang))
).tuples()
)
def subscription(db: Session, user: User, channel_id: str) -> Subscription | None:
return db.execute(
select(Subscription).where(
owned_by(Subscription.user_id, user), Subscription.channel_id == channel_id
)
).scalar_one_or_none()
def subscribed_channels(user: User):
"""The channel ids the user subscribes to, as a subquery."""
return select(Subscription.channel_id).where(owned_by(Subscription.user_id, user))
def exploration(db: Session, user: User, channel_id: str) -> ExploredChannel | None:
return db.execute(
select(ExploredChannel).where(
owned_by(ExploredChannel.user_id, user), ExploredChannel.channel_id == channel_id
)
).scalar_one_or_none()
def is_blocked(db: Session, user: User, channel_id: str) -> bool:
return (
db.execute(
select(BlockedChannel.id).where(
owned_by(BlockedChannel.user_id, user), BlockedChannel.channel_id == channel_id
)
).first()
is not None
)
def blocked_ids(db: Session, user: User) -> set[str]:
return set(
db.scalars(select(BlockedChannel.channel_id).where(owned_by(BlockedChannel.user_id, user)))
)
def blocked_channels(db: Session, user: User, lang: str) -> list[Channel]:
return list(
db.execute(
select(Channel)
.join(BlockedChannel, BlockedChannel.channel_id == Channel.id)
.where(owned_by(BlockedChannel.user_id, user))
.order_by(collated(Channel.title, lang))
).scalars()
)
# --- sync progress ---------------------------------------------------------------------------
def catalog_totals(db: Session, user: User) -> dict:
"""The instance's catalog size and this user's subscription count (the sync status card)."""
return {
"subscriptions": db.scalar(
select(func.count())
.select_from(Subscription)
.where(owned_by(Subscription.user_id, user))
),
"channels_total": db.scalar(select(func.count()).select_from(Channel)),
"channels_backfilling": db.scalar(
select(func.count()).select_from(Channel).where(Channel.backfill_done.is_(False))
),
"videos_total": db.scalar(select(func.count()).select_from(Video)),
"pending_enrich": db.scalar(
select(func.count()).select_from(Video).where(Video.enriched_at.is_(None))
),
}
def my_progress(db: Session, user: User) -> tuple[int, int, int, int]:
"""Over the channels this user subscribes to: how many, how many have details, how many their
recent uploads, how many their whole history."""
total, details_synced, recent_synced, all_deep_done = db.execute(
select(
func.count(Channel.id),
func.count(Channel.details_synced_at),
func.count(Channel.recent_synced_at),
func.sum(case((Channel.backfill_done.is_(True), 1), else_=0)),
)
.select_from(Channel)
.join(
Subscription,
and_(Subscription.channel_id == Channel.id, owned_by(Subscription.user_id, user)),
)
).one()
return total or 0, details_synced or 0, recent_synced or 0, int(all_deep_done or 0)
def my_deep_requested(db: Session, user: User) -> list[Channel]:
"""The channels this user asked full history for."""
return list(
db.execute(
select(Channel).join(
Subscription,
and_(
Subscription.channel_id == Channel.id,
owned_by(Subscription.user_id, user),
Subscription.deep_requested.is_(True),
),
)
).scalars()
)
def my_video_count(db: Session, user: User) -> int:
"""The videos of the channels this user subscribes to and has not hidden."""
return (
db.scalar(
select(func.count(Video.id)).join(
Subscription,
and_(
Subscription.channel_id == Video.channel_id,
owned_by(Subscription.user_id, user),
Subscription.hidden.is_(False),
),
)
)
or 0
)
def video_total(db: Session) -> int:
return db.scalar(select(func.count(Video.id))) or 0
def deep_backlog(db: Session) -> list[Channel]:
"""Every channel someone asked full history for that does not have it yet."""
deep_requested = (
select(Subscription.id)
.where(Subscription.channel_id == Channel.id, Subscription.deep_requested.is_(True))
.exists()
)
return list(
db.execute(
select(Channel).where(Channel.backfill_done.is_(False), deep_requested)
).scalars()
)
def work_queue(db: Session) -> dict:
"""What the catalog's jobs still have queued, instance-wide (the scheduler card)."""
return {
# The same selection run_recent_backfill works through, or the card counts channels the
# job will never touch and never reaches zero.
"channels_recent_pending": db.scalar(
select(func.count())
.select_from(Channel)
.where(Channel.recent_synced_at.is_(None), followed_channel())
)
or 0,
"videos_pending_enrich": db.scalar(
select(func.count()).select_from(Video).where(Video.enriched_at.is_(None))
)
or 0,
"videos_pending_shorts": db.scalar(
select(func.count())
.select_from(Video)
.where(Video.shorts_probed.is_(False), Video.enriched_at.is_not(None))
)
or 0,
"videos_live_refresh": db.scalar(
select(func.count())
.select_from(Video)
.where(
or_(
Video.live_status.in_(("live", "upcoming")),
and_(Video.live_status == "was_live", Video.duration_seconds.is_(None)),
)
)
)
or 0,
}
+232
View File
@@ -0,0 +1,232 @@
"""Live YouTube search, the catalog's side: what a page of results leaves in the shared catalog.
A result becomes a catalog row so it renders with the normal feed cards and the in-app player and
gains per-user state. The ingest runs in steps with a YouTube call between each, and each step's
rows are committed before the next call: an enrich that fails rolls back to the stubs, which stay
(the FK targets for what follows, backfilled later). Results are flagged (`videos.via_search`,
`channels.from_search`) so the Library can hide this search-discovered noise; Shorts and
live/upcoming videos are never kept. The HTTP layer (`routes/search.py`) runs the paging, the caps
and the refusals.
"""
from collections.abc import Sequence
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from sqlalchemy import delete, func, select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.engine import CursorResult
from sqlalchemy.orm import Session
from app import logs, quota, sysconfig
from app.catalog import queries
from app.catalog.models import LIVE_OR_UPCOMING, Channel, SearchFind, Subscription, Video
from app.models import User
from app.reader import queries as reader_queries
from app.sync.subscriptions import apply_channel_details
from app.sync.videos import apply_video_details, insert_stubs, parse_dt
from app.userscope import owned_by
from app.utils import like_escape
from app.youtube.client import YouTubeClient, YouTubeError
from app.youtube.shorts import make_client, probe_is_short
log = logs.get_logger(__name__)
def classify_shorts(db: Session, videos: Sequence[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 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, new_channel_ids)
db.commit()
except YouTubeError:
# The stubs remain and the details can be backfilled later — but until they are, those
# channels sit in the results with no thumbnail and no subscriber count, looking like
# channels that have neither.
log.warning(
"the channels found by a search could not be enriched",
extra=logs.event("search.channel_enrich_failed", channels=len(new_channel_ids)),
)
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:
# Every candidate then stays unenriched, and an unenriched video cannot be judged live or
# Short — so it is dropped. A working search returns nothing and reads as "no results".
log.warning(
"the videos found by a search could not be enriched",
extra=logs.event("search.video_enrich_failed", candidates=len(ids)),
)
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 _fold_term(db: Session, video_ids: list[str], term: str) -> None:
"""Append `term` to each video's space-joined `search_terms`, unless it is already one of them.
"Already one of them" is a whole-term match, as a literal: a substring test skipped `cat` on a
video already found for `category`, and an unescaped one read the `_` in `my_channel` as a
wildcard — either way the term was never folded in, and local search lost that relevance."""
padded = func.concat(" ", func.coalesce(Video.search_terms, ""), " ")
db.execute(
update(Video)
.where(Video.id.in_(video_ids))
.where(padded.notilike(f"% {like_escape(term)} %"))
.values(
search_terms=func.trim(
func.coalesce(Video.search_terms, "").op("||")(" ").op("||")(term)
)
)
)
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
have = set(
db.scalars(
select(Video.id).where(
Video.id.in_(ids), Video.channel_id.in_(queries.subscribed_channels(user))
)
)
) | reader_queries.held_by(db, user, ids)
return [i for i in ids if i not in have] + [i for i in ids if i in have]
def record_finds(db: Session, user: User, video_ids: list[str], term: str) -> None:
"""These results as THIS user's search finds (the Mine feed, the Source filter), and the term
folded into each one's `search_terms` so local full-text search inherits YouTube's relevance.
Idempotent."""
db.execute(
pg_insert(SearchFind)
.values([{"user_id": user.id, "video_id": vid} for vid in video_ids])
.on_conflict_do_nothing(index_elements=["user_id", "video_id"])
)
_fold_term(db, video_ids, term)
def clear(db: Session, user: User, video_ids: list[str]) -> int:
"""Discard these results "as if never added": drop this user's finds of them, then delete the
ones nobody kept — a `via_search` video no reader holds (`reader.queries.kept_by_a_reader`),
whose channel nobody subscribes to, and that no other user still has as a find. Returns how
many videos went."""
db.execute(
delete(SearchFind).where(
owned_by(SearchFind.user_id, user), SearchFind.video_id.in_(video_ids)
)
)
subscribed = select(Subscription.id).where(Subscription.channel_id == Video.channel_id).exists()
other_find = select(SearchFind.id).where(SearchFind.video_id == Video.id).exists()
result = db.execute(
delete(Video).where(
Video.id.in_(video_ids),
Video.via_search.is_(True),
~reader_queries.kept_by_a_reader(Video.id),
~subscribed,
~other_find,
)
)
return result.rowcount if isinstance(result, CursorResult) else 0
+133
View File
@@ -0,0 +1,133 @@
"""Catalog's writes a request makes: a user's subscription settings, what they explore or block, a
subscription recorded or dropped, a channel's sync markers reset.
Every function here flushes and leaves the commit to its caller (the handler). The YouTube side of a
subscribe or an unsubscribe is the handler's (it needs the user's client and its refusals); what
lands here is the local record of it. The ingest jobs (`sync/`, `catalog/search`) write the
catalog's shared rows themselves.
"""
from datetime import datetime, timezone
from sqlalchemy import delete, update
from sqlalchemy.engine import CursorResult
from sqlalchemy.orm import Session
from app.catalog import queries
from app.catalog.models import BlockedChannel, Channel, ExploredChannel, Subscription, Video
from app.models import User
from app.userscope import owned_by
def remember_own_channel(user: User, channel_id: str) -> None:
"""The user's own YouTube channel, so discovery never offers it to them."""
user.yt_channel_id = channel_id
def update_subscription(
db: Session,
sub: Subscription,
*,
priority: int | None = None,
hidden: bool | None = None,
deep_requested: bool | None = None,
) -> bool:
"""Change the settings given (None leaves one as it is). Returns whether full history was just
turned ON — the caller then pulls the channel now rather than at the scheduler's next tick."""
turned_on = False
if priority is not None:
sub.priority = priority
if hidden is not None:
sub.hidden = hidden
if deep_requested is not None:
# Turning it off won't undo already-fetched videos; it stops further deep paging if nobody
# else still wants it.
turned_on = deep_requested and not sub.deep_requested
sub.deep_requested = deep_requested
db.flush()
return turned_on
def set_deep_all(db: Session, user: User, on: bool) -> int:
"""Every one of the user's channels in (or out of) full-history backfill; how many rows moved."""
result = db.execute(
update(Subscription).where(owned_by(Subscription.user_id, user)).values(deep_requested=on)
)
return result.rowcount if isinstance(result, CursorResult) else 0
def explore(db: Session, user: User, channel: Channel) -> None:
"""Record (or refresh the grace clock of) the user's exploration of a channel, and flag the
channel ephemeral unless someone follows it — the cleanup job's keep rule reads both."""
existing = queries.exploration(db, user, channel.id)
if existing is None:
db.add(ExploredChannel(user_id=user.id, channel_id=channel.id))
else:
existing.created_at = datetime.now(timezone.utc)
if not channel.from_explore and not queries.followed(db, channel.id):
channel.from_explore = True
db.flush()
def reset_backfill(db: Session, channel: Channel) -> None:
"""Fetch the channel again from scratch: its backfill markers cleared, and full history asked
for on EVERY subscription to it — the deep scheduler gates on an EXISTS over those, so the admin
who resets it need not subscribe themselves."""
channel.backfill_done = False
channel.backfill_cursor = None
channel.recent_synced_at = None
db.execute(
update(Subscription)
.where(Subscription.channel_id == channel.id)
.values(deep_requested=True)
)
db.flush()
def record_subscription(db: Session, user: User, channel: Channel, yt_subscription_id: str) -> None:
"""The subscription YouTube now holds, recorded here. Subscribing is the "keep" signal: an
explored channel becomes an organic one, so the cleanup job leaves it alone and its videos join
the shared Library like any subscribed channel's uploads."""
db.add(
Subscription(
user_id=user.id,
channel_id=channel.id,
yt_subscription_id=yt_subscription_id or None,
subscribed_at=datetime.now(timezone.utc),
)
)
if channel.from_explore:
channel.from_explore = False
db.execute(
update(Video)
.where(Video.channel_id == channel.id, Video.via_explore.is_(True))
.values(via_explore=False)
)
db.flush()
def drop_subscription(db: Session, sub: Subscription) -> None:
db.delete(sub)
db.flush()
def block(db: Session, user: User, channel_id: str) -> None:
"""Block a channel for this user (idempotent), and end any exploration of it so the cleanup job
can reclaim it."""
if queries.is_blocked(db, user, channel_id):
return
db.add(BlockedChannel(user_id=user.id, channel_id=channel_id))
db.execute(
delete(ExploredChannel).where(
owned_by(ExploredChannel.user_id, user), ExploredChannel.channel_id == channel_id
)
)
db.flush()
def unblock(db: Session, user: User, channel_id: str) -> None:
db.execute(
delete(BlockedChannel).where(
owned_by(BlockedChannel.user_id, user), BlockedChannel.channel_id == channel_id
)
)
+1 -1
View File
@@ -22,7 +22,7 @@ to the client, which has both the native payload and the translations.
"""
from app.discover.budget import remaining_seconds
from app.models import Channel
from app.catalog.models import Channel
from app.reader.models import Playlist
# Plex artwork is a poster for the things you browse and a landscape still for the things you play.
+2 -1
View File
@@ -22,7 +22,8 @@ from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.discover import cards, queries
from app.models import Channel, PlexItem, PlexShow, User, Video
from app.models import PlexItem, PlexShow, User
from app.catalog.models import Channel, Video
from app.reader.models import ComfortPin, Playlist
from app.plex import rollup
from app.plex.cards import SHOW_NEW, leaf_card, show_card
+2 -1
View File
@@ -29,7 +29,8 @@ from app.feedquery import (
serialize,
sort_keys,
)
from app.models import Channel, PlexItem, PlexShow, PlexState, User, Video
from app.models import PlexItem, PlexShow, PlexState, User
from app.catalog.models import Channel, Video
from app.reader.models import PlaylistItem, VideoState
from app.plex import rollup
from app.plex.cards import display_title, leaf_card, movie_card, show_card
+2 -1
View File
@@ -24,7 +24,8 @@ from app.downloads import credentials
from app.downloads import edit as editmod
from app.downloads import folders, formats, mediakind, quota, storage
from app.downloads.status import HOLDING, JobKind, JobStatus
from app.models import DISPLAY_TEXT_MAX, Channel, PlexItem, PlexShow, User, Video
from app.models import DISPLAY_TEXT_MAX, PlexItem, PlexShow, User
from app.catalog.models import Channel, Video
from app.downloads.models import (
DownloadJob,
DownloadProfile,
+2 -2
View File
@@ -17,14 +17,14 @@ from sqlalchemy import BigInteger, Select, and_, case, cast, false, func, or_, s
from sqlalchemy.orm import Session, aliased
from app.collation import collated
from app.models import (
from app.models import User
from app.catalog.models import (
LIVE_OR_UPCOMING,
BlockedChannel,
Channel,
ExploredChannel,
SearchFind,
Subscription,
User,
Video,
)
from app.reader.models import ChannelTag, Playlist, PlaylistItem, Tag, VideoState
+4 -237
View File
@@ -1,4 +1,5 @@
from datetime import date, datetime
from typing import TYPE_CHECKING
from sqlalchemy import (
BigInteger,
@@ -26,6 +27,9 @@ from app.db import Base
from app.runstatus import RunStatus, RunTrigger, sql_in
from app.utils import canonical_email
if TYPE_CHECKING:
from app.catalog.models import Subscription
# Every user-supplied display string in this schema shares one width, and it lives HERE so the
# columns below are declared with it rather than with a literal: a name a person typed goes into
# `User.display_name` (their Google profile name), `DownloadJob.display_name` and
@@ -220,243 +224,6 @@ class Invite(Base, CanonicalEmailMixin):
lang: Mapped[str | None] = mapped_column(String(8))
class Channel(Base, TimestampMixin):
"""A YouTube channel. Shared across all users (one channel's videos are the same
for everyone), so its expensive metadata is fetched and stored only once."""
__tablename__ = "channels"
id: Mapped[str] = mapped_column(String(32), primary_key=True) # "UC..." channel id
title: Mapped[str | None] = mapped_column(String(255))
handle: Mapped[str | None] = mapped_column(String(255))
description: Mapped[str | None] = mapped_column(Text)
thumbnail_url: Mapped[str | None] = mapped_column(String(1024))
uploads_playlist_id: Mapped[str | None] = mapped_column(String(64))
subscriber_count: Mapped[int | None] = mapped_column(BigInteger)
video_count: Mapped[int | None] = mapped_column(BigInteger)
# Channel "About" metadata (fetched with the same channels.list call, no extra quota):
# lifetime view count, the channel's join date, banner image, and external links.
total_view_count: Mapped[int | None] = mapped_column(BigInteger)
published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
banner_url: Mapped[str | None] = mapped_column(String(1024))
external_links: Mapped[list | None] = mapped_column(JSON)
topic_categories: Mapped[list | None] = mapped_column(JSON)
default_language: Mapped[str | None] = mapped_column(String(16))
country: Mapped[str | None] = mapped_column(String(8))
# Creator keyword tags (brandingSettings.channel.keywords) — a single space-separated string,
# multi-word tags quoted; the client parses it into chips for the About tab.
keywords: Mapped[str | None] = mapped_column(Text)
# Sync bookkeeping.
details_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
recent_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
last_rss_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
# When the newest uploads were last read through the Data API because the RSS feed would not
# answer (`sync/runner.py` `run_feed_fallback`): the once-per-`rss_fallback_hours` clock.
feed_fallback_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
backfill_cursor: Mapped[str | None] = mapped_column(String(128))
backfill_done: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
# True when this channel only entered the catalog because a live YouTube search
# surfaced one of its videos (nobody subscribes to it / it has no organic upload sync).
# Lets the Library hide search-only "noise" by default.
from_search: Mapped[bool] = mapped_column(
Boolean, default=False, server_default="false", index=True
)
# True when this channel exists only because a user EXPLORED it (opened its page without
# subscribing) — an ephemeral, per-explorer addition. The cleanup job hard-deletes such a
# channel (and its via_explore videos) once nobody keeps it (no subscription, no live
# ExploredChannel row, no per-user state) after a grace period. Cleared when someone
# subscribes (the "keep" signal promotes it to organic).
from_explore: Mapped[bool] = mapped_column(
Boolean, default=False, server_default="false", index=True
)
videos: Mapped[list["Video"]] = relationship(back_populates="channel")
subscriptions: Mapped[list["Subscription"]] = relationship(back_populates="channel")
class Subscription(Base, TimestampMixin):
"""Per-user link to a channel, with the user's own priority/visibility overrides."""
__tablename__ = "subscriptions"
__table_args__ = (UniqueConstraint("user_id", "channel_id", name="uq_user_channel"),)
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
channel_id: Mapped[str] = mapped_column(
ForeignKey("channels.id", ondelete="CASCADE"), index=True
)
# YouTube's subscription resource id, needed to unsubscribe via the API.
yt_subscription_id: Mapped[str | None] = mapped_column(String(128))
priority: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
hidden: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
# Per-user opt-in to full-history (deep) backfill of this channel. Default off so a
# new user's unique channels don't trigger a big shared-quota burst; the deep
# scheduler only picks up channels at least one user has requested.
deep_requested: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
subscribed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
user: Mapped["User"] = relationship(back_populates="subscriptions")
channel: Mapped["Channel"] = relationship(back_populates="subscriptions")
# Video.live_status values that mean "currently a live/upcoming broadcast" — transient states
# hidden from feeds/search and revisited by the refresh job until they settle into a VOD.
LIVE_OR_UPCOMING = ("live", "upcoming")
class Video(Base, TimestampMixin):
"""A YouTube video. Shared across users; per-user state lives elsewhere."""
__tablename__ = "videos"
id: Mapped[str] = mapped_column(String(16), primary_key=True) # video id
# No standalone channel_id index: the composite `ix_videos_channel_pub_cover`
# (channel_id, published_at DESC NULLS LAST, id) from migration 0095 leads with channel_id, so it
# serves every channel_id lookup (FK cascade, sync upsert, channel-detail) and the per-channel feed
# sort. Its INCLUDEd columns make the channel manager's sums and the feed sidebar's counts Index Only
# Scans (no heap): a column those queries start reading belongs in it (test_channel_list_index_only).
# Like the search_vector GIN, this composite is migration-defined, not an inline `index=True`.
channel_id: Mapped[str] = mapped_column(ForeignKey("channels.id", ondelete="CASCADE"))
title: Mapped[str | None] = mapped_column(Text) # normalized for display (see app.titles)
original_title: Mapped[str | None] = mapped_column(Text) # raw YouTube title, preserved
description: Mapped[str | None] = mapped_column(Text)
# NOT NULL (migration 0068): in reality every video carries a publish date, and a nullable key
# forced the feed's keyset predicate into a non-sargable `... OR published_at IS NULL` disjunction
# that blocked the ix_videos_published_at_desc range (deep pages scanned from the top and filtered —
# PERF-15). server_default now() is the safety net for the rare date-less ingest (the enrich fills
# the real date); the ix_videos_published_at index stays for the `oldest` sort.
published_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), index=True, nullable=False, server_default=func.now()
)
thumbnail_url: Mapped[str | None] = mapped_column(String(1024))
# The feed's views / duration / title sorts each ride a migration-defined index in their exact
# ORDER BY, one per direction (and per ICU collation for title): migration 0096, held to
# `feedquery.order_by` by test_feed_sort_indexes.
duration_seconds: Mapped[int | None] = mapped_column(Integer)
view_count: Mapped[int | None] = mapped_column(BigInteger)
like_count: Mapped[int | None] = mapped_column(BigInteger)
category_id: Mapped[int | None] = mapped_column(Integer)
topic_categories: Mapped[list | None] = mapped_column(JSON)
default_language: Mapped[str | None] = mapped_column(String(16))
detected_language: Mapped[str | None] = mapped_column(String(16))
# Creator-supplied keyword tags (snippet.tags joined by spaces) — fetched for free with the
# snippet we already request. Indexed into search_vector so the feed search matches the
# uploader's own keywords, not just words in the title.
keywords: Mapped[str | None] = mapped_column(Text)
# Accumulated distinct live-search queries that surfaced this video (across all users) —
# folds YouTube's own relevance judgement into the local search: a video YouTube returned
# for "ti amo magyarul" becomes findable by that query locally even if its title lacks it.
search_terms: Mapped[str | None] = mapped_column(Text)
# No btree here or on the flag columns below: migration 0069 (PERF-3) dropped the never-picked
# boolean btrees and replaced the useful ones with partial indexes over the target subset
# (via_search=true, NOT shorts_probed, unavailable_since IS NOT NULL) — migration-defined, like the
# search_vector GIN.
is_short: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
# Whether the youtube.com/shorts/<id> probe has run for this video.
shorts_probed: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
# True when this video first entered the catalog via a live YouTube search (not an
# organic channel/playlist sync). Used to hide search-discovered "noise" from the
# Library by default; left False by the organic ingest paths.
via_search: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
# True when this video entered the catalog because a user EXPLORED its (un-subscribed)
# channel. Like via_search it's hidden from the shared Library by default, but it's
# additionally PER-USER private: a via_explore video is only visible to users who
# explored or subscribe to its channel (a gate in the feed query), so one user's
# curiosity never floods everyone else's catalog. Left False by the organic ingest paths.
via_explore: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
# none | live | upcoming | premiere | was_live
live_status: Mapped[str] = mapped_column(
String(16), default="none", server_default="none", index=True
)
enriched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
# videos.list?part=status fields (populated during enrichment); used by the
# maintenance/validation job to decide whether a video is playable anywhere.
embeddable: Mapped[bool | None] = mapped_column(Boolean)
privacy_status: Mapped[str | None] = mapped_column(String(16)) # public|unlisted|private
upload_status: Mapped[str | None] = mapped_column(
String(16)
) # processed|failed|rejected|deleted
# Maintenance/validation lifecycle. last_checked_at drives the rolling re-validation
# (least-recently-checked first); unavailable_since marks a video as currently
# unplayable (hidden from the feed immediately, hard-deleted after a grace period);
# unavailable_reason records why (deleted|private|paywalled|abandoned|stuck_live).
# last_checked_at: no plain btree — migration 0069 (PERF-18) replaced it with the partial
# ix_videos_recheck_order (last_checked_at NULLS FIRST WHERE unavailable_since IS NULL) that the
# revalidate job's exact ORDER BY rides; unavailable_since is a partial WHERE IS NOT NULL (PERF-3).
last_checked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
unavailable_since: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
unavailable_reason: Mapped[str | None] = mapped_column(String(24))
# Weighted full-text search document (DB-generated, never written by the app): title (A) >
# creator keywords + search queries (B) > truncated description (C). ts_rank honours the
# weights so title matches outrank description matches. Backed by a GIN index (migration
# 0032). The accent-insensitive `unaccent_simple` config comes from migration 0031.
search_vector: Mapped[object | None] = mapped_column(
TSVECTOR,
Computed(
"setweight(to_tsvector('public.unaccent_simple', coalesce(title, '')), 'A') || "
"setweight(to_tsvector('public.unaccent_simple', coalesce(keywords, '') || ' ' || "
"coalesce(search_terms, '')), 'B') || "
"setweight(to_tsvector('public.unaccent_simple', left(coalesce(description, ''), 1000)), 'C')",
persisted=True,
),
)
channel: Mapped["Channel"] = relationship(back_populates="videos")
class SearchFind(Base, TimestampMixin):
"""Per-user record of a video the user surfaced via their own live YouTube search.
Lets "videos I searched for" appear in that user's own (Mine) feed and the Mine Source
filter — distinct from the GLOBAL `Video.via_search` provenance, which marks a video as
search-discovered by *anyone* and drives the shared Library's Source filter."""
__tablename__ = "search_finds"
__table_args__ = (UniqueConstraint("user_id", "video_id", name="uq_user_search_video"),)
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
video_id: Mapped[str] = mapped_column(ForeignKey("videos.id", ondelete="CASCADE"), index=True)
class ExploredChannel(Base, TimestampMixin):
"""Per-user record of a channel the user opened the page of WITHOUT subscribing.
Exploring an un-subscribed channel ingests its recent uploads (flagged via_explore) so the
user can browse them, but keeps them out of everyone else's catalog. This row (a) gates the
explorer's access to those ephemeral videos in the feed query and (b) drives the grace-period
purge: a from_explore channel with no live ExploredChannel row (and no subscription / no
per-user state on its videos) is hard-deleted by the cleanup job. Subscribing is the "keep"
signal — it promotes the channel to organic and these rows become irrelevant. `created_at`
(from TimestampMixin) is the grace clock; a re-visit refreshes it."""
__tablename__ = "explored_channels"
__table_args__ = (UniqueConstraint("user_id", "channel_id", name="uq_user_explored_channel"),)
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
channel_id: Mapped[str] = mapped_column(
ForeignKey("channels.id", ondelete="CASCADE"), index=True
)
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"))
channel_id: Mapped[str] = mapped_column(
ForeignKey("channels.id", ondelete="CASCADE"), index=True
)
class ApiQuotaUsage(Base):
"""Tracks YouTube Data API units spent per Pacific-time day (the quota resets at
midnight Pacific). The whole app shares one daily budget."""
+93 -2
View File
@@ -10,8 +10,10 @@ than joining Reader's tables itself.
from sqlalchemy import ColumnElement, and_, func, or_, select
from sqlalchemy.orm import InstrumentedAttribute, Session, aliased
from app.feedquery import feed_columns
from app.models import Channel, User, Video
from app.collation import collated
from app.feedquery import feed_columns, serialize
from app.models import User
from app.catalog.models import Channel, Subscription, Video
from app.reader.models import ChannelTag, Playlist, PlaylistItem, SavedView, Tag, VideoState
from app.userscope import owned_by, owned_row
@@ -268,3 +270,92 @@ def tags_in_feed_views(db: Session) -> set[int]:
for (filters,) in db.execute(select(SavedView.filters).where(SavedView.source == "feed"))
for t in view_tag_ids(filters)
}
def held_by(db: Session, user: User, video_ids: list[str]) -> set[str]:
"""Which of these videos the user already holds: watched, in progress or hidden, or in one of
their playlists (a live search sinks them below the new finds)."""
if not video_ids:
return set()
has_state = (
select(VideoState.id)
.where(VideoState.video_id == Video.id, owned_by(VideoState.user_id, user))
.exists()
)
in_playlist = (
select(PlaylistItem.id)
.join(Playlist, Playlist.id == PlaylistItem.playlist_id)
.where(PlaylistItem.video_id == Video.id, owned_by(Playlist.user_id, user))
.exists()
)
return set(
db.scalars(select(Video.id).where(Video.id.in_(video_ids), or_(has_state, in_playlist)))
)
def feed_rows_by_ids(db: Session, user: User, video_ids: list[str]) -> list[dict]:
"""These videos serialized with the feed's projection (their channel, this user's watch state,
saved), in the order given."""
if not video_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, owned_by(state.user_id, user)))
.where(Video.id.in_(video_ids))
).all()
by_id = {r.id: serialize(r) for r in rows}
return [by_id[i] for i in video_ids if i in by_id]
def playlist_channels(db: Session, user: User, exclude: str | None, lang: str) -> list:
"""The channels of the videos in the user's playlists that they don't subscribe to (`exclude`:
their own channel), each with how many of those videos it has and across how many playlists;
most present first, then by title in the reader's alphabet."""
subscribed = select(Subscription.channel_id).where(owned_by(Subscription.user_id, user))
q = (
select(
Channel,
func.count(func.distinct(PlaylistItem.video_id)),
func.count(func.distinct(Playlist.id)),
)
.select_from(PlaylistItem)
.join(Playlist, Playlist.id == PlaylistItem.playlist_id)
.join(Video, Video.id == PlaylistItem.video_id)
.join(Channel, Channel.id == Video.channel_id)
.where(owned_by(Playlist.user_id, user), Channel.id.not_in(subscribed))
)
if exclude:
q = q.where(Channel.id != exclude)
return list(
db.execute(
q.group_by(Channel.id).order_by(
func.count(func.distinct(PlaylistItem.video_id)).desc(),
collated(Channel.title, lang),
)
).all()
)
def user_tag_links(db: Session, user: User, channel_id: str | None = None) -> dict[str, list[int]]:
"""The user's own tag links by channel (one channel's, when given)."""
stmt = select(ChannelTag.channel_id, ChannelTag.tag_id).where(
owned_by(ChannelTag.user_id, user)
)
if channel_id is not None:
stmt = stmt.where(ChannelTag.channel_id == channel_id)
out: dict[str, list[int]] = {}
for cid, tag_id in db.execute(stmt).tuples():
out.setdefault(cid, []).append(tag_id)
return out
def user_tag_link(db: Session, user: User, channel_id: str, tag_id: int) -> ChannelTag | None:
return db.execute(
select(ChannelTag).where(
ChannelTag.channel_id == channel_id,
ChannelTag.tag_id == tag_id,
owned_by(ChannelTag.user_id, user),
)
).scalar_one_or_none()
+14
View File
@@ -464,3 +464,17 @@ def reorder_items(db: Session, pl: Playlist, order: list[str]) -> int:
touch(pl)
db.flush()
return pos
def attach_user_tag(db: Session, user: User, channel_id: str, tag_id: int) -> None:
"""The user's own label on a channel (idempotent). The caller has checked the tag is theirs."""
if queries.user_tag_link(db, user, channel_id, tag_id) is None:
db.add(ChannelTag(channel_id=channel_id, tag_id=tag_id, user_id=user.id, source="user"))
db.flush()
def detach_user_tag(db: Session, user: User, channel_id: str, tag_id: int) -> None:
row = queries.user_tag_link(db, user, channel_id, tag_id)
if row is not None:
db.delete(row)
db.flush()
+70 -258
View File
@@ -4,28 +4,22 @@ hidden, tags) and per-channel sync state. Local reads/writes plus YouTube-side w
from datetime import datetime, timezone
from functools import partial
from typing import TYPE_CHECKING
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from sqlalchemy import Select, func, select, update
from sqlalchemy.orm import Session
from app.collation import collated, sorted_words
from app.collation import sorted_words
from app.mailtext import lang_for_user
from app import logs, quota, scheduler
from app.auth import admin_user, current_user, require_write_scope
from app.db import get_db
from app.models import (
LIVE_OR_UPCOMING,
BlockedChannel,
Channel,
ExploredChannel,
Subscription,
User,
Video,
)
from app.reader.models import ChannelTag, Playlist, PlaylistItem, Tag
from app.routes._common import get_or_404, not_found, owned_or_404, youtube_refusal
from app.models import User
from app.catalog import queries, service
from app.reader import queries as reader_queries
from app.reader import service as reader_service
from app.routes._common import not_found, youtube_refusal
from app.sync.explore import explore_ingest_page
from app.sync.runner import backfill_channel_now, run_recent_backfill
from app.sync.subscriptions import apply_channel_details, details_stale
@@ -33,35 +27,17 @@ from app.utils import iso
from app.youtube.client import YouTubeClient, YouTubeError
from app.refusal import refused
if TYPE_CHECKING:
from app.catalog.models import Channel, Subscription
router = APIRouter(prefix="/api/channels", tags=["channels"])
log = logs.get_logger(__name__)
def channel_aggregates(channel_ids: list[str]) -> Select:
"""Per-channel stored count, last upload and total duration, in one grouped pass. Every column it
reads is in `ix_videos_channel_pub_cover` (migration 0095), so Postgres answers it from the index
alone; a test holds that (test_channel_list_index_only)."""
return (
select(
Video.channel_id,
func.count(Video.id),
func.max(Video.published_at),
func.coalesce(func.sum(Video.duration_seconds), 0),
)
.where(Video.channel_id.in_(channel_ids))
.group_by(Video.channel_id)
)
@router.get("")
def list_channels(user: User = Depends(current_user), db: Session = Depends(get_db)) -> list[dict]:
rows = db.execute(
select(Subscription, Channel)
.join(Channel, Channel.id == Subscription.channel_id)
.where(Subscription.user_id == user.id)
.order_by(Subscription.priority.desc(), collated(Channel.title, lang_for_user(user)))
).all()
rows = queries.subscriptions(db, user, lang_for_user(user))
# Per-channel aggregates over the shared catalog (channel_aggregates), computed on read rather than
# denormalised: an Index Only Scan over a ~20 MB index, where it used to read every video's heap page
@@ -70,36 +46,21 @@ def list_channels(user: User = Depends(current_user), db: Session = Depends(get_
# computed here — the manager table dropped its N/S/L column.)
channel_ids = [c.id for _, c in rows]
agg: dict[str, dict] = {}
if channel_ids:
for cid, n, last_up, total_dur in db.execute(channel_aggregates(channel_ids)).all():
agg[cid] = {
"stored": n,
"last_video_at": iso(last_up),
"total_duration_seconds": int(total_dur or 0),
}
for cid, n, last_up, total_dur in queries.aggregates(db, channel_ids):
agg[cid] = {
"stored": n,
"last_video_at": iso(last_up),
"total_duration_seconds": int(total_dur or 0),
}
# Channels already in the *global* deep-backfill queue — i.e. at least one user (any
# user) has requested full history and it isn't done yet. Their whole back-catalogue is
# coming for everyone (video data is shared), so the UI shouldn't offer "get full
# history" to a user who simply hasn't personally opted in.
deep_queued: set[str] = set()
if channel_ids:
for (cid,) in db.execute(
select(Subscription.channel_id)
.where(
Subscription.channel_id.in_(channel_ids),
Subscription.deep_requested.is_(True),
)
.distinct()
).all():
deep_queued.add(cid)
deep_queued = queries.deep_queued(db, channel_ids)
# The user's personal tag links, grouped by channel.
tags_by_channel: dict[str, list[int]] = {}
for cid, tag_id in db.execute(
select(ChannelTag.channel_id, ChannelTag.tag_id).where(ChannelTag.user_id == user.id)
).all():
tags_by_channel.setdefault(cid, []).append(tag_id)
tags_by_channel = reader_queries.user_tag_links(db, user)
return [
{
@@ -128,27 +89,7 @@ DISCOVERY_ENRICH_CAP = 200
def _discovery_rows(db: Session, user: User) -> list:
"""The user's playlist channels they don't subscribe to (and that aren't their own),
each with how many playlist videos come from it and across how many playlists."""
subscribed = select(Subscription.channel_id).where(Subscription.user_id == user.id)
q = (
select(
Channel,
func.count(func.distinct(PlaylistItem.video_id)),
func.count(func.distinct(Playlist.id)),
)
.select_from(PlaylistItem)
.join(Playlist, Playlist.id == PlaylistItem.playlist_id)
.join(Video, Video.id == PlaylistItem.video_id)
.join(Channel, Channel.id == Video.channel_id)
.where(Playlist.user_id == user.id, Channel.id.not_in(subscribed))
)
if user.yt_channel_id:
q = q.where(Channel.id != user.yt_channel_id)
return db.execute(
q.group_by(Channel.id).order_by(
func.count(func.distinct(PlaylistItem.video_id)).desc(),
collated(Channel.title, lang_for_user(user)),
)
).all()
return reader_queries.playlist_channels(db, user, user.yt_channel_id, lang_for_user(user))
@router.get("/discovery")
@@ -170,7 +111,7 @@ def discover_channels(
):
own = yt.get_my_channel_id()
if own:
user.yt_channel_id = own
service.remember_own_channel(user, own)
db.commit()
except YouTubeError as exc:
# No traceback anywhere in this file's YouTube handlers: for an upstream answer
@@ -226,7 +167,7 @@ def discover_channels(
]
def _channel_summary(ch: Channel) -> dict:
def _channel_summary(ch: "Channel") -> dict:
"""The channel fields common to the list and discovery projections."""
return {
"id": ch.id,
@@ -240,7 +181,7 @@ def _channel_summary(ch: Channel) -> dict:
def _channel_detail_dict(
channel: Channel,
channel: "Channel",
*,
subscribed: bool,
explored: bool,
@@ -309,7 +250,7 @@ def channel_detail(
needed (a stub) and refreshes it once it is stale (`DETAILS_MAX_AGE`), which is how a channel nobody subscribes to (an explored one, outside the
subscription import) gets a banner that still resolves. The enrich uses the API key (1 unit,
cached after)."""
channel = get_or_404(db, Channel, channel_id, what="channel")
channel = _channel_or_404(db, channel_id)
# No `published_at is None` sentinel: a terminated channel never gets one, so it re-fetched on
# every view; rows enriched before that field existed are long past DETAILS_MAX_AGE anyway.
stale = details_stale(channel, datetime.now(timezone.utc))
@@ -321,7 +262,7 @@ def channel_detail(
):
apply_channel_details(db, yt.get_channels([channel_id]), [channel_id])
db.commit()
channel = db.get(Channel, channel_id)
channel = _channel_or_404(db, channel_id)
except YouTubeError as exc:
log.warning(
"channel detail enrich failed: %s",
@@ -331,41 +272,15 @@ def channel_detail(
),
)
db.rollback()
sub = db.execute(
select(Subscription).where(
Subscription.user_id == user.id, Subscription.channel_id == channel_id
)
).scalar_one_or_none()
sub = queries.subscription(db, user, channel_id)
subscribed = sub is not None
explored = (
db.execute(
select(ExploredChannel.id).where(
ExploredChannel.user_id == user.id, ExploredChannel.channel_id == channel_id
)
).first()
is not None
)
blocked = _is_blocked(db, user, channel_id)
explored = queries.exploration(db, user, channel_id) is not None
blocked = queries.is_blocked(db, user, channel_id)
# Stored total + the Normal/Shorts/Live split in one grouped pass (same filters as the manager
# list), so the channel page can show the per-type breakdown that moved off the table.
n_all, n_short, n_live = db.execute(
select(
func.count(Video.id),
func.count(Video.id).filter(Video.is_short.is_(True)),
func.count(Video.id).filter(Video.live_status.in_(LIVE_OR_UPCOMING)),
).where(Video.channel_id == channel_id)
).one()
stored = int(n_all or 0)
count_short = int(n_short or 0)
count_live = int(n_live or 0)
stored, count_short, count_live = queries.video_counts(db, channel_id)
count_normal = max(0, stored - count_short - count_live)
tag_ids = list(
db.scalars(
select(ChannelTag.tag_id).where(
ChannelTag.user_id == user.id, ChannelTag.channel_id == channel_id
)
).all()
)
tag_ids = reader_queries.user_tag_links(db, user, channel_id).get(channel_id, [])
return _channel_detail_dict(
channel,
subscribed=subscribed,
@@ -395,28 +310,15 @@ def explore_channel(
it, then ingests one page of its uploads (newest-first) marked via_explore + enriched.
Returns the next page token so the page can "load more". Guarded against the shared
backfill quota reserve."""
channel = get_or_404(db, Channel, channel_id, what="channel")
if _is_blocked(db, user, channel_id):
channel = _channel_or_404(db, channel_id)
if queries.is_blocked(db, user, channel_id):
raise refused(403, "channel_blocked")
if quota.backfill_reserve_reached():
raise refused(429, "quota_reserve")
# Record/refresh the exploration (DB only) before ingesting, so the explorer can see the
# videos as soon as they land. Flag the channel ephemeral unless it's organically followed.
existing = db.execute(
select(ExploredChannel).where(
ExploredChannel.user_id == user.id, ExploredChannel.channel_id == channel_id
)
).scalar_one_or_none()
if existing is None:
db.add(ExploredChannel(user_id=user.id, channel_id=channel_id))
else:
existing.created_at = datetime.now(timezone.utc) # refresh the grace clock
followed = db.execute(
select(Subscription.id).where(Subscription.channel_id == channel_id).limit(1)
).first()
if followed is None and not channel.from_explore:
channel.from_explore = True
service.explore(db, user, channel)
db.commit()
try:
@@ -430,7 +332,7 @@ def explore_channel(
if details_stale(channel, datetime.now(timezone.utc)):
apply_channel_details(db, yt.get_channels([channel_id]), [channel_id])
db.commit()
channel = db.get(Channel, channel_id)
channel = _channel_or_404(db, channel_id)
result = explore_ingest_page(db, yt, channel, page_token)
except YouTubeError as exc:
raise youtube_refusal(exc) from exc
@@ -442,30 +344,20 @@ def explore_channel(
}
def _user_subscription(db: Session, user: User, channel_id: str) -> Subscription:
sub = db.execute(
select(Subscription).where(
Subscription.user_id == user.id, Subscription.channel_id == channel_id
)
).scalar_one_or_none()
def _channel_or_404(db: Session, channel_id: str) -> "Channel":
channel = queries.channel(db, channel_id)
if channel is None:
raise not_found("channel")
return channel
def _user_subscription(db: Session, user: User, channel_id: str) -> "Subscription":
sub = queries.subscription(db, user, channel_id)
if sub is None:
raise not_found("subscription")
return sub
def _is_blocked(db: Session, user: User, channel_id: str) -> bool:
"""Whether this user has blocked this channel."""
return (
db.execute(
select(BlockedChannel.id).where(
BlockedChannel.user_id == user.id,
BlockedChannel.channel_id == channel_id,
)
).first()
is not None
)
class ChannelUpdateIn(BaseModel):
# PATCH semantics: a field left unset (None) is untouched. Every field is optional so the
# UI can send just the one control it changed. int/bool coercion + rejection of garbage
@@ -483,19 +375,14 @@ def update_channel(
db: Session = Depends(get_db),
) -> dict:
sub = _user_subscription(db, user, channel_id)
channel = db.get(Channel, channel_id)
deep_turned_on = False
if body.priority is not None:
sub.priority = body.priority
if body.hidden is not None:
sub.hidden = body.hidden
if body.deep_requested is not None:
# Opt this channel into (or out of) full-history backfill. The deep scheduler
# picks the flag up on its next run; turning it off won't undo already-fetched
# videos, it just stops further deep paging if nobody else still wants it.
want = body.deep_requested
deep_turned_on = want and not sub.deep_requested
sub.deep_requested = want
channel = queries.channel(db, channel_id)
deep_turned_on = service.update_subscription(
db,
sub,
priority=body.priority,
hidden=body.hidden,
deep_requested=body.deep_requested,
)
db.commit()
# Opting a channel into full history should give an immediate feed: if we haven't even
@@ -539,19 +426,8 @@ def reset_backfill(
state (a "reset" trigger). Clears the channel's backfill markers, opts it back into deep
backfill, and re-runs its recent pull immediately; the deep scheduler re-pages the full
back-catalog on its next run. Idempotent — videos upsert by id, so nothing duplicates."""
channel = get_or_404(db, Channel, channel_id, what="channel")
channel.backfill_done = False
channel.backfill_cursor = None
channel.recent_synced_at = None
# Re-arm the demand-driven deep scheduler for this channel. run_deep_backfill gates on an
# EXISTS over subscriptions with deep_requested=True, so flip it on EVERY subscriber rather
# than requiring the admin to be personally subscribed (they reset from the manager, not
# their own feed). No-op row count if the channel has no subscribers.
db.execute(
update(Subscription)
.where(Subscription.channel_id == channel_id)
.values(deep_requested=True)
)
channel = _channel_or_404(db, channel_id)
service.reset_backfill(db, channel)
db.commit()
with quota.attribute(user.id, quota.QuotaAction.VIDEOS_BACKFILL_RECENT):
run_recent_backfill(db, [channel], max_channels=1)
@@ -569,13 +445,8 @@ def subscribe(
local subscription with YouTube's returned resource id and enriches the channel's metadata
if it's still a stub, so the new row renders properly in the manager. It does NOT pull the
channel's videos — the scheduler picks those up on its next run, like any subscribed channel."""
channel = get_or_404(db, Channel, channel_id, what="channel")
existing = db.execute(
select(Subscription).where(
Subscription.user_id == user.id, Subscription.channel_id == channel_id
)
).scalar_one_or_none()
if existing is not None:
channel = _channel_or_404(db, channel_id)
if queries.subscription(db, user, channel_id) is not None:
raise refused(409, "already_subscribed")
with (
quota.attribute(user.id, quota.QuotaAction.CHANNELS_SUBSCRIBE),
@@ -615,24 +486,7 @@ def subscribe(
reason=exc.reason,
),
)
db.add(
Subscription(
user_id=user.id,
channel_id=channel_id,
yt_subscription_id=yt_sub_id or None,
subscribed_at=datetime.now(timezone.utc),
)
)
# Subscribing is the "keep" signal: a previously-explored channel becomes a normal organic
# one. Drop its ephemeral flags so the cleanup job leaves it alone and its videos stop being
# per-user private (they join the shared Library like any subscribed channel's uploads).
if channel.from_explore:
channel.from_explore = False
db.execute(
update(Video)
.where(Video.channel_id == channel_id, Video.via_explore.is_(True))
.values(via_explore=False)
)
service.record_subscription(db, user, channel, yt_sub_id)
db.commit()
return {"ok": True, "subscribed": channel_id}
@@ -659,7 +513,7 @@ def unsubscribe(
# 422 (a real, explainable error → speaking modal), not 502 which the client treats as
# a transient "connection lost" gateway blip.
raise youtube_refusal(exc) from exc
db.delete(sub)
service.drop_subscription(db, sub)
db.commit()
return {"ok": True, "unsubscribed": channel_id}
@@ -667,16 +521,7 @@ def unsubscribe(
@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(collated(Channel.title, lang_for_user(user)))
)
.scalars()
.all()
)
rows = queries.blocked_channels(db, user, lang_for_user(user))
return [
{"id": c.id, "title": c.title, "handle": c.handle, "thumbnail_url": c.thumbnail_url}
for c in rows
@@ -692,16 +537,9 @@ def block_channel(
"""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."""
get_or_404(db, Channel, channel_id, what="channel")
if not _is_blocked(db, user, channel_id):
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()
_channel_or_404(db, channel_id)
service.block(db, user, channel_id)
db.commit()
return {"ok": True, "blocked": channel_id}
@@ -711,11 +549,7 @@ def unblock_channel(
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)
)
)
service.unblock(db, user, channel_id)
db.commit()
return {"ok": True, "unblocked": channel_id}
@@ -732,26 +566,12 @@ def attach_tag(
db: Session = Depends(get_db),
) -> dict:
_user_subscription(db, user, channel_id)
# Only the user's own tags may be attached as personal links (owned_or_404 → 404 if the tag
# is missing OR belongs to someone else, without leaking which).
owned_or_404(db, Tag, body.tag_id, user, what="tag")
exists = db.execute(
select(ChannelTag).where(
ChannelTag.channel_id == channel_id,
ChannelTag.tag_id == body.tag_id,
ChannelTag.user_id == user.id,
)
).scalar_one_or_none()
if exists is None:
db.add(
ChannelTag(
channel_id=channel_id,
tag_id=body.tag_id,
user_id=user.id,
source="user",
)
)
db.commit()
# Only the user's own tags may be attached as personal links (a 404 if the tag is missing OR
# belongs to someone else, without leaking which).
if reader_queries.own_tag(db, user, body.tag_id) is None:
raise not_found("tag")
reader_service.attach_user_tag(db, user, channel_id, body.tag_id)
db.commit()
return {"ok": True, "channel_id": channel_id, "tag_id": body.tag_id}
@@ -762,14 +582,6 @@ def detach_tag(
user: User = Depends(current_user),
db: Session = Depends(get_db),
) -> dict:
row = db.execute(
select(ChannelTag).where(
ChannelTag.channel_id == channel_id,
ChannelTag.tag_id == tag_id,
ChannelTag.user_id == user.id,
)
).scalar_one_or_none()
if row is not None:
db.delete(row)
db.commit()
reader_service.detach_user_tag(db, user, channel_id, tag_id)
db.commit()
return {"ok": True, "channel_id": channel_id, "tag_id": tag_id}
+3 -5
View File
@@ -18,7 +18,8 @@ from sqlalchemy.orm import Session
from app import scheduler
from app.auth import admin_user
from app.db import get_db
from app.models import Channel, JobRun, LogEvent, User
from app.models import JobRun, LogEvent, User
from app.catalog import queries as catalog_queries
from app.routes import _account as account
from app.routes._keyset import PAGE_MAX, accounts, older_than, page, utc
from app.runstatus import RunStatus
@@ -116,10 +117,7 @@ def list_runs(
lambda r: r.started_at,
)
channels = {r.target for r in runs if r.job in scheduler.AD_HOC_JOBS and r.target}
titles: dict[str, str] = {}
if channels:
rows = db.execute(select(Channel.id, Channel.title).where(Channel.id.in_(channels)))
titles = {cid: title for cid, title in rows.tuples() if title is not None}
titles = catalog_queries.channel_titles(db, channels)
lines: dict[int, int] = {}
if runs:
counted = db.execute(
+10 -55
View File
@@ -4,7 +4,6 @@ and the shared quota picture."""
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from sqlalchemy import and_, func, or_, select
from sqlalchemy.orm import Session
from app import audit, quota, state, sysconfig
@@ -12,7 +11,8 @@ from app.audit import AuditAction
from app.auth import admin_user
from app.config import settings
from app.db import get_db
from app.models import Channel, Subscription, User, Video
from app.models import User
from app.catalog import queries as catalog_queries
from app.scheduler import (
JOB_INTERVALS,
MAX_INTERVAL,
@@ -23,7 +23,6 @@ from app.scheduler import (
trigger_all,
trigger_job,
)
from app.sync.explore import followed_channel
from app.sync.runner import estimate_deep_backfill
from app.routes._common import not_found
from app.refusal import refused
@@ -122,58 +121,14 @@ def get_scheduler(_: User = Depends(admin_user), db: Session = Depends(get_db))
snap = scheduler_snapshot(db)
# Work still queued (shared catalog, so these are instance-wide, not per-user).
deep_requested = (
select(Subscription.id)
.where(
Subscription.channel_id == Channel.id,
Subscription.deep_requested.is_(True),
)
.exists()
)
deep_channels = (
db.execute(select(Channel).where(Channel.backfill_done.is_(False), deep_requested))
.scalars()
.all()
)
deep_eta = estimate_deep_backfill(db, deep_channels)
queue = {
"channels_recent_pending": db.scalar(
# The same selection run_recent_backfill works through, or the card counts channels
# the job will never touch and never reaches zero.
select(func.count())
.select_from(Channel)
.where(Channel.recent_synced_at.is_(None), followed_channel())
)
or 0,
# The whole opted-in BACKLOG, deliberately not `run_deep_backfill`'s next-pass selection
# (which also wants `recent_synced_at`): the same figure /api/sync/my-status shows the user,
# and a channel missing its recent pull is one the same job tick fetches first, not one the
# job will never touch. Narrowing it here would under-report work that is genuinely queued.
"channels_deep_pending": deep_eta["channels_pending"],
"deep_eta_seconds": deep_eta["eta_seconds"],
"videos_pending_enrich": db.scalar(
select(func.count()).select_from(Video).where(Video.enriched_at.is_(None))
)
or 0,
"videos_pending_shorts": db.scalar(
select(func.count())
.select_from(Video)
.where(Video.shorts_probed.is_(False), Video.enriched_at.is_not(None))
)
or 0,
"videos_live_refresh": db.scalar(
select(func.count())
.select_from(Video)
.where(
or_(
Video.live_status.in_(("live", "upcoming")),
and_(Video.live_status == "was_live", Video.duration_seconds.is_(None)),
)
)
)
or 0,
}
deep_eta = estimate_deep_backfill(db, catalog_queries.deep_backlog(db))
queue = catalog_queries.work_queue(db)
# The whole opted-in BACKLOG, deliberately not `run_deep_backfill`'s next-pass selection
# (which also wants `recent_synced_at`): the same figure /api/sync/my-status shows the user,
# and a channel missing its recent pull is one the same job tick fetches first, not one the
# job will never touch. Narrowing it here would under-report work that is genuinely queued.
queue["channels_deep_pending"] = deep_eta["channels_pending"]
queue["deep_eta_seconds"] = deep_eta["eta_seconds"]
used = quota.units_used_today(db)
quota_info = {
+14 -243
View File
@@ -11,37 +11,24 @@ 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)."""
dropped from the search response up front).
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
This is the HTTP side — the paging, the caps and the refusals; what a page of results leaves in the
catalog is `catalog/search`."""
from fastapi import APIRouter, Depends
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 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 (
LIVE_OR_UPCOMING,
BlockedChannel,
Channel,
SearchFind,
Subscription,
User,
Video,
)
from app.reader.models import Playlist, PlaylistItem, VideoState
from app.feedquery import feed_columns, serialize
from app.utils import like_escape
from app.sync.subscriptions import apply_channel_details
from app.sync.videos import insert_stubs, apply_video_details, parse_dt
from app.models import User
from app.reader import queries as reader_queries
from app.youtube import search_scrape
from app.youtube.client import YouTubeClient, YouTubeError
from app.youtube.shorts import make_client, probe_is_short
from app.refusal import refused
from app.routes._common import youtube_refusal
@@ -52,198 +39,6 @@ router = APIRouter(prefix="/api/search", tags=["search"])
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, new_channel_ids)
db.commit()
except YouTubeError:
# The stubs remain and the details can be backfilled later — but until they are, those
# channels sit in the results with no thumbnail and no subscriber count, looking like
# channels that have neither.
log.warning(
"the channels found by a search could not be enriched",
extra=logs.event("search.channel_enrich_failed", channels=len(new_channel_ids)),
)
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:
# Every candidate then stays unenriched, and an unenriched video cannot be judged live or
# Short — so it is dropped. A working search returns nothing and reads as "no results".
log.warning(
"the videos found by a search could not be enriched",
extra=logs.event("search.video_enrich_failed", candidates=len(ids)),
)
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]
def _fold_term(db: Session, video_ids: list[str], term: str) -> None:
"""Append `term` to each video's space-joined `search_terms`, unless it is already one of them.
"Already one of them" is a whole-term match, as a literal: a substring test skipped `cat` on a
video already found for `category`, and an unescaped one read the `_` in `my_channel` as a
wildcard — either way the term was never folded in, and local search lost that relevance."""
padded = func.concat(" ", func.coalesce(Video.search_terms, ""), " ")
db.execute(
update(Video)
.where(Video.id.in_(video_ids))
.where(padded.notilike(f"% {like_escape(term)} %"))
.values(
search_terms=func.trim(
func.coalesce(Video.search_terms, "").op("||")(" ").op("||")(term)
)
)
)
@router.get("/youtube")
def search_youtube(
q: str | None = None,
@@ -285,9 +80,7 @@ def search_youtube(
raise refused(429, "quota_spent")
# 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()
)
blocked = queries.blocked_ids(db, user)
collected: list[str] = []
next_cursor = cursor
@@ -328,7 +121,7 @@ def search_youtube(
# 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))
collected.extend(catalog_search.ingest_candidates(db, yt, user, candidates))
except Exception as exc: # any ingest failure, not just YouTubeError
db.rollback()
# The term is NOT in this line. A search string is exactly what the privacy
@@ -376,17 +169,12 @@ def search_youtube(
# 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"])
)
_fold_term(db, ordered, term)
catalog_search.record_finds(db, user, ordered, term)
db.commit()
ordered = _new_first(db, user, ordered)
ordered = catalog_search.new_first(db, user, ordered)
return {
"items": _serialize_ids(db, user, ordered),
"items": reader_queries.feed_rows_by_ids(db, user, ordered),
"next_cursor": next_cursor,
"has_more": bool(next_cursor),
"source": source,
@@ -410,23 +198,6 @@ def clear_search(
ids = body.video_ids
if not ids:
return {"ok": True, "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
removed = catalog_search.clear(db, user, ids)
db.commit()
return {"ok": True, "removed": removed}
+11 -54
View File
@@ -1,12 +1,12 @@
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import and_, case, func, select, update
from sqlalchemy.orm import Session
from app import audit, quota, state
from app.audit import AuditAction
from app.auth import admin_user, current_user, has_read_scope
from app.db import get_db
from app.models import Channel, Subscription, User, Video
from app.models import User
from app.catalog import queries, service
from app.routes._common import youtube_refusal
from app.ratelimit import RateLimiter
from app.sync.runner import estimate_deep_backfill
@@ -47,19 +47,8 @@ def sync_subscriptions(user: User = Depends(current_user), db: Session = Depends
@router.get("/status")
def sync_status(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
sub_count = db.scalar(
select(func.count()).select_from(Subscription).where(Subscription.user_id == user.id)
)
return {
"subscriptions": sub_count,
"channels_total": db.scalar(select(func.count()).select_from(Channel)),
"channels_backfilling": db.scalar(
select(func.count()).select_from(Channel).where(Channel.backfill_done.is_(False))
),
"videos_total": db.scalar(select(func.count()).select_from(Video)),
"pending_enrich": db.scalar(
select(func.count()).select_from(Video).where(Video.enriched_at.is_(None))
),
**queries.catalog_totals(db, user),
"quota_used_today": quota.units_used_today(db),
"quota_remaining_today": quota.remaining_today(db),
"paused": state.is_sync_paused(db),
@@ -70,51 +59,21 @@ def sync_status(user: User = Depends(current_user), db: Session = Depends(get_db
@router.get("/my-status")
def my_status(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
"""Per-user sync progress: how far the channels *this* user subscribes to have synced."""
sub_join = and_(Subscription.channel_id == Channel.id, Subscription.user_id == user.id)
total, details_synced, recent_synced, all_deep_done = db.execute(
select(
func.count(Channel.id),
func.count(Channel.details_synced_at),
func.count(Channel.recent_synced_at),
func.sum(case((Channel.backfill_done.is_(True), 1), else_=0)),
)
.select_from(Channel)
.join(Subscription, sub_join)
).one()
total = total or 0
all_deep_done = int(all_deep_done or 0)
total, details_synced, recent_synced, all_deep_done = queries.my_progress(db, user)
# Per-user full-history (deep) backfill: channels this user has opted in, and an ETA
# for the still-pending ones (quota-bound estimate).
deep_requested_channels = (
db.execute(
select(Channel).join(
Subscription,
and_(sub_join, Subscription.deep_requested.is_(True)),
)
)
.scalars()
.all()
)
deep_requested_channels = queries.my_deep_requested(db, user)
deep_eta = estimate_deep_backfill(db, deep_requested_channels)
# "done / requested" is shown as one ratio, so both halves count the SAME channels: the ones this
# user asked full history for. Counting every subscribed channel whose back-catalog happens to be
# complete (another user asked, or it was short) gave "342/329".
deep_done = sum(1 for c in deep_requested_channels if c.backfill_done)
my_videos = db.scalar(
select(func.count(Video.id)).join(
Subscription,
and_(
Subscription.channel_id == Video.channel_id,
Subscription.user_id == user.id,
Subscription.hidden.is_(False),
),
)
)
my_videos = queries.my_video_count(db, user)
# Whole-catalog total (every video any user's subscriptions have pulled in), so the
# header can show "yours" against the shared library size.
total_videos = db.scalar(select(func.count(Video.id)))
total_videos = queries.video_total(db)
return {
"channels_total": total,
"channels_details_synced": details_synced,
@@ -125,8 +84,8 @@ def my_status(user: User = Depends(current_user), db: Session = Depends(get_db))
"channels_deep_requested": len(deep_requested_channels),
"deep_pending_count": deep_eta["channels_pending"],
"deep_eta_seconds": deep_eta["eta_seconds"],
"my_videos": my_videos or 0,
"total_videos": total_videos or 0,
"my_videos": my_videos,
"total_videos": total_videos,
"quota_used_today": quota.units_used_today(db),
"quota_remaining_today": quota.remaining_today(db),
"paused": state.is_sync_paused(db),
@@ -143,11 +102,9 @@ def deep_all(
on: bool = True,
) -> dict:
"""Opt every one of the user's channels into (or out of) full-history backfill."""
result = db.execute(
update(Subscription).where(Subscription.user_id == user.id).values(deep_requested=on)
)
updated = service.set_deep_all(db, user, on)
db.commit()
return {"ok": True, "updated": result.rowcount, "deep_requested": on}
return {"ok": True, "updated": updated, "deep_requested": on}
@router.post("/pause")
+3 -3
View File
@@ -9,13 +9,13 @@ from typing import Callable
from apscheduler.executors.pool import ThreadPoolExecutor
from apscheduler.jobstores.base import ConflictingIdError
from apscheduler.schedulers.background import BackgroundScheduler
from sqlalchemy import select
from app import audit, debugsession, jobruns, logs, logsink, progress, quota, sysconfig
from app.config import settings
from app.db import SessionLocal
from app.downloads.gc import run_download_gc
from app.models import Channel, JobRun
from app.models import JobRun
from app.catalog import queries as catalog_queries
from app.plex.sync import sync as run_plex_sync
from app.plex.tmdb import run_enrich as run_tmdb_enrich
from app.plex.tmdb import run_people_enrich
@@ -305,7 +305,7 @@ def scheduler_snapshot(db) -> dict:
"running": going is not None,
"target": shown.target if shown else None,
"target_title": (
db.scalar(select(Channel.title).where(Channel.id == shown.target))
catalog_queries.channel_title(db, shown.target)
if shown is not None and shown.target
else None
),
+1 -1
View File
@@ -12,7 +12,7 @@ from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app import logs, progress, sysconfig
from app.models import Channel, Video
from app.catalog.models import Channel, Video
from app.reader import queries as reader_queries
from app.reader import service as reader_service
+1 -1
View File
@@ -14,7 +14,7 @@ from sqlalchemy import delete, or_, select
from sqlalchemy.orm import Session
from app import logs, sysconfig
from app.models import Channel, ExploredChannel, SearchFind, Subscription, Video
from app.catalog.models import Channel, ExploredChannel, SearchFind, Subscription, Video
from app.reader import queries as reader_queries
from app.sync.videos import insert_stubs, stub_from_playlist_item, apply_video_details
from app.youtube.client import YouTubeClient, YouTubeError, is_playlist_gone
+1 -1
View File
@@ -27,7 +27,7 @@ from sqlalchemy.orm import Session
from app import logs, progress, quota, sysconfig
from app.config import settings
from app.models import Video
from app.catalog.models import Video
from app.reader import queries as reader_queries
from app.notifications import create_notification
from app.sync.runner import get_service_user
+2 -1
View File
@@ -20,7 +20,8 @@ from sqlalchemy.pool import NullPool
from app import logs, progress, quota
from app.auth import has_read_scope
from app.models import User, Video
from app.models import User
from app.catalog.models import Video
from app.reader.models import Playlist, PlaylistItem
from app.sync.playlist_merge import merge_plan
from app.sync.runner import token_users
+2 -1
View File
@@ -16,7 +16,8 @@ from sqlalchemy.orm import Session
from app import logs, progress, quota, sysconfig
from app.db import SessionLocal
from app.models import Channel, OAuthToken, Subscription, User
from app.models import OAuthToken, User
from app.catalog.models import Channel, Subscription
from app.sync.explore import followed_channel
from app.sync.subscriptions import import_subscriptions
from app.sync.videos import (
+7 -3
View File
@@ -7,9 +7,11 @@ from sqlalchemy import or_, select
from sqlalchemy.orm import Session
from app import logs
from app.models import Channel, Subscription, User
from app.models import User
from app.catalog.models import Channel, Subscription
from app.utils import normalize_description
from app.youtube.client import YouTubeClient, YouTubeError, best_thumbnail
from app.userscope import owned_by
log = logs.get_logger(__name__)
@@ -172,7 +174,7 @@ def import_subscriptions(db: Session, user: User) -> dict:
new_channels += 1
sub_row = db.execute(
select(Subscription).where(
Subscription.user_id == user.id, Subscription.channel_id == cid
owned_by(Subscription.user_id, user.id), Subscription.channel_id == cid
)
).scalar_one_or_none()
if sub_row is None:
@@ -185,7 +187,9 @@ def import_subscriptions(db: Session, user: User) -> dict:
# (see `should_prune`).
removed = 0
existing = (
db.execute(select(Subscription).where(Subscription.user_id == user.id)).scalars().all()
db.execute(select(Subscription).where(owned_by(Subscription.user_id, user.id)))
.scalars()
.all()
)
prune_skipped = not should_prune(len(fetched), len(existing))
if prune_skipped:
+2 -1
View File
@@ -23,7 +23,8 @@ from collections import Counter
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models import Subscription, User
from app.models import User
from app.catalog.models import Subscription
from app.reader.models import ChannelTag, Tag
from app.userscope import owned_by
+1 -1
View File
@@ -12,7 +12,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session
from app import logs, progress, quota, sysconfig
from app.models import LIVE_OR_UPCOMING, Channel, Video
from app.catalog.models import LIVE_OR_UPCOMING, Channel, Video
from app.titles import normalize_title
from app.youtube.client import YouTubeClient, YouTubeError, best_thumbnail, is_playlist_gone
from app.youtube.shorts import make_client, probe_is_short
+1
View File
@@ -5,6 +5,7 @@ names the whole schema. Whatever needs all of it — the migrations, the test da
every mapper — imports this module.
"""
import app.catalog.models # Catalog's tables
import app.downloads.models # Capture's tables
import app.models # the core tables
import app.reader.models # noqa: F401 — Reader's tables
+2 -1
View File
@@ -9,7 +9,8 @@ from datetime import datetime, timedelta, timezone
from app.auth import admin_user
from app.main import app
from app.models import AuditLog, Channel, JobRun, LogEvent, QuotaEvent
from app.models import AuditLog, JobRun, LogEvent, QuotaEvent
from app.catalog.models import Channel
from tests.test_body_limit import _route_objects
T0 = datetime(2026, 10, 1, 12, 0, tzinfo=timezone.utc)
+1 -1
View File
@@ -7,7 +7,7 @@ drops an unknown id, so the view silently showed the whole feed under its old na
from sqlalchemy import select
from app.models import Channel
from app.catalog.models import Channel
from app.reader.models import ChannelTag, SavedView, Tag
from app.reader import service as reader_service
+74
View File
@@ -0,0 +1,74 @@
"""The channel manager's local writes, through HTTP: each one lands and is committed.
The service tests (`test_catalog_service.py`) hold what each write does; these hold that the route
calls it and commits — the `db` fixture fails a test whose handler left a write uncommitted — and
what a client sees back. None of these reach YouTube. Needs the DB.
"""
from sqlalchemy import select
from app.catalog.models import BlockedChannel, Channel, Subscription
from app.reader.models import ChannelTag, Tag
def _seed(db, user):
db.add(Channel(id="UCr", title="R"))
db.flush()
db.add(Subscription(user_id=user.id, channel_id="UCr"))
tag = Tag(user_id=user.id, name="Mine")
db.add(tag)
db.commit()
return tag
def test_a_subscriptions_settings_change_and_list_back(db, make_user, as_user):
user = make_user()
_seed(db, user)
client = as_user(user)
r = client.patch("/api/channels/UCr", json={"priority": 2, "hidden": True})
assert r.status_code == 200, r.text
assert (r.json()["priority"], r.json()["hidden"]) == (2, True)
listed = client.get("/api/channels").json()
assert [(c["id"], c["priority"], c["hidden"]) for c in listed] == [("UCr", 2, True)]
r = client.post("/api/sync/deep-all?on=false")
assert r.json() == {"ok": True, "updated": 1, "deep_requested": False}
def test_a_block_and_an_unblock(db, make_user, as_user):
user = make_user()
_seed(db, user)
client = as_user(user)
assert client.post("/api/channels/UCr/block").json() == {"ok": True, "blocked": "UCr"}
assert client.post("/api/channels/UCr/block").status_code == 200 # idempotent
assert [c["id"] for c in client.get("/api/channels/blocked/list").json()] == ["UCr"]
assert client.get("/api/channels/UCr").json()["blocked"] is True
assert client.delete("/api/channels/UCr/block").json() == {"ok": True, "unblocked": "UCr"}
db.expire_all()
assert db.execute(select(BlockedChannel)).first() is None
def test_a_tag_attached_to_a_channel_and_detached(db, make_user, as_user):
user, stranger = make_user(), make_user()
tag = _seed(db, user)
client = as_user(user)
r = client.post("/api/channels/UCr/tags", json={"tag_id": tag.id})
assert r.status_code == 200, r.text
assert client.post("/api/channels/UCr/tags", json={"tag_id": tag.id}).status_code == 200
assert client.get("/api/channels/UCr").json()["tag_ids"] == [tag.id]
db.add(Subscription(user_id=stranger.id, channel_id="UCr"))
db.commit()
# Another user's tag is not theirs to attach: the same 404 as a missing one.
r = as_user(stranger).post("/api/channels/UCr/tags", json={"tag_id": tag.id})
assert r.json()["detail"] == {"code": "not_found", "reason": "tag"}
assert client.delete(f"/api/channels/UCr/tags/{tag.id}").status_code == 200
db.expire_all()
assert db.execute(select(ChannelTag)).first() is None
def test_the_sync_status_counts_the_callers_subscriptions(db, make_user, as_user):
user = make_user()
_seed(db, user)
body = as_user(user).get("/api/sync/status").json()
assert (body["subscriptions"], body["channels_total"]) == (1, 1)
mine = as_user(user).get("/api/sync/my-status").json()
assert (mine["channels_total"], mine["channels_recent_pending"]) == (1, 1)
+295
View File
@@ -0,0 +1,295 @@
"""Catalog's service and queries, one use case each, below the HTTP layer.
The routes are thin callers now, and their own tests stay the contract a client sees. These hold
what each write does and leaves on its own: that it flushes and leaves the commit to its caller,
that a row of someone else's reads as a missing one, and the rule each write carries (an explored
channel flagged only while nobody follows it, a subscription that keeps it, a cleared search that
spares what anyone holds). DB-backed (real Postgres).
"""
from datetime import datetime, timedelta, timezone
from sqlalchemy import select
from app.catalog import queries, search, service
from app.catalog.models import (
BlockedChannel,
Channel,
ExploredChannel,
SearchFind,
Subscription,
Video,
)
from app.reader import queries as reader_queries
from app.reader import service as reader_service
def _channel(db, cid="UCc", **kw):
ch = Channel(id=cid, title=kw.pop("title", cid), **kw)
db.add(ch)
db.commit()
return ch
def _video(db, vid, cid="UCc", **kw):
v = Video(id=vid, channel_id=cid, published_at=datetime.now(timezone.utc), **kw)
db.add(v)
db.commit()
return v
def _sub(db, user, cid="UCc", **kw):
s = Subscription(user_id=user.id, channel_id=cid, **kw)
db.add(s)
db.commit()
return s
def _uncommitted(db):
return bool(db.new or db.dirty or db.deleted) or db.in_transaction()
# --- whose a row is ------------------------------------------------------------------------
def test_a_users_subscriptions_blocks_and_explorations_are_theirs_alone(db, make_user):
owner, stranger = make_user(), make_user()
ch = _channel(db)
_sub(db, owner)
service.explore(db, owner, ch)
service.block(db, owner, "UCc")
db.commit()
assert queries.subscription(db, owner, "UCc") is not None
assert queries.subscription(db, stranger, "UCc") is None
assert queries.exploration(db, stranger, "UCc") is None
assert queries.is_blocked(db, stranger, "UCc") is False
assert queries.blocked_ids(db, stranger) == set()
assert [s.channel_id for s, _ in queries.subscriptions(db, stranger, "en")] == []
# --- subscription settings -------------------------------------------------------------------
def test_full_history_reports_only_the_switch_on(db, make_user):
user = make_user()
_channel(db)
sub = _sub(db, user)
assert service.update_subscription(db, sub, priority=3) is False
assert service.update_subscription(db, sub, deep_requested=True) is True
assert service.update_subscription(db, sub, deep_requested=True) is False # already on
assert _uncommitted(db)
db.commit()
assert (sub.priority, sub.hidden, sub.deep_requested) == (3, False, True)
def test_full_history_for_all_moves_only_the_callers_rows(db, make_user):
user, other = make_user(), make_user()
_channel(db)
_sub(db, user)
theirs = _sub(db, other)
assert service.set_deep_all(db, user, True) == 1
db.commit()
db.expire_all()
assert theirs.deep_requested is False
def test_a_reset_asks_every_subscriber_for_full_history(db, make_user):
a, b = make_user(), make_user()
ch = _channel(db, backfill_done=True, recent_synced_at=datetime.now(timezone.utc))
_sub(db, a)
_sub(db, b)
service.reset_backfill(db, ch)
db.commit()
db.expire_all()
assert (ch.backfill_done, ch.recent_synced_at) == (False, None)
assert set(db.scalars(select(Subscription.deep_requested))) == {True}
# --- explore, subscribe, block ---------------------------------------------------------------
def test_an_exploration_is_one_row_and_flags_the_channel_only_while_nobody_follows_it(
db, make_user
):
explorer, follower = make_user(), make_user()
lone = _channel(db, "UClone")
kept = _channel(db, "UCkept")
_sub(db, follower, "UCkept")
service.explore(db, explorer, lone)
service.explore(db, explorer, kept)
db.commit()
first = queries.exploration(db, explorer, "UClone").created_at
service.explore(db, explorer, lone) # refreshes the grace clock
db.commit()
assert db.scalar(select(ExploredChannel.id).where(ExploredChannel.channel_id == "UClone"))
assert len(db.scalars(select(ExploredChannel.id)).all()) == 2
assert queries.exploration(db, explorer, "UClone").created_at >= first
assert (lone.from_explore, kept.from_explore) == (True, False)
def test_subscribing_keeps_an_explored_channel_and_its_videos(db, make_user):
user = make_user()
ch = _channel(db, from_explore=True)
v = _video(db, "v1", via_explore=True)
service.record_subscription(db, user, ch, "")
db.commit()
db.expire_all()
sub = queries.subscription(db, user, "UCc")
assert (sub.yt_subscription_id, ch.from_explore, v.via_explore) == (None, False, False)
service.drop_subscription(db, sub)
db.commit()
assert queries.subscription(db, user, "UCc") is None
def test_a_block_is_once_and_ends_the_exploration(db, make_user):
user = make_user()
ch = _channel(db)
service.explore(db, user, ch)
db.commit()
service.block(db, user, "UCc")
service.block(db, user, "UCc")
db.commit()
assert len(db.scalars(select(BlockedChannel.id)).all()) == 1
assert queries.exploration(db, user, "UCc") is None
assert [c.id for c in queries.blocked_channels(db, user, "en")] == ["UCc"]
service.unblock(db, user, "UCc")
db.commit()
assert queries.is_blocked(db, user, "UCc") is False
# --- what the status cards count -------------------------------------------------------------
def test_my_progress_counts_only_my_channels_and_my_unhidden_videos(db, make_user):
user, other = make_user(), make_user()
now = datetime.now(timezone.utc)
_channel(db, "UCa", details_synced_at=now, recent_synced_at=now, backfill_done=True)
_channel(db, "UCb")
_channel(db, "UCtheirs")
_video(db, "a1", "UCa")
_video(db, "b1", "UCb")
_video(db, "t1", "UCtheirs")
_sub(db, user, "UCa", deep_requested=True)
_sub(db, user, "UCb", hidden=True)
_sub(db, other, "UCtheirs")
assert queries.my_progress(db, user) == (2, 1, 1, 1)
assert [c.id for c in queries.my_deep_requested(db, user)] == ["UCa"]
assert queries.my_video_count(db, user) == 1
assert queries.video_total(db) == 3
assert queries.catalog_totals(db, user)["subscriptions"] == 2
def test_a_channel_with_no_title_is_left_out_of_the_titles(db):
_channel(db, "UCt", title="Titled")
db.add(Channel(id="UCn", title=None))
db.commit()
assert queries.channel_titles(db, {"UCt", "UCn", "UCgone"}) == {"UCt": "Titled"}
assert queries.channel_title(db, "UCt") == "Titled"
def test_the_deep_backlog_is_the_unfinished_channels_someone_asked_for(db, make_user):
user = make_user()
_channel(db, "UCwant")
_channel(db, "UCdone", backfill_done=True)
_channel(db, "UCnobody")
_sub(db, user, "UCwant", deep_requested=True)
_sub(db, user, "UCdone", deep_requested=True)
_sub(db, user, "UCnobody")
assert [c.id for c in queries.deep_backlog(db)] == ["UCwant"]
assert queries.deep_queued(db, ["UCwant", "UCdone", "UCnobody"]) == {"UCwant", "UCdone"}
# --- live search -----------------------------------------------------------------------------
def test_a_cleared_search_spares_what_anyone_holds(db, make_user):
user, other, follower = make_user(), make_user(), make_user()
_channel(db, "UCs")
_channel(db, "UCfollowed")
_sub(db, follower, "UCfollowed")
for vid in ("loose", "watched", "found_by_other"):
_video(db, vid, "UCs", via_search=True)
_video(db, "followed", "UCfollowed", via_search=True)
_video(db, "organic", "UCs")
ids = ["loose", "watched", "found_by_other", "followed", "organic"]
search.record_finds(db, user, ids, "cat")
search.record_finds(db, user, ids, "cat") # idempotent
search.record_finds(db, other, ["found_by_other"], "dog")
reader_service.mark(db, other, "watched", "watched")
db.commit()
assert search.clear(db, user, ids) == 1
db.commit()
left = set(db.scalars(select(Video.id)))
assert left == {"watched", "found_by_other", "followed", "organic"}
assert db.scalars(select(SearchFind.user_id)).all() == [other.id]
def test_a_search_puts_what_the_user_already_has_last(db, make_user):
user = make_user()
_channel(db, "UCs")
_channel(db, "UCmine")
_sub(db, user, "UCmine")
for vid, cid in (("new1", "UCs"), ("mine", "UCmine"), ("seen", "UCs"), ("new2", "UCs")):
_video(db, vid, cid)
reader_service.mark(db, user, "seen", "hidden")
db.commit()
assert search.new_first(db, user, ["mine", "new1", "seen", "new2"]) == [
"new1",
"new2",
"mine",
"seen",
]
# --- what Catalog asks Reader ----------------------------------------------------------------
def test_a_users_tag_links_are_made_once_and_only_theirs_are_read(db, make_user):
user, other = make_user(), make_user()
_channel(db)
mine = reader_service.create_tag(db, user, "Mine", "other", None)
theirs = reader_service.create_tag(db, other, "Theirs", "other", None)
reader_service.attach_user_tag(db, user, "UCc", mine.id)
reader_service.attach_user_tag(db, user, "UCc", mine.id)
reader_service.attach_user_tag(db, other, "UCc", theirs.id)
db.commit()
assert reader_queries.user_tag_links(db, user) == {"UCc": [mine.id]}
assert reader_queries.user_tag_links(db, user, "UCelse") == {}
reader_service.detach_user_tag(db, user, "UCc", mine.id)
db.commit()
assert reader_queries.user_tag_links(db, user) == {}
assert reader_queries.user_tag_links(db, other) == {"UCc": [theirs.id]}
def test_discovery_offers_the_playlist_channels_not_followed_and_not_ones_own(db, make_user):
user = make_user()
for cid in ("UCnew", "UCfollowed", "UCown"):
_channel(db, cid)
for vid, cid in (("n1", "UCnew"), ("n2", "UCnew"), ("f1", "UCfollowed"), ("o1", "UCown")):
_video(db, vid, cid)
_sub(db, user, "UCfollowed")
pl = reader_service.create_playlist(db, user, "L")
for vid in ("n1", "n2", "f1", "o1"):
reader_service.add_item(db, pl, vid, mark_dirty=False)
db.commit()
rows = reader_queries.playlist_channels(db, user, "UCown", "en")
assert [(ch.id, n, lists) for ch, n, lists in rows] == [("UCnew", 2, 1)]
def test_an_exploration_older_than_its_grace_is_still_one_row(db, make_user):
"""The grace clock is refreshed in place: re-exploring after the cutoff does not add a row."""
user = make_user()
ch = _channel(db)
db.add(
ExploredChannel(
user_id=user.id,
channel_id="UCc",
created_at=datetime.now(timezone.utc) - timedelta(days=30),
)
)
db.commit()
service.explore(db, user, ch)
db.commit()
rows = db.scalars(select(ExploredChannel)).all()
assert len(rows) == 1
assert rows[0].created_at > datetime.now(timezone.utc) - timedelta(minutes=1)
+2 -1
View File
@@ -12,7 +12,8 @@ import app.state as state
from app.auth import current_user
from app.db import get_db
from app.main import app
from app.models import Channel, Subscription, User, Video
from app.models import User
from app.catalog.models import Channel, Subscription, Video
from app.sync import runner, videos
_PUB = datetime(2020, 1, 1, tzinfo=timezone.utc)
@@ -10,7 +10,7 @@ from datetime import datetime, timedelta, timezone
import pytest
from app.models import Channel, Subscription
from app.catalog.models import Channel, Subscription
from app.routes import channels
from app.sync import subscriptions
from app.sync.subscriptions import DETAILS_MAX_AGE, import_subscriptions
@@ -14,8 +14,8 @@ from sqlalchemy import text
from sqlalchemy.dialects import postgresql
from app.feedquery import filtered_query
from app.models import Video
from app.routes.channels import channel_aggregates
from app.catalog.models import Video
from app.catalog.queries import channel_aggregates
INDEX = "ix_videos_channel_pub_cover"
@@ -9,7 +9,7 @@ import pytest
from fastapi import HTTPException
from sqlalchemy import select
from app.models import Channel, Subscription
from app.catalog.models import Channel, Subscription
from app.routes import channels
from app.youtube.client import YouTubeError
+2 -1
View File
@@ -14,7 +14,8 @@ import app.state as state
from app.auth import current_user
from app.db import get_db
from app.main import app
from app.models import Channel, User
from app.models import User
from app.catalog.models import Channel
from app.reader.models import ChannelTag, Tag
+2 -1
View File
@@ -18,7 +18,8 @@ from app.downloads import errors as dlerrors
from app.downloads import formats, pipeline, quota, runner, service
from app.downloads import probe as probemod
from app.downloads.edit import INPUT_GUARD
from app.models import Channel, PlexItem, PlexLibrary, PlexShow, Video
from app.models import PlexItem, PlexLibrary, PlexShow
from app.catalog.models import Channel, Video
from app.downloads.models import DownloadJob, MediaAsset
from app.plex import paths as plex_paths
+3 -2
View File
@@ -9,7 +9,8 @@ from app.auth import current_user
from app.collation import sorted_words
from app.db import get_db
from app.main import app
from app.models import Channel, PlexCollection, PlexItem, PlexLibrary, Subscription, User, Video
from app.models import PlexCollection, PlexItem, PlexLibrary, User
from app.catalog.models import Channel, Subscription, Video
from app.reader.models import Playlist, PlaylistItem
from datetime import datetime, timezone
@@ -222,7 +223,7 @@ def test_the_share_recipients_are_in_the_readers_alphabet(db, monkeypatch, overr
def test_the_blocked_channels_are_in_the_readers_alphabet(db, monkeypatch, override_dep):
from app.models import BlockedChannel
from app.catalog.models import BlockedChannel
user = _user(db, "hu")
db.add_all(Channel(id=f"UC{i}", title=w) for i, w in enumerate(WORDS))
+2 -11
View File
@@ -17,17 +17,8 @@ from app.discover import hero, pins, queries, rails
from app.discover.budget import BANDS, fits, remaining_seconds
from app.plex import rollup
from app.plex.cards import leaf_card
from app.models import (
BlockedChannel,
Channel,
PlexItem,
PlexLibrary,
PlexShow,
PlexState,
Subscription,
User,
Video,
)
from app.models import PlexItem, PlexLibrary, PlexShow, PlexState, User
from app.catalog.models import BlockedChannel, Channel, Subscription, Video
from app.reader.models import ComfortPin, Playlist, PlaylistItem, VideoState
NOW = datetime(2026, 6, 1, tzinfo=timezone.utc)
+2 -1
View File
@@ -12,7 +12,8 @@ contradicts the remote's own count is not believed either.
import pytest
from sqlalchemy import select
from app.models import Channel, PlexCollection, PlexItem, PlexLibrary, Video
from app.models import PlexCollection, PlexItem, PlexLibrary
from app.catalog.models import Channel, Video
from app.reader.models import Playlist, PlaylistItem
from app.plex import sync as plex_sync
from app.sync import playlists as plsync
+1 -1
View File
@@ -6,7 +6,7 @@ from datetime import datetime, timedelta, timezone
import pytest
from app import sysconfig
from app.models import Channel, Video
from app.catalog.models import Channel, Video
from app.sync import runner
from app.youtube.client import YouTubeError
+2 -1
View File
@@ -16,7 +16,8 @@ from datetime import datetime, timezone
import pytest
from app.models import Channel, Subscription, User, Video
from app.models import User
from app.catalog.models import Channel, Subscription, Video
# Every sort feedquery.sort_keys defines except "relevance" (which needs a search query to rank).
_SORTS = [
+2 -1
View File
@@ -12,7 +12,8 @@ from datetime import datetime, timezone
import pytest
from app import feedquery
from app.models import Channel, Subscription, User, Video
from app.models import User
from app.catalog.models import Channel, Subscription, Video
def _video(vid: str, day: int, title: str) -> Video:
+1 -1
View File
@@ -17,7 +17,7 @@ from sqlalchemy.dialects import postgresql
from app import collation
from app.db import engine
from app.feedquery import order_by, sort_keys
from app.models import Video
from app.catalog.models import Video
VIDEO_SORTS = [
"newest",
+1 -1
View File
@@ -14,7 +14,7 @@ import pytest
from app import feedquery
from app.models import Channel, Subscription, Video
from app.catalog.models import Channel, Subscription, Video
from app.reader.models import ChannelTag, Tag, VideoState
+1 -1
View File
@@ -7,7 +7,7 @@ by someone. The admin card counting "channels to sync" must use the same selecti
channels the backfill will never touch. DB lane.
"""
from app.models import Channel, ExploredChannel, Subscription
from app.catalog.models import Channel, ExploredChannel, Subscription
from app.sync import runner
+2 -2
View File
@@ -9,8 +9,8 @@ from datetime import datetime, timezone
from sqlalchemy import select
from app.models import Channel, Subscription, Video
from app.routes.search import _fold_term
from app.catalog.models import Channel, Subscription, Video
from app.catalog.search import _fold_term
def _video(vid: str, ch: str, title: str, search_terms: str | None = None) -> Video:
+2 -1
View File
@@ -12,7 +12,8 @@ from datetime import datetime, timedelta, timezone
import pytest
from sqlalchemy import select
from app.models import Channel, Notification, Video
from app.models import Notification
from app.catalog.models import Channel, Video
from app.reader.models import Playlist, PlaylistItem
from app.sync import maintenance
+1 -1
View File
@@ -3,7 +3,7 @@ ones this user asked full history for. The numerator used to count every subscri
back-catalog was complete (another user had asked, or the channel was short), so the page read
"342/329". "Channels without full history yet" still counts every subscribed channel. Needs the DB."""
from app.models import Channel, Subscription
from app.catalog.models import Channel, Subscription
def test_full_history_ratio_counts_only_the_requested_channels(db, make_user, as_user):
+1 -1
View File
@@ -8,7 +8,7 @@ the position is an index over the ORDERED rows, and the list is renumbered aroun
from datetime import datetime, timezone
from app.models import Channel, Video
from app.catalog.models import Channel, Video
from app.reader.models import Playlist, PlaylistItem
+1 -1
View File
@@ -10,7 +10,7 @@ from datetime import datetime, timezone
import pytest
from app.models import Channel, Video
from app.catalog.models import Channel, Video
from app.reader.models import Playlist, PlaylistItem
from app.sync import playlists as pl_mod
from app.sync.playlist_merge import merge_plan
+1 -1
View File
@@ -16,7 +16,7 @@ from datetime import datetime, timedelta, timezone
import pytest
from sqlalchemy.exc import OperationalError
from app.models import Channel, Video
from app.catalog.models import Channel, Video
from app.reader.models import Playlist, PlaylistItem
+2 -1
View File
@@ -19,7 +19,8 @@ from datetime import datetime, timezone
import pytest
from sqlalchemy import event
from app.models import Channel, PlexItem, PlexLibrary, PlexPlaylist, PlexPlaylistItem, Video
from app.models import PlexItem, PlexLibrary, PlexPlaylist, PlexPlaylistItem
from app.catalog.models import Channel, Video
from app.reader.models import Playlist, PlaylistItem
@@ -18,7 +18,7 @@ import pytest
from sqlalchemy import delete, event, text
from app.db import SessionLocal
from app.models import Channel, Video
from app.catalog.models import Channel, Video
from app.reader.models import Playlist, PlaylistItem
from app.sync import playlists as pl_mod
+1 -1
View File
@@ -14,7 +14,7 @@ from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from app.discover import pins
from app.models import Channel, Video
from app.catalog.models import Channel, Video
from app.reader import queries, service
from app.reader.models import ChannelTag, ComfortPin, Playlist, PlaylistItem, Tag, VideoState
-3
View File
@@ -20,9 +20,6 @@ COMMITTING_HELPERS = {
"not hand the token back",
"auth.py:_complete_link": "the tail of the Google callback route, which answers with its result",
"auth.py:_register_account": "a background task with its own session, i.e. a job wrapper",
"search.py:_ingest_candidates": "stages around YouTube calls: the stubs must survive an enrich "
"that fails",
"search.py:_classify_shorts": "the last stage of that ingest, after its probes",
"auth.py:upsert_pending_invite": "commits only when it writes — the contract its two "
"callers rely on (`_register_account` persists its own flushed user because a no-op here "
"commits nothing)",
+2 -1
View File
@@ -13,7 +13,8 @@ import pytest
from sqlalchemy import select
from app import jobruns, logs, logsink, quota, scheduler, state, sysconfig
from app.models import AuditLog, Channel, JobRun, Notification, QuotaEvent
from app.models import AuditLog, JobRun, Notification, QuotaEvent
from app.catalog.models import Channel
from app.routes import scheduler as scheduler_routes
from app.runstatus import RunStatus, RunTrigger
+7 -6
View File
@@ -14,8 +14,9 @@ a placeholder never becomes the video's permanent date.
from datetime import datetime, timezone
from app import sysconfig
from app.models import User, Video
from app.routes.search import _ingest_candidates
from app.models import User
from app.catalog.models import Video
from app.catalog.search import ingest_candidates
class _FakeYT:
@@ -70,7 +71,7 @@ def test_scrape_ingest_dateless_stub_ingests_and_takes_real_enrich_date(db):
"vid_bbbbbbb": _enrich_item("vid_bbbbbbb", "2023-10-30T17:00:08Z"),
}
kept = _ingest_candidates(db, _FakeYT(enrich), user, candidates)
kept = ingest_candidates(db, _FakeYT(enrich), user, candidates)
# No IntegrityError, and both non-Short/non-live videos are kept in input order.
assert kept == ["vid_aaaaaaa", "vid_bbbbbbb"]
@@ -88,7 +89,7 @@ def test_scrape_ingest_enrich_miss_keeps_nonnull_placeholder(db):
db.add(user)
db.commit()
kept = _ingest_candidates(db, _FakeYT({}), user, [_candidate("vid_ccccccc", "UC_chan_c")])
kept = ingest_candidates(db, _FakeYT({}), user, [_candidate("vid_ccccccc", "UC_chan_c")])
assert kept == []
c = db.get(Video, "vid_ccccccc")
@@ -100,7 +101,7 @@ def test_search_youtube_route_returns_422_when_ingest_fails(db, monkeypatch, ove
# Route-level guard (B-core-1): an ingest failure with nothing yet collected must surface as a 422
# + message on this JSON endpoint (the FE hides 5xx behind a generic "connection lost" + auto-retry,
# burning a second quota-costed search), and the session is rolled back. Here the page fetch succeeds
# (one candidate) and _ingest_candidates raises → the endpoint returns 422.
# (one candidate) and ingest_candidates raises → the endpoint returns 422.
from fastapi.testclient import TestClient
import app.state as state
@@ -143,7 +144,7 @@ def test_search_youtube_route_returns_422_when_ingest_fails(db, monkeypatch, ove
def _boom(db_, yt, user_, candidates):
raise RuntimeError("ingest exploded")
monkeypatch.setattr(search_mod, "_ingest_candidates", _boom)
monkeypatch.setattr(search_mod.catalog_search, "ingest_candidates", _boom)
override_dep(get_db, db.info["request"])
override_dep(current_user, lambda: user)
+2 -1
View File
@@ -11,7 +11,8 @@ network, pure DB state.
Uses the DB lane (`db` fixture) — skipped when no Postgres is reachable.
"""
from app.models import Channel, Subscription, User
from app.models import User
from app.catalog.models import Channel, Subscription
from app.reader.models import ChannelTag, Tag
from app.sync.tagsuggest import (
AUTOAPPLY_THRESHOLD,
+42 -8
View File
@@ -110,7 +110,6 @@ UNIT_ROUTES = {
# What each HTTP file still does, by file: the debt a unit's sprint pays off. Exact counts. A core
# route's count is only the unit tables it imports; it is paid by the sprint of the unit it reaches.
ALLOWED = {
"channels.py": 83,
"plex/_shared.py": 1,
"plex/admin.py": 12,
"plex/browse.py": 93,
@@ -120,10 +119,6 @@ ALLOWED = {
"plex/people.py": 20,
"plex/playback.py": 42,
"plex/playlists.py": 80,
"runs.py": 1,
"scheduler.py": 3,
"search.py": 46,
"sync.py": 38,
}
# The session methods that read or write data. `commit` and `rollback` stay with the handler.
@@ -341,11 +336,22 @@ UNIT_PACKAGES = {
"sync/playlist_merge.py",
"sync/tagsuggest.py",
),
# The catalog's ingest jobs: the subscription import, the video sync and backfills, the explore
# and search ingests, the maintenance re-checks and the auto-tagger.
"catalog": (
"catalog/",
"sync/autotag.py",
"sync/explore.py",
"sync/maintenance.py",
"sync/runner.py",
"sync/subscriptions.py",
"sync/videos.py",
),
}
# The units fenced so far; the rest join with their sprint.
FENCED = {"capture", "reader"}
FENCED = {"capture", "reader", "catalog"}
# Whose tables a unit's own code may READ. Core code reads a fenced unit's tables through its queries.
READS = {"capture": {"catalog", "plex"}, "reader": {"catalog", "plex"}}
READS = {"capture": {"catalog", "plex"}, "reader": {"catalog", "plex"}, "catalog": set()}
# The columns that say whose a row is.
OWNER_COLUMNS = {"user_id", "to_user_id", "from_user_id", "created_by"}
# Where a fenced unit's code may commit: the jobs that own their session (the worker and what it
@@ -372,6 +378,32 @@ UNIT_COMMITS = {
"sync/playlists.py:merge_playlist",
"sync/playlists.py:repull_playlist",
},
# The catalog's ingest jobs commit per page or batch, a YouTube call between each: a page
# fetched and stored is not lost when the next call fails, and an enrich that fails rolls back
# to the stubs the step before committed (a search's results, a playlist's unseen videos).
"catalog": {
"catalog/search.py:classify_shorts",
"catalog/search.py:ingest_candidates",
"sync/autotag.py:apply_channel_autotags",
"sync/autotag.py:run_autotag_all",
"sync/explore.py:_enrich_new",
"sync/explore.py:purge_unkept_explored",
"sync/explore.py:purge_unkept_search",
"sync/maintenance.py:_notify_removed",
"sync/maintenance.py:_recheck_flagged",
"sync/maintenance.py:_revalidate_rolling",
"sync/runner.py:run_feed_fallback",
"sync/runner.py:_stamp_fallback_attempt",
"sync/subscriptions.py:import_subscriptions",
"sync/videos.py:insert_stubs",
"sync/videos.py:apply_rss_feed",
"sync/videos.py:backfill_channel_recent",
"sync/videos.py:backfill_channel_deep",
"sync/videos.py:reconcile_full_history",
"sync/videos.py:_apply_video_batch",
"sync/videos.py:run_shorts_classification",
"sync/videos.py:ensure_videos",
},
}
SQL_WRITES = {"insert", "update", "delete", "pg_insert"}
@@ -506,7 +538,9 @@ PACKAGE_CASES = {
("downloads/x.py", "from app.models import Video\nupdate(Video)"): 1,
("downloads/x.py", "from app.downloads.models import DownloadJob\nDownloadJob()"): 0,
("search.py", "from app.downloads.models import DownloadJob"): 1, # core reads capture
("search.py", "from app.models import Video\nVideo()"): 0, # neither side fenced yet
("search.py", "from app.models import PlexItem\nPlexItem()"): 0, # neither side fenced yet
("search.py", "from app.models import Video\nVideo()"): 2, # core reads and writes catalog
("sync/videos.py", "from app.models import Playlist"): 1, # no catalog→reader edge
("downloads/x.py", "q = DownloadJob.user_id == 1"): 1,
("downloads/x.py", "q = owned_by(DownloadJob.user_id, 1)"): 0,
("downloads/x.py", "def f(db):\n db.commit()"): 1,
+2 -1
View File
@@ -8,7 +8,8 @@ import httpx
import pytest
from app.auth import has_read_scope
from app.models import Channel, OAuthToken
from app.models import OAuthToken
from app.catalog.models import Channel
from app.sync.explore import explore_ingest_page
from app.sync.videos import backfill_channel_deep, backfill_channel_recent
from app.youtube import client as yt
+1 -1
View File
@@ -113,7 +113,7 @@ project checks needs a row.
| CVE gate | HIGH/CRITICAL with a fix available only | `--ignore-unfixed` + `--severity HIGH,CRITICAL`: a finding with no fix cannot be acted on by an upgrade; it shows in the scan output. A fixable one fails the publish, and the only way past it is `.trivyignore.yaml`. | 0 open (E32 S1) |
| npm audit | `--omit=dev --audit-level=high` | What ships to the browser, at high or above. The dev toolchain's advisories were burned down by hand in E32 S1 (18 → 0). | 0 |
| knip | `entry` — one file added | Not a narrowing: knip's Vite plugin reads entries from index.html only, so the shared link's page (`watch.html` → `src/watchMain.tsx`) is named here, or the lane reports that whole entry unused. It adds files to the lane, never takes any out. | 1 entry |
| unit fence | the database work the HTTP layer still does, pinned | `backend/tests/test_unit_fences.py`: a unit's route file may not build SQL, call the session's data methods, set a model's attributes or import a unit's tables; a core route may not import a unit's tables. Switched on unit by unit: `ALLOWED` pins each file's count exactly, so a rise fails and a fall fails until the pin is lowered — the list may only shrink. End state: empty, and the pin mechanism goes. A unit in `FENCED` is held below its HTTP layer too: its code reads another unit's tables only along `READS`, writes them never, filters owners through `userscope`, and commits only where `UNIT_COMMITS` names a job's own session. | 826 in 23 files (E42 S0) → 667 in 19 (S1: Capture 0, fenced) → 525 in 14 (S2: Reader 0, fenced) |
| unit fence | the database work the HTTP layer still does, pinned | `backend/tests/test_unit_fences.py`: a unit's route file may not build SQL, call the session's data methods, set a model's attributes or import a unit's tables; a core route may not import a unit's tables. Switched on unit by unit: `ALLOWED` pins each file's count exactly, so a rise fails and a fall fails until the pin is lowered — the list may only shrink. End state: empty, and the pin mechanism goes. A unit in `FENCED` is held below its HTTP layer too: its code reads another unit's tables only along `READS`, writes them never, filters owners through `userscope`, and commits only where `UNIT_COMMITS` names a job's own session. | 826 in 23 files (E42 S0) → 667 in 19 (S1: Capture 0, fenced) → 525 in 14 (S2: Reader 0, fenced) → 354 in 9 (S3: Catalog 0, fenced; core routes 0) |
| React Compiler | the functions it skips, pinned | `components/compilerCoverage.contract.test.ts` — the list may only shrink. End state: only VirtualGrid (TanStack Virtual, an IncompatibleLibrary) remains. | 93 (E32 S1) → 29 (E32 S3) → 1, VirtualGrid (E32 S4) |
## Accepted findings
+2 -7
View File
@@ -24,13 +24,8 @@ sys.path.insert(0, "/app")
from sqlalchemy import delete, func, select
from app.db import SessionLocal
from app.models import (
AppState,
Channel,
Subscription,
User,
Video,
)
from app.catalog.models import Channel, Subscription, Video
from app.models import AppState, User
from app.reader.models import Playlist, PlaylistItem, VideoState
from app.security import hash_password