torttube/addon/plugin.video.torttube/main.py

# torttube — Kodi YouTube addon
# SPDX-License-Identifier: GPL-3.0-or-later
"""
Entry point. Plugin URL shapes handled:

  plugin://plugin.video.torttube/                       (root: M3 will add browse UI)
  plugin://plugin.video.torttube/?action=play&id=ID     (resolve + play a YouTube ID)
  plugin://plugin.video.torttube/?action=play&url=URL   (extract ID from URL, then play)

Remote-control / share-to-TV pattern (Kodi JSON-RPC):

  POST http://<rpi>:8080/jsonrpc
  Basic auth (kodi user)
  {"jsonrpc":"2.0","method":"Player.Open","params":{
    "item":{"file":"plugin://plugin.video.torttube/?action=play&id=<id>"}}}

That's how Android / phone / "send to TV" flows hand off — Kodi already
exposes the endpoint, we just need to register the plugin URL.
"""

import json
import os
import re
import subprocess
import sys
from urllib.parse import parse_qsl, urlparse

import xbmc
import xbmcaddon
import xbmcgui
import xbmcplugin

ADDON = xbmcaddon.Addon()
ADDON_PATH = ADDON.getAddonInfo("path")
SIDECAR_BIN = os.path.join(ADDON_PATH, "bin", "torttube-sidecar")

_HANDLE = int(sys.argv[1]) if len(sys.argv) > 1 else -1
_QS = sys.argv[2] if len(sys.argv) > 2 else ""


def _log(msg, level=xbmc.LOGINFO):
    xbmc.log(f"[torttube] {msg}", level)


def _extract_id(url_or_id: str) -> str:
    """Accept either a bare ID or any common YouTube URL form, return the ID."""
    s = url_or_id.strip()
    # Bare 11-char ID (YouTube's canonical length).
    if re.fullmatch(r"[A-Za-z0-9_-]{11}", s):
        return s
    parsed = urlparse(s)
    # https://youtu.be/<id>
    if parsed.netloc.endswith("youtu.be"):
        return parsed.path.lstrip("/").split("/")[0]
    # https://www.youtube.com/watch?v=<id>
    if "youtube.com" in parsed.netloc:
        for k, v in parse_qsl(parsed.query):
            if k == "v":
                return v
        # /shorts/<id>, /embed/<id>, /live/<id>
        m = re.match(r"^/(shorts|embed|live)/([A-Za-z0-9_-]{11})", parsed.path)
        if m:
            return m.group(2)
    raise ValueError(f"could not extract YouTube id from {url_or_id!r}")


def _call_sidecar(request: dict, timeout_s: int = 30) -> dict:
    """Invoke the sidecar with one JSON request, parse one JSON response."""
    if not os.path.isfile(SIDECAR_BIN):
        raise RuntimeError(f"sidecar binary missing at {SIDECAR_BIN}")
    if not os.access(SIDECAR_BIN, os.X_OK):
        raise RuntimeError(f"sidecar binary not executable at {SIDECAR_BIN}")

    proc = subprocess.run(
        [SIDECAR_BIN],
        input=(json.dumps(request) + "\n").encode("utf-8"),
        capture_output=True,
        timeout=timeout_s,
    )
    if proc.returncode != 0:
        raise RuntimeError(
            f"sidecar exited {proc.returncode}: {proc.stderr.decode('utf-8', 'replace')[:500]}"
        )
    # Stdout may have multiple lines if the sidecar logged something; take the first
    # non-empty line as the response.
    for line in proc.stdout.splitlines():
        line = line.strip()
        if line:
            return json.loads(line.decode("utf-8") if isinstance(line, bytes) else line)
    raise RuntimeError("sidecar produced no response")


def _resolved_listitem(stream_url: str, title: str | None) -> xbmcgui.ListItem:
    li = xbmcgui.ListItem(label=title or "torttube")
    li.setPath(stream_url)
    li.setProperty("IsPlayable", "true")
    # Tell inputstream.adaptive to handle DASH/HLS if the URL looks like a manifest.
    if stream_url.endswith(".mpd"):
        li.setProperty("inputstream", "inputstream.adaptive")
        li.setProperty("inputstream.adaptive.manifest_type", "mpd")
    elif stream_url.endswith(".m3u8"):
        li.setProperty("inputstream", "inputstream.adaptive")
        li.setProperty("inputstream.adaptive.manifest_type", "hls")
    return li


def _play(yt_id: str) -> None:
    """Resolve via sidecar (yt-dlp combined-format path), hand URL to Kodi."""
    _log(f"play id={yt_id}")
    try:
        # resolve_play returns ONE combined audio+video URL — guaranteed
        # to play in Kodi without needing inputstream.adaptive/DASH.
        # ~3-5s overhead vs rustypipe but reliable audio sync.
        resp = _call_sidecar({"op": "resolve_play", "id": yt_id}, timeout_s=45)
    except Exception as e:
        _log(f"sidecar resolve_play failed: {e}", xbmc.LOGERROR)
        xbmcgui.Dialog().notification(
            "torttube", f"resolve failed: {e}", xbmcgui.NOTIFICATION_ERROR, 5000
        )
        xbmcplugin.setResolvedUrl(_HANDLE, False, xbmcgui.ListItem())
        return

    if not resp.get("ok"):
        kind = resp.get("kind", "unknown")
        err = resp.get("error", "unknown")
        _log(f"sidecar returned not-ok: {kind}: {err}", xbmc.LOGWARNING)
        xbmcgui.Dialog().notification(
            "torttube", f"{kind}: {err}", xbmcgui.NOTIFICATION_WARNING, 5000
        )
        xbmcplugin.setResolvedUrl(_HANDLE, False, xbmcgui.ListItem())
        return

    stream_url = resp.get("stream_url")
    title = resp.get("title")
    if not stream_url:
        _log("no usable stream URL in sidecar response", xbmc.LOGERROR)
        xbmcgui.Dialog().notification(
            "torttube", "no stream URL", xbmcgui.NOTIFICATION_ERROR, 5000
        )
        xbmcplugin.setResolvedUrl(_HANDLE, False, xbmcgui.ListItem())
        return

    _log(f"resolved via {resp.get('source')}, playing")
    xbmcplugin.setResolvedUrl(_HANDLE, True, _resolved_listitem(stream_url, title))


def _root_directory() -> None:
    """M0/M3 placeholder root menu — gets replaced by real browse UI in M4."""
    xbmcgui.Dialog().notification(
        "torttube",
        "M3 — send a play URL via JSON-RPC. Browse UI lands in M4.",
        xbmcgui.NOTIFICATION_INFO,
        3000,
    )
    xbmcplugin.endOfDirectory(_HANDLE)


def main() -> None:
    params = dict(parse_qsl(_QS.lstrip("?")))
    action = params.get("action")

    if action == "play":
        try:
            yt_id = _extract_id(params.get("url") or params.get("id") or "")
        except ValueError as e:
            _log(str(e), xbmc.LOGERROR)
            xbmcgui.Dialog().notification(
                "torttube", str(e), xbmcgui.NOTIFICATION_ERROR, 4000
            )
            xbmcplugin.setResolvedUrl(_HANDLE, False, xbmcgui.ListItem())
            return
        _play(yt_id)
    else:
        _root_directory()


if __name__ == "__main__":
    main()