torttube/addon/plugin.video.torttube/service.py

# service.py - Kodi background service for torttube.
# SPDX-License-Identifier: GPL-3.0-or-later
#
# Two jobs:
#  1. First-run install of the bundled plugin.video.youtube zip, so users
#     get one-addon UX. The bundled zip is the upstream release; we just
#     extract it into Kodi's addons dir and force a rescan.
#  2. SponsorBlock for ANY YouTube playback, not just plays initiated by
#     torttube. That includes phone-cast plays where Kodi receives a
#     plugin://plugin.video.youtube/play/... URL directly and our plugin
#     code never runs. The xbmc.Player subclass below observes every play,
#     extracts the YouTube ID from the resolved URL, and runs the skip
#     loop in a thread.

import json
import os
import re
import subprocess
import threading
import time
import zipfile
from typing import Any

import xbmc
import xbmcaddon
import xbmcgui
import xbmcvfs

ADDON = xbmcaddon.Addon()
ADDON_ID = ADDON.getAddonInfo("id")
ADDON_PATH = ADDON.getAddonInfo("path")


def _log(msg: str, level: int = xbmc.LOGINFO) -> None:
    xbmc.log(f"[{ADDON_ID}.service] {msg}", level=level)


# YouTube IDs appear in several URL shapes once pv.youtube + IA finish
# resolving: ?v=XXX (uri2addon), ?video_id=XXX (play handler),
# ?file=XXX.mpd (the localhost MPD server pv.youtube spins up). Match all.
_YT_ID_RE = re.compile(r"(?:[?&](?:v|file|video_id)=)([A-Za-z0-9_-]{11})")


def _extract_yt_id(url: str) -> str | None:
    if not url:
        return None
    m = _YT_ID_RE.search(url)
    return m.group(1) if m else None


def _call_sidecar(payload: dict, timeout_s: float = 10.0) -> dict:
    binary = os.path.join(ADDON_PATH, "bin", "torttube-sidecar")
    proc = subprocess.run(
        [binary],
        input=json.dumps(payload),
        capture_output=True,
        text=True,
        timeout=timeout_s,
    )
    if proc.returncode != 0:
        raise RuntimeError(f"sidecar rc={proc.returncode}: {proc.stderr[:200]}")
    return json.loads(proc.stdout)


def _fetch_sb_segments(yt_id: str) -> list[dict[str, Any]]:
    try:
        resp = _call_sidecar({"op": "sponsorblock", "id": yt_id}, timeout_s=8)
        if resp.get("ok"):
            segs = resp.get("segments") or []
            return [s for s in segs if s.get("actionType") == "skip"]
    except Exception as e:
        _log(f"sponsorblock fetch failed for {yt_id}: {e}", xbmc.LOGWARNING)
    return []


class TorttubePlayerMonitor(xbmc.Player):
    """Watches every playback. On AV-started, pulls the YouTube ID from
    the playing-file URL (works for torttube plays AND phone-cast plays
    that bypass our plugin entirely), fetches SponsorBlock segments, and
    runs a poll-skip loop on a background thread until playback ends."""

    POLL_S = 0.5

    def __init__(self) -> None:
        super().__init__()
        self._lock = threading.Lock()
        self._current_id: str | None = None
        self._skip_thread: threading.Thread | None = None
        self._stop_event = threading.Event()

    def onAVStarted(self) -> None:
        try:
            playing_file = self.getPlayingFile()
        except Exception:
            return
        yt_id = _extract_yt_id(playing_file)
        if not yt_id:
            return
        with self._lock:
            if yt_id == self._current_id:
                # Same video — segments already armed, skip thread alive.
                return
            self._stop_event.set()
            self._current_id = yt_id

        segments = _fetch_sb_segments(yt_id)
        _log(f"armed {len(segments)} segments for {yt_id} (file={playing_file[:80]})")
        if not segments:
            return

        stop_event = threading.Event()
        thread = threading.Thread(
            target=self._skip_loop,
            args=(yt_id, segments, stop_event),
            daemon=True,
        )
        with self._lock:
            self._stop_event = stop_event
            self._skip_thread = thread
        thread.start()

    def onPlayBackStopped(self) -> None:
        self._stop_event.set()
        with self._lock:
            self._current_id = None

    def onPlayBackEnded(self) -> None:
        self.onPlayBackStopped()

    def onPlayBackError(self) -> None:
        self.onPlayBackStopped()

    def _skip_loop(
        self,
        yt_id: str,
        segments: list[dict[str, Any]],
        stop_event: threading.Event,
    ) -> None:
        segs = sorted(segments, key=lambda s: float(s.get("segment", [0, 0])[0]))
        skipped: set[str] = set()
        while not stop_event.is_set():
            try:
                if not self.isPlaying():
                    return
                pos = float(self.getTime())
                # Once we switch to a different video the new monitor will
                # also start; bail this loop if the underlying id changed.
                current_file = self.getPlayingFile()
            except Exception:
                return
            if _extract_yt_id(current_file) != yt_id:
                return
            for seg in segs:
                uuid = seg.get("UUID", "")
                if uuid in skipped:
                    continue
                start = float(seg["segment"][0])
                end = float(seg["segment"][1])
                if pos < start:
                    break
                if pos < end:
                    category = seg.get("category", "?")
                    duration = end - start
                    _log(f"skip {yt_id} {category} {start:.1f}-{end:.1f} ({duration:.0f}s)")
                    try:
                        self.seekTime(end)
                    except Exception as e:
                        _log(f"seekTime failed: {e}", xbmc.LOGWARNING)
                    skipped.add(uuid)
                    try:
                        xbmcgui.Dialog().notification(
                            "SponsorBlock",
                            f"Skipped {category} ({duration:.0f}s)",
                            xbmcgui.NOTIFICATION_INFO,
                            2500,
                        )
                    except Exception:
                        pass
                    break
            if stop_event.wait(self.POLL_S):
                return


def _ensure_pv_youtube() -> None:
    addons_dir = xbmcvfs.translatePath("special://home/addons/")
    target = os.path.join(addons_dir, "plugin.video.youtube")
    if os.path.isdir(target):
        return

    bundled = os.path.join(ADDON_PATH, "vendored", "plugin.video.youtube-7.4.3.zip")
    if not os.path.isfile(bundled):
        _log(f"bundled pv.youtube zip missing at {bundled}", xbmc.LOGERROR)
        return

    _log("first-run: installing bundled pv.youtube 7.4.3")
    try:
        with zipfile.ZipFile(bundled) as zf:
            zf.extractall(addons_dir)
    except Exception as e:
        _log(f"pv.youtube extract failed: {e}", xbmc.LOGERROR)
        return

    xbmc.executebuiltin("UpdateLocalAddons")
    time.sleep(1.5)

    enable_cmd = json.dumps({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "Addons.SetAddonEnabled",
        "params": {"addonid": "plugin.video.youtube", "enabled": True},
    })
    try:
        xbmc.executeJSONRPC(enable_cmd)
    except Exception as e:
        _log(f"failed to enable pv.youtube: {e}", xbmc.LOGWARNING)

    try:
        xbmcgui.Dialog().notification(
            "torttube",
            "Installed YouTube player addon",
            xbmcgui.NOTIFICATION_INFO,
            5000,
        )
    except Exception:
        pass
    _log("pv.youtube install complete")


def main() -> None:
    _log("service starting")
    try:
        _ensure_pv_youtube()
    except Exception as e:
        _log(f"_ensure_pv_youtube failed (non-fatal): {e}", xbmc.LOGWARNING)

    player_monitor = TorttubePlayerMonitor()
    abort = xbmc.Monitor()
    # The Player instance only stays alive while there's a reference to it;
    # keep the reference here for the lifetime of the service.
    _ = player_monitor
    while not abort.abortRequested():
        if abort.waitForAbort(2.0):
            break
    _log("service stopping")


if __name__ == "__main__":
    main()