Add unified channel feed

channel_config.py

@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+_CHANNEL_ID_PATTERN = re.compile(r"(?:https?://)?(?:www\.)?youtube\.com/channel/([^/?#]+)")
+_HANDLE_PATTERN = re.compile(r"(?:https?://)?(?:www\.)?youtube\.com/@([^/?#]+)")
+
+
+def _strip_quotes(value: str) -> str:
+    if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
+        return value[1:-1]
+    return value
+
+
+def _parse_yaml_channels(text: str) -> List[Dict[str, str]]:
+    channels: List[Dict[str, str]] = []
+    current: Dict[str, str] = {}
+
+    for raw_line in text.splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("#"):
+            continue
+        if line == "channels:":
+            continue
+        if line.startswith("- "):
+            if current:
+                channels.append(current)
+            current = {}
+            line = line[2:].strip()
+        if not line:
+            continue
+        if ":" not in line:
+            continue
+        key, value = line.split(":", 1)
+        current[key.strip()] = _strip_quotes(value.strip())
+
+    if current:
+        channels.append(current)
+    return channels
+
+
+def _extract_from_url(url: str) -> Dict[str, Optional[str]]:
+    channel_id = None
+    handle = None
+
+    channel_match = _CHANNEL_ID_PATTERN.search(url)
+    if channel_match:
+        channel_id = channel_match.group(1)
+
+    handle_match = _HANDLE_PATTERN.search(url)
+    if handle_match:
+        handle = handle_match.group(1)
+
+    return {"id": channel_id, "handle": handle}
+
+
+def _normalize_handle(handle: Optional[str]) -> Optional[str]:
+    if not handle:
+        return None
+    return handle.lstrip("@").strip() or None
+
+
+def _parse_bool(value: Optional[object]) -> Optional[bool]:
+    if isinstance(value, bool):
+        return value
+    if value is None:
+        return None
+    text = str(value).strip().lower()
+    if text in {"1", "true", "yes", "y"}:
+        return True
+    if text in {"0", "false", "no", "n"}:
+        return False
+    return None
+
+
+def _normalize_entry(entry: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    channel_id = entry.get("id") or entry.get("channel_id")
+    handle = _normalize_handle(entry.get("handle") or entry.get("username"))
+    url = entry.get("url")
+    name = entry.get("name")
+    rss_flag = _parse_bool(
+        entry.get("rss_enabled") or entry.get("rss") or entry.get("include_in_feed")
+    )
+
+    if url:
+        extracted = _extract_from_url(url)
+        channel_id = channel_id or extracted.get("id")
+        handle = handle or extracted.get("handle")
+
+    if not url:
+        if channel_id:
+            url = f"https://www.youtube.com/channel/{channel_id}"
+        elif handle:
+            url = f"https://www.youtube.com/@{handle}"
+
+    if not name:
+        name = handle or channel_id
+
+    if not name or not url:
+        return None
+
+    normalized = {
+        "id": channel_id or "",
+        "handle": handle or "",
+        "name": name,
+        "url": url,
+        "rss_enabled": True if rss_flag is None else rss_flag,
+    }
+    return normalized
+
+
+def load_channel_entries(path: Path) -> List[Dict[str, str]]:
+    if not path.exists():
+        raise FileNotFoundError(path)
+
+    if path.suffix.lower() == ".json":
+        payload = json.loads(path.read_text(encoding="utf-8"))
+        if isinstance(payload, dict):
+            raw_entries = payload.get("channels", [])
+        else:
+            raw_entries = payload
+    else:
+        raw_entries = _parse_yaml_channels(path.read_text(encoding="utf-8"))
+
+    entries: List[Dict[str, str]] = []
+    for raw in raw_entries:
+        if not isinstance(raw, dict):
+            continue
+        raw_payload: Dict[str, Any] = {}
+        for key, value in raw.items():
+            if value is None:
+                continue
+            if isinstance(value, bool):
+                raw_payload[str(key).strip()] = value
+            else:
+                raw_payload[str(key).strip()] = str(value).strip()
+        normalized = _normalize_entry(raw_payload)
+        if normalized:
+            entries.append(normalized)
+
+    entries.sort(key=lambda item: item["name"].lower())
+    return entries
+
+
+def build_rss_bridge_url(entry: Dict[str, str], rss_bridge_host: str = "rss-bridge") -> Optional[str]:
+    channel_id = entry.get("id") or ""
+    handle = _normalize_handle(entry.get("handle"))
+
+    if channel_id:
+        return (
+            f"http://{rss_bridge_host}/?action=display&bridge=YoutubeBridge"
+            f"&context=By+channel+id&c={channel_id}&format=Mrss"
+        )
+    if handle:
+        return (
+            f"http://{rss_bridge_host}/?action=display&bridge=YoutubeBridge"
+            f"&context=By+username&u={handle}&format=Mrss"
+        )
+    return None