# 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: # Must return fast — Kodi's player event dispatch is serialized and # blocking here while pv.youtube is mid-resolve can drop the next # play. The actual segment fetch + skip loop runs in a background # thread. 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: return self._stop_event.set() self._current_id = yt_id stop_event = threading.Event() self._stop_event = stop_event self._skip_thread = threading.Thread( target=self._arm_and_skip, args=(yt_id, playing_file, stop_event), daemon=True, ) self._skip_thread.start() def _arm_and_skip( self, yt_id: str, playing_file: str, stop_event: threading.Event, ) -> None: segments = _fetch_sb_segments(yt_id) _log(f"armed {len(segments)} segments for {yt_id} (file={playing_file[:80]})") if not segments or stop_event.is_set(): return self._skip_loop(yt_id, segments, stop_event) 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()