Houmeres/SageTube
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity
Files
a8ebe89cc14752ab3ff8e00cb9879bd5af8e06ff
SageTube/ytsage/utils/ytsage_library_manager.py
T
Jaroslav Beneš 3afe96a8cd Add local subscriptions and the Feed tab 
New ytsage/utils/ytsage_library_manager.py: sagetube_library.db
(separate from the upstream download-history DB) with WAL from day one,
holding subscriptions, cached feed_items, watch_history with resume
positions, and the persisted play_queue. Watch history and queue tables
are wired up by the next commit.

FeedPage:
- Local mode: FeedRefreshWorker refreshes each subscribed channel
  sequentially (feed.per_channel_items, default 15) and the grid fills
  incrementally per channel; results are cached so the feed is
  populated instantly on startup.
- Account mode: fetches youtube.com/feed/subscriptions with the user's
  browser cookies - the real logged-in feed; the option is enabled only
  while cookies are active and errors surface as a status banner.
- Sidebar lists subscriptions (double-click opens the channel in
  Browse; context menu unsubscribes). Browse's Subscribe button now
  toggles subscription state through the Feed page.

Verified live: subscribe -> refresh -> 15 videos cached in SQLite and
rendered as cards.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-25 01:56:12 +02:00

268 lines
11 KiB
Python
Raw Blame History

"""
LibraryManager - SageTube's watch-side persistence
==================================================
Separate SQLite database (sagetube_library.db) holding local channel
subscriptions, the aggregated feed cache, watch history with resume
positions, and the persisted play queue. Kept apart from the upstream
download-history DB so upstream merges never collide.
Same concurrency pattern as HistoryManager: one persistent connection,
check_same_thread=False, an RLock around every operation, WAL journaling.
"""
import sqlite3
import threading
import time
from typing import Any, Dict, List, Optional
from .ytsage_constants import APP_DATA_DIR
from .ytsage_logger import logger
_SCHEMA_VERSION = 1
class LibraryManager:
_lock = threading.RLock()
_connection: Optional[sqlite3.Connection] = None
_db_file = APP_DATA_DIR / "sagetube_library.db"
# ------------------------------------------------------------- plumbing
@classmethod
def _conn(cls) -> sqlite3.Connection:
with cls._lock:
if cls._connection is None:
cls._db_file.parent.mkdir(parents=True, exist_ok=True)
cls._connection = sqlite3.connect(cls._db_file, check_same_thread=False)
cls._connection.row_factory = sqlite3.Row
cls._connection.execute("PRAGMA journal_mode=WAL")
cls._connection.execute("PRAGMA busy_timeout=5000")
cls._init_schema()
return cls._connection
@classmethod
def _init_schema(cls) -> None:
assert cls._connection is not None
cur = cls._connection.cursor()
cur.executescript(
"""
CREATE TABLE IF NOT EXISTS subscriptions (
channel_id TEXT PRIMARY KEY,
title TEXT NOT NULL,
url TEXT NOT NULL,
avatar_url TEXT,
added_at INTEGER NOT NULL,
last_refreshed INTEGER
);
CREATE TABLE IF NOT EXISTS feed_items (
video_id TEXT PRIMARY KEY,
channel_id TEXT NOT NULL,
title TEXT,
url TEXT NOT NULL,
duration REAL,
thumbnail_url TEXT,
published_ts INTEGER,
fetched_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_feed_channel ON feed_items(channel_id, fetched_at DESC);
CREATE TABLE IF NOT EXISTS watch_history (
video_id TEXT PRIMARY KEY,
url TEXT NOT NULL,
title TEXT,
channel TEXT,
channel_id TEXT,
duration REAL,
position REAL DEFAULT 0,
completed INTEGER DEFAULT 0,
watched_at INTEGER NOT NULL,
thumbnail_url TEXT
);
CREATE TABLE IF NOT EXISTS play_queue (
id INTEGER PRIMARY KEY AUTOINCREMENT,
video_id TEXT,
url TEXT NOT NULL,
title TEXT,
channel TEXT,
duration REAL,
thumbnail_url TEXT,
sort_order INTEGER NOT NULL,
added_at INTEGER NOT NULL
);
"""
)
cur.execute(f"PRAGMA user_version = {_SCHEMA_VERSION}")
cls._connection.commit()
# -------------------------------------------------------- subscriptions
@classmethod
def subscribe(cls, channel_id: str, title: str, url: str, avatar_url: Optional[str] = None) -> None:
with cls._lock:
cls._conn().execute(
"INSERT INTO subscriptions (channel_id, title, url, avatar_url, added_at) VALUES (?,?,?,?,?) "
"ON CONFLICT(channel_id) DO UPDATE SET title=excluded.title, url=excluded.url",
(channel_id, title, url, avatar_url, int(time.time())),
)
cls._conn().commit()
logger.info(f"Subscribed to {title} ({channel_id})")
@classmethod
def unsubscribe(cls, channel_id: str) -> None:
with cls._lock:
cls._conn().execute("DELETE FROM subscriptions WHERE channel_id = ?", (channel_id,))
cls._conn().execute("DELETE FROM feed_items WHERE channel_id = ?", (channel_id,))
cls._conn().commit()
@classmethod
def is_subscribed(cls, channel_id: str) -> bool:
with cls._lock:
row = cls._conn().execute("SELECT 1 FROM subscriptions WHERE channel_id = ?", (channel_id,)).fetchone()
return row is not None
@classmethod
def subscriptions(cls) -> List[Dict[str, Any]]:
with cls._lock:
rows = cls._conn().execute("SELECT * FROM subscriptions ORDER BY title COLLATE NOCASE").fetchall()
return [dict(r) for r in rows]
@classmethod
def mark_refreshed(cls, channel_id: str) -> None:
with cls._lock:
cls._conn().execute(
"UPDATE subscriptions SET last_refreshed = ? WHERE channel_id = ?", (int(time.time()), channel_id)
)
cls._conn().commit()
# ---------------------------------------------------------------- feed
@classmethod
def upsert_feed_items(cls, channel_id: str, entries: List[Dict[str, Any]]) -> None:
now = int(time.time())
with cls._lock:
conn = cls._conn()
for i, e in enumerate(entries):
video_id = e.get("id")
url = e.get("url") or (f"https://www.youtube.com/watch?v={video_id}" if video_id else None)
if not video_id or not url:
continue
# fetched_at encodes per-channel recency order (newest first)
conn.execute(
"INSERT INTO feed_items (video_id, channel_id, title, url, duration, thumbnail_url, published_ts, fetched_at) "
"VALUES (?,?,?,?,?,?,?,?) "
"ON CONFLICT(video_id) DO UPDATE SET title=excluded.title, duration=excluded.duration",
(
video_id,
channel_id,
e.get("title"),
url,
e.get("duration"),
e.get("thumbnail") or (e.get("thumbnails") or [{}])[-1].get("url"),
e.get("timestamp") or e.get("release_timestamp"),
now - i,
),
)
conn.commit()
@classmethod
def feed_items(cls, limit: int = 120) -> List[Dict[str, Any]]:
with cls._lock:
rows = cls._conn().execute(
"SELECT f.*, s.title AS channel FROM feed_items f "
"LEFT JOIN subscriptions s ON s.channel_id = f.channel_id "
"ORDER BY COALESCE(f.published_ts, f.fetched_at) DESC LIMIT ?",
(limit,),
).fetchall()
return [dict(r) for r in rows]
# ------------------------------------------------------- watch history
@classmethod
def upsert_watch(cls, entry: Dict[str, Any]) -> None:
video_id = entry.get("id") or entry.get("url")
url = entry.get("url") or entry.get("webpage_url")
if not video_id or not url:
return
with cls._lock:
cls._conn().execute(
"INSERT INTO watch_history (video_id, url, title, channel, channel_id, duration, watched_at, thumbnail_url) "
"VALUES (?,?,?,?,?,?,?,?) "
"ON CONFLICT(video_id) DO UPDATE SET watched_at=excluded.watched_at, title=excluded.title, url=excluded.url",
(
str(video_id),
url,
entry.get("title"),
entry.get("channel") or entry.get("uploader"),
entry.get("channel_id"),
entry.get("duration"),
int(time.time()),
entry.get("thumbnail"),
),
)
cls._conn().commit()
@classmethod
def update_position(cls, video_id: str, position: float, duration: Optional[float] = None) -> None:
with cls._lock:
completed = 0
if duration and duration > 0 and position >= duration * 0.95:
completed = 1
cls._conn().execute(
"UPDATE watch_history SET position = ?, completed = MAX(completed, ?), "
"duration = COALESCE(?, duration) WHERE video_id = ?",
(position, completed, duration, str(video_id)),
)
cls._conn().commit()
@classmethod
def get_resume_position(cls, video_id: str) -> float:
with cls._lock:
row = cls._conn().execute(
"SELECT position, duration, completed FROM watch_history WHERE video_id = ?", (str(video_id),)
).fetchone()
if row is None or row["completed"]:
return 0.0
position = row["position"] or 0.0
duration = row["duration"] or 0.0
if position < 30 or (duration and position >= duration * 0.95):
return 0.0
return float(position)
@classmethod
def watch_history(cls, limit: int = 200) -> List[Dict[str, Any]]:
with cls._lock:
rows = cls._conn().execute(
"SELECT * FROM watch_history ORDER BY watched_at DESC LIMIT ?", (limit,)
).fetchall()
return [dict(r) for r in rows]
# ----------------------------------------------------------- play queue
@classmethod
def save_queue(cls, entries: List[Dict[str, Any]]) -> None:
now = int(time.time())
with cls._lock:
conn = cls._conn()
conn.execute("DELETE FROM play_queue")
for i, e in enumerate(entries or []):
url = e.get("url") or e.get("webpage_url")
if not url:
continue
conn.execute(
"INSERT INTO play_queue (video_id, url, title, channel, duration, thumbnail_url, sort_order, added_at) "
"VALUES (?,?,?,?,?,?,?,?)",
(e.get("id"), url, e.get("title"), e.get("channel") or e.get("uploader"),
e.get("duration"), e.get("thumbnail"), i, now),
)
conn.commit()
@classmethod
def load_queue(cls) -> List[Dict[str, Any]]:
with cls._lock:
rows = cls._conn().execute("SELECT * FROM play_queue ORDER BY sort_order").fetchall()
return [
{"id": r["video_id"], "url": r["url"], "title": r["title"], "channel": r["channel"],
"duration": r["duration"], "thumbnail": r["thumbnail_url"]}
for r in rows
]
Reference in New Issue View Git Blame Copy Permalink