youtube channels: per-channel overrides (custom show-name + quality) — core

storage + wiring + API for the channel settings modal (UI next):
- db.get/set_channel_settings(channel_id): per-channel {custom_name, quality}
  in the settings KV store (no schema change); blanks clear the override.
- enqueue applies them: custom_name overrides the $channel folder token; a
  quality override is stashed in the download row's search_ctx. the worker reads
  it back (quality_override_from_download) and uses it instead of the global
  youtube quality profile. global default still applies when no override.
- API GET/POST /youtube/channel/<id>/settings (GET also returns the global
  default quality for the modal's 'using default' hint).

pure helpers (enqueue_ctx, quality_override_from_download) + db + api seam-tested.

api/video/youtube.py

@@ -187,6 +187,34 @@ def register_routes(bp):
             logger.exception("youtube channels list failed")
             return jsonify({"success": False, "error": "Failed"}), 500
 
+    @bp.route("/youtube/channel/<channel_id>/settings", methods=["GET"])
+    def video_youtube_channel_settings(channel_id):
+        """Per-channel overrides (custom show-name + quality), plus the global quality
+        default so the modal can show 'using default' until the user overrides."""
+        from . import get_video_db
+        from core.video.youtube_quality import default_profile, load as load_quality
+        db = get_video_db()
+        return jsonify({"success": True, "settings": db.get_channel_settings(channel_id) or {},
+                        "default_quality": load_quality(db) or default_profile()})
+
+    @bp.route("/youtube/channel/<channel_id>/settings", methods=["POST"])
+    def video_youtube_channel_settings_save(channel_id):
+        """Save per-channel overrides. A blank custom_name / absent quality clears that
+        override (falls back to the channel title / global quality). Body:
+        {custom_name?, quality?{...}}."""
+        from . import get_video_db
+        from core.video.youtube_quality import normalize as normalize_quality
+        db = get_video_db()
+        body = request.get_json(silent=True) or {}
+        out = {}
+        name = str(body.get("custom_name") or "").strip()
+        if name:
+            out["custom_name"] = name
+        if body.get("quality"):       # falsy/absent → no override (use the global default)
+            out["quality"] = normalize_quality(body.get("quality"))
+        db.set_channel_settings(channel_id, out)
+        return jsonify({"success": True, "settings": out})
+
     @bp.route("/youtube/wishlist", methods=["GET"])
     def video_youtube_wishlist():
         """Wished videos grouped by channel (channel = group, videos = feed)."""

core/automation/handlers/video_process_youtube_wishlist.py

@@ -44,6 +44,20 @@ def slots_free(running: int, max_concurrent: int) -> int:
     return max(0, int(max_concurrent) - max(0, int(running)))
 
 
+def enqueue_ctx(video: Dict[str, Any], channel_settings: Dict[str, Any]) -> Dict[str, Any]:
+    """The download row's ``search_ctx``, applying per-channel overrides: a custom show-name
+    (the ``$channel`` folder token) and/or a quality override. Pure."""
+    cs = channel_settings if isinstance(channel_settings, dict) else {}
+    ctx = {
+        "channel": cs.get("custom_name") or video.get("channel_title"),
+        "video_title": video.get("video_title"),
+        "published_at": video.get("published_at"),
+    }
+    if cs.get("quality"):
+        ctx["quality"] = cs["quality"]
+    return ctx
+
+
 # ── production seams ──────────────────────────────────────────────────────────
 def _default_youtube_root() -> str:
     from api.video import get_video_db
@@ -72,15 +86,15 @@ def _default_enqueue(video: Dict[str, Any], root: str) -> Any:
     import json
     from api.video import get_video_db
     from core.video.sources import resolve_video_server
-    return get_video_db().add_video_download({
+    db = get_video_db()
+    ctx = enqueue_ctx(video, db.get_channel_settings(video.get("channel_id")))
+    ctx["server_source"] = resolve_video_server()
+    return db.add_video_download({
         "kind": "youtube", "source": "youtube", "media_source": "youtube",
         "title": video.get("video_title") or video.get("channel_title"),
         "media_id": video.get("video_id"), "target_dir": root, "status": "queued",
         "year": video.get("published_at"), "poster_url": video.get("thumbnail_url"),
-        "search_ctx": json.dumps({"channel": video.get("channel_title"),
-                                  "video_title": video.get("video_title"),
-                                  "published_at": video.get("published_at"),
-                                  "server_source": resolve_video_server()}),
+        "search_ctx": json.dumps(ctx),
     })
 
 

core/video/youtube_download.py

@@ -60,6 +60,20 @@ def youtube_fields_from_download(dl: Dict[str, Any]) -> Dict[str, Any]:
     }
 
 
+def quality_override_from_download(dl: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    """A per-channel quality override stashed in the row's ``search_ctx`` at enqueue time,
+    or None (use the global YouTube quality profile)."""
+    ctx = dl.get("search_ctx")
+    if isinstance(ctx, str):
+        try:
+            ctx = json.loads(ctx)
+        except (ValueError, TypeError):
+            ctx = {}
+    ctx = ctx if isinstance(ctx, dict) else {}
+    q = ctx.get("quality")
+    return q if isinstance(q, dict) else None
+
+
 def plan_destination(dl: Dict[str, Any], settings: Dict[str, Any], container: str) -> Dict[str, str]:
     """Where this video lands in the library: ``{dir, filename, path}`` under the youtube
     root (``target_dir``), organised by the youtube template. Pure."""
@@ -242,7 +256,9 @@ def run_youtube_download(dl_id: Any, db_provider: Callable) -> None:
         _active_worker_ids.discard(dl_id)
         start_next_queued(db_provider)        # keep the queue moving even on a stale id
         return
-    profile = youtube_quality.load(db)
+    # Per-channel quality override (stashed in search_ctx at enqueue) wins over the global.
+    override = quality_override_from_download(dl)
+    profile = youtube_quality.normalize(override) if override else youtube_quality.load(db)
     settings = organization.load(db)
     from datetime import datetime, timezone
 
@@ -288,5 +304,5 @@ def run_youtube_download(dl_id: Any, db_provider: Callable) -> None:
 __all__ = [
     "youtube_fields_from_download", "plan_destination", "ydl_download_opts",
     "download_one", "process_youtube_download", "run_youtube_download",
-    "start_next_queued", "requeue_orphaned_youtube",
+    "start_next_queued", "requeue_orphaned_youtube", "quality_override_from_download",
 ]

database/video_database.py

@@ -3071,6 +3071,34 @@ class VideoDatabase:
         finally:
             conn.close()
 
+    def get_channel_settings(self, channel_id) -> dict:
+        """Per-channel YouTube overrides — ``{custom_name?, quality?}`` — or {} if none.
+        ``custom_name`` overrides the show-name (the ``$channel`` folder token); ``quality``
+        is a youtube_quality profile that forces a different quality than the global default
+        for this channel. Kept in the settings KV store, so no schema change."""
+        if not channel_id:
+            return {}
+        raw = self.get_setting("youtube_channel_settings:" + str(channel_id))
+        if not raw:
+            return {}
+        try:
+            import json
+            d = json.loads(raw)
+            return d if isinstance(d, dict) else {}
+        except (ValueError, TypeError):
+            return {}
+
+    def set_channel_settings(self, channel_id, settings: dict) -> bool:
+        """Persist per-channel overrides (or clear them with an empty/blank dict)."""
+        if not channel_id:
+            return False
+        import json
+        clean = settings if isinstance(settings, dict) else {}
+        # Drop empty values so a blank form clears the override rather than storing noise.
+        clean = {k: v for k, v in clean.items() if v not in (None, "", {})}
+        self.set_setting("youtube_channel_settings:" + str(channel_id), json.dumps(clean))
+        return True
+
     def wishlisted_video_ids_for_channel(self, channel_id) -> list:
         """The youtube video ids wished under a channel (the per-video date fallback set)."""
         if not channel_id:

tests/test_youtube_channel_settings.py

@@ -0,0 +1,93 @@
+"""Per-channel YouTube overrides: a custom show-name (the $channel folder token) and a
+quality override. Storage (KV), the enqueue/worker wiring that applies them, and the API.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from core.automation.handlers.video_process_youtube_wishlist import enqueue_ctx
+from core.video.youtube_download import quality_override_from_download
+from database.video_database import VideoDatabase
+
+
+@pytest.fixture()
+def db(tmp_path):
+    return VideoDatabase(database_path=str(tmp_path / "video_library.db"))
+
+
+# ── storage ───────────────────────────────────────────────────────────────────
+def test_channel_settings_roundtrip(db):
+    assert db.get_channel_settings("UC1") == {}                      # none yet
+    db.set_channel_settings("UC1", {"custom_name": "My Show", "quality": {"max_resolution": "720p"}})
+    cs = db.get_channel_settings("UC1")
+    assert cs["custom_name"] == "My Show" and cs["quality"]["max_resolution"] == "720p"
+
+
+def test_channel_settings_blank_clears(db):
+    db.set_channel_settings("UC1", {"custom_name": "X"})
+    db.set_channel_settings("UC1", {"custom_name": "", "quality": None})   # blanks dropped
+    assert db.get_channel_settings("UC1") == {}
+
+
+def test_channel_settings_isolated_per_channel(db):
+    db.set_channel_settings("UC1", {"custom_name": "One"})
+    assert db.get_channel_settings("UC2") == {}
+
+
+# ── enqueue applies the overrides into search_ctx ─────────────────────────────
+def _video():
+    return {"video_id": "v1", "channel_id": "UC1", "channel_title": "Real Channel Name",
+            "video_title": "Ep 1", "published_at": "2024-03-15"}
+
+
+def test_enqueue_ctx_uses_channel_title_when_no_override():
+    ctx = enqueue_ctx(_video(), {})
+    assert ctx["channel"] == "Real Channel Name" and "quality" not in ctx
+
+
+def test_enqueue_ctx_applies_custom_name_and_quality():
+    ctx = enqueue_ctx(_video(), {"custom_name": "Custom Show", "quality": {"max_resolution": "4320p"}})
+    assert ctx["channel"] == "Custom Show"                       # overrides the $channel token
+    assert ctx["quality"] == {"max_resolution": "4320p"}
+    assert ctx["video_title"] == "Ep 1" and ctx["published_at"] == "2024-03-15"
+
+
+# ── worker reads the quality override back ────────────────────────────────────
+def test_quality_override_from_download_reads_search_ctx():
+    dl = {"search_ctx": json.dumps({"channel": "X", "quality": {"max_resolution": "720p"}})}
+    assert quality_override_from_download(dl) == {"max_resolution": "720p"}
+
+
+def test_quality_override_absent_returns_none():
+    assert quality_override_from_download({"search_ctx": json.dumps({"channel": "X"})}) is None
+    assert quality_override_from_download({"search_ctx": "{bad"}) is None
+    assert quality_override_from_download({}) is None
+
+
+# ── API ───────────────────────────────────────────────────────────────────────
+def test_channel_settings_api_roundtrip(tmp_path):
+    from flask import Flask
+    import api.video as videoapi
+    db = VideoDatabase(database_path=str(tmp_path / "video_library.db"))
+    videoapi._video_db = db
+    app = Flask(__name__)
+    app.register_blueprint(videoapi.create_video_blueprint(), url_prefix="/api/video")
+    client = app.test_client()
+    try:
+        r = client.post("/api/video/youtube/channel/UC1/settings",
+                        json={"custom_name": "My Show", "quality": {"max_resolution": "720p"}}).get_json()
+        assert r["success"] and r["settings"]["custom_name"] == "My Show"
+        assert r["settings"]["quality"]["max_resolution"] == "720p"   # normalized profile
+
+        g = client.get("/api/video/youtube/channel/UC1/settings").get_json()
+        assert g["settings"]["custom_name"] == "My Show"
+        assert "default_quality" in g                                 # for the 'using default' hint
+
+        # blank custom_name + no quality clears the override
+        client.post("/api/video/youtube/channel/UC1/settings", json={"custom_name": "", "quality": None})
+        assert client.get("/api/video/youtube/channel/UC1/settings").get_json()["settings"] == {}
+    finally:
+        videoapi._video_db = None