# coding: utf-8
"""
Background service for plugin.video.glotzbox


Goals:
- Periodically check remote JSONs (movies, series, favorites) on HTTPS.
- Use conditional GET (ETag / Last-Modified).
- Only trigger refresh when something really changed.
- Stay robust: never crash Kodi, never require a valid addon_handle.
- NEW: Orphan Artwork Cache Cleanup on start + max 1x/24h

Notes:
- Auth happens via common.get_authenticated_request().
- Service usually has NO addon_handle (sys.argv has no handle) -> handle stays -1.
"""
from typing import Optional, Dict, Any, Tuple, List

import os
import sys
import json
import time
import ssl
import hashlib
import traceback
import urllib.request
import urllib.error
import urllib.parse
import re
import threading

import xbmc
import xbmcaddon
import xbmcgui
import xbmcvfs

import gb_common as common

# ---------------------------------------------------------------------------
# Prefetch throttles
#
# Kodi starts the service on every boot. To avoid hammering the server we
# throttle the *details.json* refresh loop. You asked to keep it aggressive,
# so we default to 1 hour.
# ---------------------------------------------------------------------------

# How often the service is allowed to run the series-details prefetch when the
# series list itself is unchanged.
SERIES_PREFETCH_INTERVAL_HOURS = 1  # hours

# Where we store the per-series details.json cache (one file per series) and
# the meta information used for conditional requests (ETag/Last-Modified).
# NOTE: Must match series.py (v18+) naming.
SERIES_DETAILS_CACHE_DIR = os.path.join(common.addon_data_dir, "cache_series_details_json")
# Artwork/Poster Cache (gleiches Verzeichnis wie in series.py)
SERIES_IMAGES_CACHE_DIR = os.path.join(common.addon_data_dir, "cache_series_images")
SERIES_DETAILS_META_DIR  = os.path.join(common.addon_data_dir, "cache_series_details_json_meta")
LEGACY_SERIES_DETAILS_CACHE_DIR = os.path.join(common.addon_data_dir, "series_details_cache")
SERIES_PAYLOAD_CACHE_DIR = os.path.join(common.addon_data_dir, "payload_cache")
SERIES_SEEN_URLS_FILE = os.path.join(common.addon_data_dir, "series_seen_urls.json")
SERIES_DETAILS_META_FILE = os.path.join(common.addon_data_dir, "series_details_meta.json")

# State file for throttling + book-keeping (per installation)
SERIES_PREFETCH_STATE_FILE = os.path.join(common.addon_data_dir, "series_details_prefetch_state.json")

# Max number of details.json files to fetch per run (you asked to fetch all; 200 covers all typical lists)
SERIES_PREFETCH_LIMIT_PER_RUN = 200

# Max artwork URLs to prefetch per run (poster/fanart/stills)
SERIES_ARTWORK_PREFETCH_LIMIT = 400


# Window property to prevent multiple concurrent prefetch runs
SERIES_PREFETCH_PROP = f"{common.addon_id}.series.details.prefetch.running"
SERIES_PREFETCH_TRIGGER_PROP = f"{common.addon_id}.series.details.prefetch.trigger"
# Trigger so movies.py can signal service.py to run a fresh image prefetch mid-cycle.
MOVIES_IMAGE_PREFETCH_TRIGGER_PROP = f"{common.addon_id}.movies.image.prefetch.trigger"

# When the service updates details.json on disk, bump this timestamp so series.py can drop stale in-memory caches.
DETAILS_BUMP_PROP = f"{common.addon_id}.series.details.bump"

def _bump_details_cache(reason=""):
    """Notify series.py to drop any in-memory details cache after background refresh."""
    try:
        win = xbmcgui.Window(10000)
        ts = str(int(time.time()))
        win.setProperty(DETAILS_BUMP_PROP, ts)
        if reason:
            log(f"Details-Bump -> {ts} ({reason})", xbmc.LOGDEBUG)
    except Exception:
        pass


LOGP = f"[{common.addon_id} service.py]"
SERVICE_BUILD = 'details-prefetch-v2'


def log(msg, lvl=xbmc.LOGINFO):
    try:
        xbmc.log(f"{LOGP} {msg}", lvl)
    except Exception:
        pass


# ---------------------------------------------------------------------
# Addon handle safe getter (Services haben normalerweise KEIN Handle)
# ---------------------------------------------------------------------
def get_addon_handle_safe(default=-1):
    argv = getattr(sys, "argv", []) or []
    if len(argv) > 1:
        try:
            h = int(argv[1])
            if h >= 0:
                return h
        except Exception:
            pass
    return default


# ---------------------------------------------------------------------
# Settings helpers
# ---------------------------------------------------------------------
def _get_setting_str(key, default=""):
    try:
        val = common.addon.getSetting(key)
        return val if isinstance(val, str) else default
    except Exception:
        return default


def _get_setting_int(key, default=0):
    try:
        raw = common.addon.getSetting(key)
        return int(raw)
    except Exception:
        return default


def _get_interval_seconds(default=3600):
    for k in ("service_interval", "bg_interval", "background_refresh_interval"):
        v = _get_setting_int(k, 0)
        if v and v > 0:
            return v
    return default


NET_TIMEOUT = max(5, _get_setting_int("net_timeout", 20))


# ---------------------------------------------------------------------
# SSL + urlopen wrapper
# ---------------------------------------------------------------------
try:
    _SSL_CTX = common.get_ssl_context()
except Exception:
    try:
        _SSL_CTX = ssl.create_default_context()
    except Exception:
        _SSL_CTX = None



# Startup refresh behavior
# Series + Movies werden beim Start und im Loop via sha1-Sidecar auf Änderungen geprüft.
# HEAD-Probes sind deaktiviert – sha1-Sidecar-Vergleich ist ausreichend und günstiger.

def _robust_urlopen(req, timeout=NET_TIMEOUT):
    if _SSL_CTX:
        return urllib.request.urlopen(req, context=_SSL_CTX, timeout=timeout)
    return urllib.request.urlopen(req, timeout=timeout)


# ---------------------------------------------------------------------
# Only refresh UI when our addon container is visible
# ---------------------------------------------------------------------

SERVICE_START_TS = int(time.time())


def _refresh_if_addon_visible():
    """Refresh nur dann auslösen, wenn wir im Glotzbox-Container sind.

    Wichtig:
    - Keine Refreshes während Fullscreen/Playback.
    - Refresh nur im Root bzw. in den Top-Level-Listen (Filme/Serien/Favoriten),
      damit Navigation (Staffeln/Episoden) nicht „zurückspringt“.
    """
    try:
        try:
            if xbmc.getCondVisibility("Window.IsVisible(fullscreenvideo)"):
                return
        except Exception:
            pass
        try:
            if xbmc.getCondVisibility("Player.HasMedia"):
                return
        except Exception:
            pass

        path = xbmc.getInfoLabel("Container.FolderPath") or ""
        p = path.lower()
        base = f"plugin://{common.addon_id}".lower()
        if not p.startswith(base):
            return

        # Root (Hauptmenü)
        if p.rstrip("/") in (base, base + "/"):
            xbmc.executebuiltin("Container.Refresh")
            return

        # Nur Top-Level Listen refreshen
        allow = ("action=show_serien", "action=show_filme", "action=show_favorites")
        suppress = (
            "action=show_seasons",
            "action=show_episodes",
            "action=show_fav_seasons",
            "action=show_fav_episodes",
            "action=play_",
        )

        if any(a in p for a in allow) and not any(s in p for s in suppress):
            xbmc.executebuiltin("Container.Refresh")
        else:
            log(f"UI Refresh suppressed (not top-level): {path}", xbmc.LOGDEBUG)
    except Exception:
        pass


# ---------------------------------------------------------------------
# Remote host/path normalization
# ---------------------------------------------------------------------
def _normalize_host_and_basepath():
    """Use the shared normalization helper, but keep the historic return order.

    Returns: (scheme, host, base_path)
    """
    try:
        host, base_path, scheme = common.normalize_ftp_host_basepath()
        return scheme, host, base_path
    except Exception:
        return "https", "", "/user/downloads/kodiAddon/"


# ---------------------------------------------------------------------
# Meta I/O
# ---------------------------------------------------------------------
def _meta_path(filename):
    return os.path.join(common.addon_data_dir, f"{filename}.meta.json")


def _load_meta(path):
    try:
        if not xbmcvfs.exists(path):
            return {}
        with xbmcvfs.File(path, "r") as fh:
            raw = fh.read()
        j = json.loads(raw) if raw else {}
        return j if isinstance(j, dict) else {}
    except Exception:
        return {}


def _save_meta(path, headers=None, size_bytes=None, sha1=None):
    try:
        meta = _load_meta(path)
        headers = headers or {}

        # Content-Length ist optional, aber hilfreich für schnelle HEAD-Checks.
        cl = None
        try:
            cl_raw = headers.get("Content-Length") or headers.get("content-length")
            if cl_raw is not None:
                cl_s = str(cl_raw).strip()
                if cl_s.isdigit():
                    cl = int(cl_s)
        except Exception:
            cl = None

        meta.update({
            "etag": headers.get("ETag", meta.get("etag", "")) or "",
            "last_modified": headers.get("Last-Modified", meta.get("last_modified", "")) or "",
            "content_length": int(cl) if isinstance(cl, int) and cl > 0 else int(meta.get("content_length", 0) or 0),
            "size": int(size_bytes if size_bytes is not None else meta.get("size", 0) or 0),
            "sha1": sha1 if sha1 is not None else (meta.get("sha1", "") or ""),
            "fetched": int(time.time())
        })

        # Falls wir size haben, aber content_length noch 0 ist -> sinnvoll befuellen.
        try:
            if not meta.get("content_length") and meta.get("size"):
                meta["content_length"] = int(meta.get("size") or 0)
        except Exception:
            pass

        with xbmcvfs.File(path, "w") as fh:
            fh.write(json.dumps(meta, ensure_ascii=False))
        return True
    except Exception:
        return False


# ---------------------------------------------------------------------
# Conditional fetch
# ---------------------------------------------------------------------
def _conditional_fetch(url, meta_path, use_conditional=True):
    """
    Returns tuple: (status, data_bytes, headers)
      status: "not_modified" | "changed" | "error"
    """
    meta = _load_meta(meta_path)
    req = common.get_authenticated_request(url)
    if not req:
        return "error", None, None

    # KodiNerds/NGINX setups sometimes keep ETag/Last-Modified stable even if the
    # underlying JSON changes (e.g. regenerated file with same validators).
    # For critical payloads (movies), we can disable conditional headers and rely
    # purely on SHA1 comparison to detect updates.
    if use_conditional:
        # Optional conditional headers.
        # Some servers/proxies can return 304 even when the underlying JSON changed.
        # For such cases (e.g. movies.json on some setups), we allow disabling
        # conditional headers and rely on a lightweight sha1 comparison instead.
        if meta.get("etag"):
            try:
                req.add_header("If-None-Match", meta["etag"])
            except Exception:
                pass
        if meta.get("last_modified"):
            try:
                req.add_header("If-Modified-Since", meta["last_modified"])
            except Exception:
                pass

    try:
        with _robust_urlopen(req, timeout=NET_TIMEOUT) as resp:
            code = resp.getcode()

            if code == 200:
                data_bytes = resp.read() or b""
                body_sha1 = hashlib.sha1(data_bytes).hexdigest()

                if meta.get("sha1") and meta.get("sha1") == body_sha1:
                    _save_meta(meta_path, headers=resp.headers, size_bytes=len(data_bytes), sha1=body_sha1)
                    return "not_modified", None, resp.headers

                _save_meta(meta_path, headers=resp.headers, size_bytes=len(data_bytes), sha1=body_sha1)
                return "changed", data_bytes, resp.headers

            log(f"Conditional fetch unexpected code {code} for {common.mask_url(url)}", xbmc.LOGDEBUG)
            return "error", None, getattr(resp, "headers", None)

    except urllib.error.HTTPError as e:
        if e.code == 304:
            _save_meta(meta_path, headers=e.headers, size_bytes=meta.get("size", 0), sha1=meta.get("sha1", ""))
            return "not_modified", None, e.headers
        log(f"Conditional fetch HTTPError for {common.mask_url(url)}: {e}", xbmc.LOGDEBUG)
        return "error", None, getattr(e, "headers", None)
    except Exception as e:
        log(f"Conditional fetch error for {common.mask_url(url)}: {e}", xbmc.LOGDEBUG)
        return "error", None, None


# ---------------------------------------------------------------------
# Cache JSON write helper
# ---------------------------------------------------------------------

def _head_not_modified(url: str, meta_path: str, timeout: int = 8) -> Optional[bool]:
    """Lightweight HEAD probe with If-None-Match / If-Modified-Since.

    Returns:
      - True  -> remote NOT modified (safe to skip refresh)
      - False -> remote modified (caller should refresh)
      - None  -> probe failed (caller should fall back to refresh)
    """
    meta = _load_meta(meta_path) or {}
    try:
        # Build an authenticated request if the addon provides it (keeps credentials / headers consistent).
        try:
            req = common.get_authenticated_request(url)
        except Exception:
            req = None

        if req is None:
            req = urllib.request.Request(url)

        # Conditional headers from our last successful response.
        etag = meta.get("etag")
        last_mod = meta.get("last_modified")
        if etag:
            try:
                req.add_header("If-None-Match", str(etag))
            except Exception:
                pass
        if last_mod:
            try:
                req.add_header("If-Modified-Since", str(last_mod))
            except Exception:
                pass

        # Force HEAD even if the request helper created a GET.
        try:
            req.get_method = lambda: "HEAD"  # type: ignore
        except Exception:
            pass

        with _robust_urlopen(req, timeout=timeout) as resp:
            # urllib returns 200 for non-304 responses; 304 is raised as HTTPError, but be defensive.
            code = getattr(resp, "status", getattr(resp, "code", 200))
            if int(code) == 304:
                return True

            hdrs = getattr(resp, "headers", None)
            new_etag = None
            new_last_mod = None
            new_len = None
            try:
                if hdrs:
                    new_etag = hdrs.get("ETag") or hdrs.get("Etag")
                    new_last_mod = hdrs.get("Last-Modified")
                    cl = hdrs.get("Content-Length")
                    if cl:
                        try:
                            new_len = int(cl)
                        except Exception:
                            new_len = None
            except Exception:
                pass

            # Decide "not modified" using best available validators.
            not_modified = False
            if etag and new_etag and str(etag) == str(new_etag):
                not_modified = True
            elif last_mod and new_last_mod and str(last_mod) == str(new_last_mod):
                not_modified = True
            elif meta.get("content_length") and new_len and int(meta.get("content_length")) == int(new_len) and not (new_etag or new_last_mod):
                # only length-based when no stronger validators are available
                not_modified = True

            # Persist the latest validators (even when modified) so next start is fast.
            _save_meta(meta_path, headers={k:v for k,v in {
                "ETag": (new_etag or etag),
                "Last-Modified": (new_last_mod or last_mod),
                "Content-Length": (str(new_len) if new_len is not None else None),
            }.items() if v})

            return True if not_modified else False

    except urllib.error.HTTPError as e:
        if int(getattr(e, "code", 0)) == 304:
            return True
        log(f"HEAD probe failed ({e.code}) for {common.mask_url(url)}: {e}", xbmc.LOGWARNING)
        return None
    except Exception as e:
        log(f"HEAD probe exception for {common.mask_url(url)}: {e}", xbmc.LOGWARNING)
        return None

def _save_cache_json(cache_path, data_bytes):
    try:
        text = data_bytes.decode("utf-8-sig")
        data = json.loads(text)
        return bool(common.atomic_write_json(cache_path, data, indent=None, durable=True, backup=True))
    except Exception as e:
        log(f"Cache write failed {cache_path}: {e}", xbmc.LOGWARNING)
        return False


# ---------------------------------------------------------------------
# Module refresh wrappers (if present)
# ---------------------------------------------------------------------
_MOVIES_REFRESH = None
_SERIES_REFRESH = None
_FAV_REFRESH = None
series = None

try:
    import movies
    _MOVIES_REFRESH = getattr(movies, "background_refresh_movies_once", None)
except Exception as e:
    log(f"movies import failed: {e}", xbmc.LOGDEBUG)

try:
    import series
    _SERIES_REFRESH = getattr(series, "background_refresh_series_once", None)
except Exception as e:
    log(f"series import failed: {e}", xbmc.LOGDEBUG)

try:
    import favorites
    if hasattr(favorites, "load_favorites_json"):
        def _fav_refresh_wrapper():
            data, err = favorites.load_favorites_json()
            return bool(data) and not err
        _FAV_REFRESH = _fav_refresh_wrapper
except Exception as e:
    log(f"favorites import failed: {e}", xbmc.LOGDEBUG)



# ---------------------------------------------------------------------
# Details URL key normalization
# - Prefer the implementation from series.py to guarantee identical cache keys.
# - Fallback to local implementation (older installs / import issues).
# ---------------------------------------------------------------------
try:
    _SERIES_DETAILS_URL_KEY = getattr(series, '_details_url_key', None) if series is not None else None
    if _SERIES_DETAILS_URL_KEY is None:
        _SERIES_DETAILS_URL_KEY = getattr(sys.modules.get('series'), '_details_url_key', None)
except Exception:
    _SERIES_DETAILS_URL_KEY = None

# ---------------------------------------------------------------------
# Per-kind refresh logic
# ---------------------------------------------------------------------
MOVIES_JSON_FILENAME = "kodiMovie.json"
SERIES_JSON_FILENAME = "kodiSeries.json"
FAV_JSON_FILENAME    = "seriesFav.json"

MOVIES_CACHE_PATH = os.path.join(common.addon_data_dir, "movies.cache.json")
SERIES_CACHE_PATH = os.path.join(common.addon_data_dir, "series.cache.json")
FAV_CACHE_PATH    = os.path.join(common.addon_data_dir, "seriesFav.cache.json")


def _refresh_kind(kind_name, filename, cache_path, module_refresh_fn=None):
    scheme, host, base_path = _normalize_host_and_basepath()

    if not host:
        log(f"{kind_name}: Host missing -> skip", xbmc.LOGWARNING)
        return False

    url = f"{scheme}://{host}{base_path}{filename}"
    meta_p = _meta_path(filename)

    # NEW: cheap change detection via tiny sidecar hash (e.g. kodiSeries.json.sha1)
    remote_sha = None
    try:
        if scheme.startswith("http"):
            remote_sha = common.fetch_remote_sha1(url, timeout_s=8)
    except Exception:
        remote_sha = None

    # If we have a remote hash and it matches our stored sha1, we can skip without downloading the JSON.
    try:
        meta_now = _load_meta(meta_p)
        local_sha = (meta_now.get("sha1") or "").strip()
        if remote_sha and local_sha and remote_sha == local_sha and xbmcvfs.exists(cache_path):
            log(f"{kind_name}: unchanged (sha1 sidecar) -> skip refresh", xbmc.LOGINFO)
            return False
    except Exception:
        pass

    # NOTE: For movies we disable conditional request headers.
    # Some servers/proxies returned a stale 304 for kodiMovie.json even when the
    # content changed (ETag/Last-Modified not updated). We still detect changes
    # via sha1 inside _conditional_fetch, so this only increases network traffic.
    use_conditional = (kind_name != "movies")
    # If the sidecar hash says it changed, force a full GET (avoid stale 304).
    if remote_sha:
        try:
            meta_now = _load_meta(meta_p)
            if (meta_now.get("sha1") or "").strip() != remote_sha:
                use_conditional = False
        except Exception:
            use_conditional = False
    status, data_bytes, _headers = _conditional_fetch(url, meta_p, use_conditional=use_conditional)

    if status == "not_modified":
        log(f"{kind_name}: not modified (304/meta/sha1) -> skip refresh", xbmc.LOGINFO)
        return False

    if status == "changed":
        log(f"{kind_name}: changed -> refreshing", xbmc.LOGINFO)

        if not data_bytes:
            log(f"{kind_name}: changed, aber keine Daten empfangen.", xbmc.LOGWARNING)
            return False

        if not _save_cache_json(cache_path, data_bytes):
            log(f"{kind_name}: Cache-Write fehlgeschlagen – breche Refresh ab.", xbmc.LOGWARNING)
            return False

        if kind_name.lower() == "series":
            try:
                import series as _series_mod
                inv = getattr(_series_mod, "invalidate_series_progress_cache", None)
                if callable(inv):
                    inv()
                    log("Series: Progress-Cache via series.invalidate_series_progress_cache() invalidiert.", xbmc.LOGINFO)

                # NEW: Prefetch details.json for series whose details_sha1 changed
                try:
                    series_list = []
                    try:
                        series_list = json.loads(data_bytes.decode("utf-8-sig"))
                    except Exception:
                        series_list = json.loads(data_bytes.decode("utf-8", errors="replace"))

                    if isinstance(series_list, list) and series_list:
                        sha_map = common.load_details_sha1_map()
                        todo = []
                        for s in series_list:
                            if not isinstance(s, dict):
                                continue
                            durl = (s.get("details_url") or "").strip()
                            dsha = (s.get("details_sha1") or "").strip()
                            if not durl or not dsha:
                                continue
                            key = _details_url_key(durl)
                            if sha_map.get(key) != dsha:
                                todo.append((durl, dsha))

                        if todo:
                            log(f"Series: {len(todo)} details.json geaendert -> prefetch", xbmc.LOGINFO)
                            # Limit to avoid accidental storms
                            for durl, dsha in todo[:25]:
                                try:
                                    _series_mod.load_series_details_json(durl, timeout_s=20, expected_sha1=dsha)
                                except Exception:
                                    pass
                        _bump_details_cache(f"series refresh details todo={len(todo)}")
                except Exception as e:
                    log(f"Series: Details-Prefetch fehlgeschlagen: {e}", xbmc.LOGDEBUG)
            except Exception as e:
                log(f"Series: Fehler beim Invalidieren des Progress-Caches: {e}", xbmc.LOGDEBUG)

        if callable(module_refresh_fn):
            try:
                # Pass source so movies.py (and future hooks) can log which path triggered them.
                try:
                    module_refresh_fn(source="service")
                except TypeError:
                    module_refresh_fn()
            except Exception as e:
                log(f"{kind_name}: module refresh failed (Hook): {e}", xbmc.LOGWARNING)

        return True

    log(f"{kind_name}: conditional check failed (url={common.mask_url(url)})", xbmc.LOGWARNING)
    return False


def background_refresh_all_once(do_series: bool = True, do_movies: bool = True, do_fav: bool = True,
                               head_series: bool = False, head_movies: bool = False) -> tuple:
    """
    One-shot background refresh.
    Returns (changed_any, series_changed).

    - If head_series/head_movies is True, do a lightweight HEAD probe first.
      When HEAD indicates "not modified", we skip the heavier conditional GET.
      On probe failure we fall back to the normal refresh.
    """
    changed_any = False
    series_changed = False

    scheme, host, base_path = _normalize_host_and_basepath()
    if not host:
        log("Service: Host missing -> skip refresh", xbmc.LOGWARNING)
        return False, False

    # --- Series ---
    if do_series:
        try:
            series_url = f"{scheme}://{host}{base_path}{SERIES_JSON_FILENAME}"
            series_meta = _meta_path(SERIES_JSON_FILENAME)

            if head_series:
                h = _head_not_modified(series_url, series_meta, timeout=8)
                if h is True:
                    log("Service: Series list unchanged (HEAD) -> skip series refresh", xbmc.LOGINFO)
                elif h is None:
                    log("Service: Series HEAD probe failed -> fallback to conditional GET", xbmc.LOGINFO)
                    sc = _refresh_kind("series", SERIES_JSON_FILENAME, SERIES_CACHE_PATH, _SERIES_REFRESH)
                    if sc:
                        changed_any = True
                        series_changed = True
                else:
                    sc = _refresh_kind("series", SERIES_JSON_FILENAME, SERIES_CACHE_PATH, _SERIES_REFRESH)
                    if sc:
                        changed_any = True
                        series_changed = True
            else:
                sc = _refresh_kind("series", SERIES_JSON_FILENAME, SERIES_CACHE_PATH, _SERIES_REFRESH)
                if sc:
                    changed_any = True
                    series_changed = True
        except Exception as e:
            log(f"Service: series refresh failed: {e}", xbmc.LOGWARNING)

    # --- Movies ---
    if do_movies:
        try:
            movies_url = f"{scheme}://{host}{base_path}{MOVIES_JSON_FILENAME}"
            movies_meta = _meta_path(MOVIES_JSON_FILENAME)

            if head_movies:
                h = _head_not_modified(movies_url, movies_meta, timeout=8)
                if h is True:
                    log("Service: Movies list unchanged (HEAD) -> skip movies refresh", xbmc.LOGINFO)
                elif h is None:
                    log("Service: Movies HEAD probe failed -> fallback to conditional GET", xbmc.LOGINFO)
                    mc = _refresh_kind("movies", MOVIES_JSON_FILENAME, MOVIES_CACHE_PATH, _MOVIES_REFRESH)
                    if mc:
                        changed_any = True
                else:
                    mc = _refresh_kind("movies", MOVIES_JSON_FILENAME, MOVIES_CACHE_PATH, _MOVIES_REFRESH)
                    if mc:
                        changed_any = True
            else:
                mc = _refresh_kind("movies", MOVIES_JSON_FILENAME, MOVIES_CACHE_PATH, _MOVIES_REFRESH)
                if mc:
                    changed_any = True
        except Exception as e:
            log(f"Service: movies refresh failed: {e}", xbmc.LOGWARNING)

    # --- Favorites ---
    if do_fav:
        try:
            fc = _refresh_kind("favorites", FAV_JSON_FILENAME, FAV_CACHE_PATH, _FAV_REFRESH)
            if fc:
                changed_any = True
        except Exception as e:
            log(f"Service: favorites refresh failed: {e}", xbmc.LOGWARNING)

    return changed_any, series_changed


def _safe_mkdir(p):
    try:
        if p and not xbmcvfs.exists(p):
            xbmcvfs.mkdirs(p)
    except Exception:
        pass


def _details_url_key(url: str) -> str:
    """Stable key for details_url to avoid token/ts/sig churn but keep identity (pid/id).

    Keeps only selected query params (pid,id,imdb,tmdb,slug,season,episode,lang).
    Removes common volatile auth params (token,ts,expires,sig,signature,key,auth).
    """
    try:
        if callable(_SERIES_DETAILS_URL_KEY):
            return str(_SERIES_DETAILS_URL_KEY(url) or '').strip()
    except Exception:
        pass

    try:
        u = (url or '').strip()
        if not u:
            return ''
        p = urllib.parse.urlsplit(u)
        host = p.hostname or ""
        if p.port:
            host = f"{host}:{p.port}"
        if not host:
            host = str(p.netloc).split("@")[-1]
        base = f"{p.scheme}://{host}{p.path}"
        q = urllib.parse.parse_qsl(p.query, keep_blank_values=True)
        if not q:
            return base
        keep = {'pid','id','imdb','tmdb','slug','season','episode','lang'}
        drop = {'token','ts','t','time','expires','exp','sig','signature','key','auth','hash'}
        kept = [(k,v) for (k,v) in q if (k.lower() in keep) and (k.lower() not in drop)]
        if not kept:
            # If nothing kept, fall back to base only
            return base
        kept.sort(key=lambda kv: (kv[0].lower(), kv[1]))
        return base + '?' + urllib.parse.urlencode(kept)
    except Exception:
        return (url or '').strip()


def _hash_url(u: str) -> str:
    """Hash for details cache filenames.

    IMPORTANT: Must match series.py (_details_cache_path) => sha1(details_key).
    """
    try:
        return hashlib.sha1((u or "").encode("utf-8", "ignore")).hexdigest()
    except Exception:
        try:
            return hashlib.sha1(str(u).encode("utf-8", "ignore")).hexdigest()
        except Exception:
            return ""


def _read_state():
    try:
        if not _exists_any(SERIES_PREFETCH_STATE_FILE):
            return {"last_ts": 0, "seen": {}}
        with xbmcvfs.File(SERIES_PREFETCH_STATE_FILE, "r") as fh:
            raw = fh.read()
        st = json.loads(raw) if raw else {}
        if not isinstance(st, dict):
            return {"last_ts": 0, "seen": {}}
        st.setdefault("last_ts", 0)
        st.setdefault("seen", {})
        if not isinstance(st["seen"], dict):
            st["seen"] = {}
        return st
    except Exception:
        return {"last_ts": 0, "seen": {}}


def _write_state(st):
    try:
        return bool(common.atomic_write_json(SERIES_PREFETCH_STATE_FILE, st or {}, indent=None, durable=True, backup=True))
    except Exception:
        return False


def _load_series_list_for_prefetch():
    """
    Nutzt bevorzugt den bereits lokal gecachten series.cache.json (von _refresh_kind).
    Fallback: direkte Remote-Abfrage (auth).
    """
    # 1) lokaler Cache
    try:
        if _exists_any(SERIES_CACHE_PATH):
            with xbmcvfs.File(SERIES_CACHE_PATH, "r") as fh:
                raw = fh.read()
            data = json.loads(raw) if raw else []
            if isinstance(data, list):
                return data
    except Exception:
        pass

    # 2) Remote Fallback
    try:
        scheme, host, base_path = _normalize_host_and_basepath()
        if not host:
            return []
        url = f"{scheme}://{host}{base_path}{SERIES_JSON_FILENAME}"
        req = common.get_authenticated_request(url)
        if not req:
            return []
        with _robust_urlopen(req, timeout=NET_TIMEOUT) as resp:
            if resp.getcode() != 200:
                return []
            data = json.loads((resp.read() or b"").decode("utf-8-sig"))
            return data if isinstance(data, list) else []
    except Exception:
        return []


def _download_and_cache_artwork(url: str):
    try:
        if not url or not isinstance(url, str):
            return False
        u = url.strip()
        if not (u.startswith("http://") or u.startswith("https://")):
            return False
        # in den Serien-Image-Cache (wie sonst auch)
        cache_dir = CACHE_DIRS.get("series") or os.path.join(common.addon_data_dir, "cache_series_images")
        return bool(common.download_and_cache_image(u, cache_dir=cache_dir, timeout_s=15))
    except Exception:
        return False


def run_series_details_prefetch(force=False, ignore_interval=False):
    """Background prefetch for series details.json + artwork.

    Optimized: uses `details_sha1` from kodiSeries.json (manifest) to decide what to refresh.
    => No per-details HEAD/ETag requests when nothing changed.
    """

    # Prevent overlapping prefetch threads (and allow series.py to invalidate in-memory details cache)
    win = None
    try:
        win = xbmcgui.Window(10000)
        if win.getProperty(SERIES_PREFETCH_PROP) == '1':
            log('details_prefetch: skip (already running)', xbmc.LOGDEBUG)
            return
        win.setProperty(SERIES_PREFETCH_PROP, '1')
    except Exception:
        win = None

    try:
        def _inner():
            interval_h = SERIES_PREFETCH_INTERVAL_HOURS
            limit = SERIES_PREFETCH_LIMIT_PER_RUN

            _safe_mkdir(SERIES_DETAILS_CACHE_DIR)
            _safe_mkdir(SERIES_IMAGES_CACHE_DIR)

            if series is None or not hasattr(series, "load_series_details_json"):
                log("details_prefetch: skip (series module unavailable)", xbmc.LOGWARNING)
                return
            fetch_details_fn = getattr(series, "load_series_details_json", None)
            if not callable(fetch_details_fn):
                log("details_prefetch: skip (series details loader unavailable)", xbmc.LOGWARNING)
                return

            now_ts = int(time.time())
            state = common.safe_read_json(SERIES_PREFETCH_STATE_FILE, default={})
            last_run = int(state.get("last_run", 0) or 0)
            # Defensive local copy: avoids any accidental scope issues if this function
            # gets refactored later and a local assignment is introduced.
            effective_ignore_interval = bool(ignore_interval)

            if (not force) and (not effective_ignore_interval) and last_run and (now_ts - last_run) < (interval_h * 3600):
                remain = int((interval_h * 3600) - (now_ts - last_run))
                log(f"details_prefetch: skip (noch {remain}s bis zum nächsten Lauf)")
                # details.json bereits lokal -> Artwork direkt prefetchen (kein Race Condition)
                _run_series_details_artwork_prefetch()
                return

            seen = state.get("seen", {}) or {}
            sha_map = common.load_details_sha1_map()

            series_list = common.safe_read_json(SERIES_CACHE_PATH, default=[])
            if not isinstance(series_list, list) or not series_list:
                log("details_prefetch: no series cache -> skip", xbmc.LOGINFO)
                return

            checked = 0
            changed = 0
            not_modified = 0
            missing = 0
            fetched = 0
            artwork_cached = 0
            artwork_fetched = 0
            artwork_failed = 0
            fetched_titles = []
            changed_titles = []
            missing_titles = []

            def _sha1_of_file_vfs(fp):
                try:
                    h = hashlib.sha1()
                    with xbmcvfs.File(fp, "rb") as fh:
                        while True:
                            chunk = fh.read(1024 * 64)
                            if not chunk:
                                break
                            if isinstance(chunk, str):
                                chunk = chunk.encode("utf-8", errors="ignore")
                            h.update(chunk)
                    return h.hexdigest()
                except Exception:
                    return None

            def _read_json_vfs(fp):
                try:
                    with xbmcvfs.File(fp, "r") as fh:
                        raw = fh.read()
                    if isinstance(raw, bytes):
                        raw = raw.decode("utf-8", errors="replace")
                    raw = (raw or "").lstrip("\ufeff").strip()
                    return json.loads(raw) if raw else None
                except Exception:
                    return None

            for serie in series_list:
                if not isinstance(serie, dict):
                    continue

                details_url = (serie.get("details_url") or "").strip()
                if not details_url.startswith("http"):
                    continue

                expected_sha1 = (serie.get("details_sha1") or "").strip()

                details_key = _details_url_key(details_url)
                cache_fp = os.path.join(SERIES_DETAILS_CACHE_DIR, f"{_hash_url(details_key)}.json")
                have_cache = xbmcvfs.exists(cache_fp)

                local_sha = sha_map.get(details_key)
                if have_cache and not local_sha:
                    local_sha = _sha1_of_file_vfs(cache_fp)
                    if local_sha:
                        sha_map[details_key] = local_sha
                        try:
                            common.save_details_sha1_map(sha_map)
                        except Exception:
                            pass

                need_fetch = False
                if force:
                    need_fetch = True
                else:
                    if expected_sha1:
                        if (not have_cache) or (local_sha != expected_sha1):
                            need_fetch = True
                    else:
                        # fallback: no expected hash -> only fetch if cache missing
                        if not have_cache:
                            need_fetch = True

                # per-item throttle (avoid hammering on repeated failures)
                if need_fetch and not force:
                    last = int(seen.get(details_key, 0) or 0)
                    if last and (now_ts - last) < 1800:
                        continue

                details_json = None
                if need_fetch:
                    _serie_title = (serie.get("title") or serie.get("name") or details_key or "?")
                    try:
                        details_json = fetch_details_fn(
                            details_url,
                            timeout_s=NET_TIMEOUT,
                            expected_sha1=(expected_sha1 or None),
                        )
                        seen[details_key] = now_ts

                        if details_json:
                            fetched += 1
                            fetched_titles.append(_serie_title)
                            if expected_sha1 and local_sha != expected_sha1:
                                changed += 1
                                changed_titles.append(_serie_title)
                        else:
                            missing += 1
                            missing_titles.append(_serie_title)
                    except Exception as e:
                        missing += 1
                        missing_titles.append(_serie_title)
                        log(f"details_prefetch: fetch failed url={common.mask_url(details_url)} serie='{_serie_title}': {e}", xbmc.LOGWARNING)
                else:
                    not_modified += 1

                # for season artwork, read cached json if we don't have it already
                if details_json is None and have_cache:
                    details_json = _read_json_vfs(cache_fp)

                # Artwork URLs sammeln
                art_urls = set()

                poster = (serie.get("poster") or "").strip()
                if poster.startswith("http"):
                    art_urls.add(poster)

                backcover = (serie.get("backcover") or serie.get("fanart") or "").strip()
                if backcover.startswith("http"):
                    art_urls.add(backcover)

                if isinstance(details_json, dict):
                    # details.json uses top-level keys like "season1", "season2", "staffel1" etc.
                    # NOT a "seasons" list – using the same logic as _run_series_details_artwork_prefetch.
                    _season_keys = [k for k in details_json.keys()
                                    if isinstance(k, str) and (k.lower().startswith("season") or k.lower().startswith("staffel"))]
                    for sk in _season_keys:
                        sv = details_json.get(sk)
                        if not isinstance(sv, dict):
                            continue
                        for _art_key in ("poster", "still", "fanart", "thumb"):
                            _u = (sv.get(_art_key) or "").strip()
                            if _u.startswith("http"):
                                art_urls.add(_u)

                for u in art_urls:
                    # Erst lokales Cache-Hit prüfen, sonst wirklich im Hintergrund downloaden.
                    if common.get_cached_image(u, cache_dir=SERIES_IMAGES_CACHE_DIR):
                        artwork_cached += 1
                    else:
                        try:
                            if _download_and_cache_artwork(u):
                                artwork_fetched += 1
                            else:
                                artwork_failed += 1
                        except Exception:
                            artwork_failed += 1

                checked += 1
                if checked >= limit:
                    break

            state["last_run"] = now_ts
            state["seen"] = seen
            common.atomic_write_json(SERIES_PREFETCH_STATE_FILE, state)

            # Persist sha_map (useful if we computed local hashes)
            try:
                common.save_details_sha1_map(sha_map)
            except Exception:
                pass

            log(
                f"details_prefetch: checked={checked} changed={changed} not_modified={not_modified} "
                f"missing={missing} fetched={fetched} artwork_cached={artwork_cached} "
                f"artwork_fetched={artwork_fetched} artwork_failed={artwork_failed}"
            )
            if changed_titles:
                log(f"details_prefetch: Geaenderte details.json fuer: {', '.join(changed_titles)}", xbmc.LOGINFO)
            if fetched_titles and not changed_titles:
                log(f"details_prefetch: Neu gecacht (kein sha1-Change): {', '.join(fetched_titles)}", xbmc.LOGDEBUG)
            if missing_titles:
                log(f"details_prefetch: Fehlgeschlagen/leer fuer: {', '.join(missing_titles)}", xbmc.LOGWARNING)

            # If we fetched any fresh details.json, notify series.py to drop stale in-memory cache
            if 'fetched' in locals() and fetched:
                _bump_details_cache(f'prefetch fetched={fetched}')

            # Artwork aus gecachten details.json laden (direkt/sequenziell - details.json sind jetzt auf Disk)
            _run_series_details_artwork_prefetch()

        _inner()
    finally:
        if win:
            try:
                win.clearProperty(SERIES_PREFETCH_PROP)
            except Exception:
                pass


def _run_movies_image_prefetch():
    """Lädt fehlende Movie-Poster/Fanart proaktiv in den Cache."""
    try:
        cache_dir = CACHE_DIRS.get("movies") or os.path.join(common.addon_data_dir, "cache_movies_images")
        if not cache_dir:
            return
        try:
            os.makedirs(cache_dir, exist_ok=True)
        except Exception:
            pass
        if not os.path.exists(MOVIES_CACHE_PATH):
            log("movies_image_prefetch: cache nicht gefunden, skip", xbmc.LOGDEBUG)
            return
        try:
            import json as _json
            with open(MOVIES_CACHE_PATH, "r", encoding="utf-8") as f:
                movie_list = _json.load(f)
        except Exception as e:
            log(f"movies_image_prefetch: cache lesen fehlgeschlagen: {e}", xbmc.LOGDEBUG)
            return
        if not isinstance(movie_list, list):
            return
        art_keys = ["poster", "backcover", "fanart", "thumb", "logo", "clearart", "disc", "discart", "banner"]
        downloaded = skipped = failed = 0
        downloaded_titles = []
        for movie in movie_list:
            if not isinstance(movie, dict):
                continue
            urls = set()
            for k in art_keys:
                u = (movie.get(k) or "").strip()
                if u.startswith("http"):
                    urls.add(u)
            images = movie.get("images", {})
            if isinstance(images, dict):
                for u in images.values():
                    if isinstance(u, str) and u.startswith("http"):
                        urls.add(u.strip())
            movie_downloaded = 0
            for u in urls:
                if common.get_cached_image(u, cache_dir=cache_dir):
                    skipped += 1
                else:
                    ok = common.download_and_cache_image(u, cache_dir=cache_dir, timeout_s=12)
                    if ok:
                        downloaded += 1
                        movie_downloaded += 1
                    else:
                        failed += 1
                        title = movie.get("title") or movie.get("name") or "?"
                        log(f"movies_image_prefetch: FAILED url={common.mask_url(u)} movie='{title}'", xbmc.LOGWARNING)
            if movie_downloaded:
                title = movie.get("title") or movie.get("name") or "?"
                downloaded_titles.append(f"'{title}' ({movie_downloaded} Bilder)")
        if downloaded_titles:
            log(f"movies_image_prefetch: Neue Bilder heruntergeladen fuer: {', '.join(downloaded_titles)}", xbmc.LOGINFO)
        log(f"movies_image_prefetch: downloaded={downloaded} skipped(cached)={skipped} failed={failed} total_movies={len(movie_list)}", xbmc.LOGINFO)
    except Exception as e:
        log(f"movies_image_prefetch: Fehler: {e}", xbmc.LOGDEBUG)


def _start_movies_image_prefetch_thread():
    """Startet Movie-Artwork-Prefetch in eigenem Thread."""
    def _run():
        try:
            _run_movies_image_prefetch()
        except Exception as e:
            log(f"movies_image_prefetch thread: {e}", xbmc.LOGDEBUG)
    try:
        threading.Thread(target=_run, daemon=True).start()
    except Exception as e:
        log(f"movies_image_prefetch thread start: {e}", xbmc.LOGDEBUG)


def _run_series_image_prefetch():
    """Lädt fehlende Serien-Poster/Backcover proaktiv in den Cache."""
    try:
        cache_dir = CACHE_DIRS.get("series") or os.path.join(common.addon_data_dir, "cache_series_images")
        if not cache_dir:
            return
        try:
            os.makedirs(cache_dir, exist_ok=True)
        except Exception:
            pass
        if not os.path.exists(SERIES_CACHE_PATH):
            log("series_image_prefetch: cache nicht gefunden, skip", xbmc.LOGDEBUG)
            return
        try:
            import json as _json
            with open(SERIES_CACHE_PATH, "r", encoding="utf-8") as f:
                series_list = _json.load(f)
        except Exception as e:
            log(f"series_image_prefetch: cache lesen fehlgeschlagen: {e}", xbmc.LOGDEBUG)
            return
        if not isinstance(series_list, list):
            return
        art_keys = ["poster", "backcover", "fanart", "thumb", "banner"]
        downloaded = skipped = failed = 0
        downloaded_titles = []
        for serie in series_list:
            if not isinstance(serie, dict):
                continue
            urls = set()
            for k in art_keys:
                u = (serie.get(k) or "").strip()
                if u.startswith("http"):
                    urls.add(u)
            serie_downloaded = 0
            for u in urls:
                if common.get_cached_image(u, cache_dir=cache_dir):
                    skipped += 1
                else:
                    ok = common.download_and_cache_image(u, cache_dir=cache_dir, timeout_s=12)
                    if ok:
                        downloaded += 1
                        serie_downloaded += 1
                    else:
                        failed += 1
                        title = serie.get("title") or serie.get("name") or "?"
                        log(f"series_image_prefetch: FAILED url={common.mask_url(u)} serie='{title}'", xbmc.LOGWARNING)
            if serie_downloaded:
                title = serie.get("title") or serie.get("name") or "?"
                downloaded_titles.append(f"'{title}' ({serie_downloaded} Bilder)")
        if downloaded_titles:
            log(f"series_image_prefetch: Neue Bilder heruntergeladen fuer: {', '.join(downloaded_titles)}", xbmc.LOGINFO)
        log(f"series_image_prefetch: downloaded={downloaded} skipped(cached)={skipped} failed={failed} total_series={len(series_list)}", xbmc.LOGINFO)
    except Exception as e:
        log(f"series_image_prefetch: Fehler: {e}", xbmc.LOGDEBUG)


def _start_series_image_prefetch_thread():
    """Startet Serien-Artwork-Prefetch in eigenem Thread."""
    def _run():
        try:
            _run_series_image_prefetch()
        except Exception as e:
            log(f"series_image_prefetch thread: {e}", xbmc.LOGDEBUG)
    try:
        threading.Thread(target=_run, daemon=True).start()
    except Exception as e:
        log(f"series_image_prefetch thread start: {e}", xbmc.LOGDEBUG)


def _run_series_details_artwork_prefetch():
    """Liest alle gecachten details.json und lädt fehlende Season/Episode-Bilder."""
    try:
        img_cache_dir = SERIES_IMAGES_CACHE_DIR
        details_dir = SERIES_DETAILS_CACHE_DIR
        if not img_cache_dir or not details_dir:
            return
        try:
            os.makedirs(img_cache_dir, exist_ok=True)
        except Exception:
            pass
        if not os.path.isdir(details_dir):
            return
        import json as _json
        downloaded = skipped = failed = 0
        processed_files = 0
        for fname in os.listdir(details_dir):
            if not fname.endswith(".json"):
                continue
            fpath = os.path.join(details_dir, fname)
            try:
                with open(fpath, "r", encoding="utf-8") as f:
                    details = _json.load(f)
            except Exception:
                continue
            if not isinstance(details, dict):
                continue
            processed_files += 1
            art_urls = set()
            # details.json Struktur: Top-Level Keys "season1","season2",... oder "staffel1",...
            # (KEINE "seasons"-Liste!)
            season_keys = [k for k in details.keys()
                           if isinstance(k, str) and (k.lower().startswith("season") or k.lower().startswith("staffel"))]
            for sk in season_keys:
                sv = details.get(sk)
                if not isinstance(sv, dict):
                    continue
                for key in ("poster", "still", "fanart", "thumb"):
                    u = (sv.get(key) or "").strip()
                    if u.startswith("http"):
                        art_urls.add(u)
                # Episoden
                eps = sv.get("episodes") or sv.get("Episoden") or []
                if isinstance(eps, list):
                    for ep in eps:
                        if not isinstance(ep, dict):
                            continue
                        for key in ("still", "thumb"):
                            u = (ep.get(key) or "").strip()
                            if u.startswith("http"):
                                art_urls.add(u)
                elif isinstance(eps, dict):
                    for ep in eps.values():
                        if not isinstance(ep, dict):
                            continue
                        for key in ("still", "thumb"):
                            u = (ep.get(key) or "").strip()
                            if u.startswith("http"):
                                art_urls.add(u)
            file_downloaded = 0
            for u in art_urls:
                if common.get_cached_image(u, cache_dir=img_cache_dir):
                    skipped += 1
                else:
                    ok = common.download_and_cache_image(u, cache_dir=img_cache_dir, timeout_s=12)
                    if ok:
                        downloaded += 1
                        file_downloaded += 1
                    else:
                        failed += 1
            if file_downloaded:
                log(f"series_details_artwork_prefetch: {file_downloaded} neue Bilder aus {fname}", xbmc.LOGDEBUG)
        log(f"series_details_artwork_prefetch: processed_files={processed_files} downloaded={downloaded} skipped(cached)={skipped} failed={failed}", xbmc.LOGINFO)
    except Exception as e:
        log(f"series_details_artwork_prefetch: Fehler: {e}", xbmc.LOGDEBUG)


def _start_series_details_artwork_prefetch_thread():
    """Startet Season/Episode-Artwork-Prefetch in eigenem Thread."""
    def _run():
        try:
            _run_series_details_artwork_prefetch()
        except Exception as e:
            log(f"series_details_artwork_prefetch thread: {e}", xbmc.LOGDEBUG)
    try:
        threading.Thread(target=_run, daemon=True).start()
    except Exception as e:
        log(f"series_details_artwork_prefetch thread start: {e}", xbmc.LOGDEBUG)


def _start_series_prefetch_thread(force=False, ignore_interval=False):
    """
    Prefetch nicht im Service-Hauptthread blockieren.
    """
    try:
        win = xbmcgui.Window(10000)
        if win.getProperty(SERIES_PREFETCH_PROP) == '1':
            log('Series-Prefetch: already running -> skip', xbmc.LOGDEBUG)
            return False
    except Exception:
        pass

    try:
        th = threading.Thread(target=run_series_details_prefetch, args=(force, ignore_interval))
        th.daemon = True
        th.start()
        log(f"Series-Prefetch: thread started force={force} ignore_interval={ignore_interval}", xbmc.LOGINFO)
        return True
    except Exception:
        return False

# ---------------------------------------------------------------------
# Orphan Artwork Cache Cleanup (Start + 1x/Tag)
# ---------------------------------------------------------------------
CACHE_CLEANUP_INTERVAL_HOURS = 24
CACHE_CLEANUP_STATE_FILE = os.path.join(common.addon_data_dir, "cache_cleanup_state.json")
CACHE_CLEANUP_PROP = f"{common.addon_id}.cache.cleanup.running"
# After a fresh Kodi start, run one cleanup this many seconds after service launch.
# Must be > grace_sec (180s) so BG-refresh threads have finished writing JSONs.
CACHE_CLEANUP_STARTUP_DELAY_SEC = 180  # 3 Minuten

CACHE_DIRS = {
    "movies": os.path.join(common.addon_data_dir, "cache_movies_images"),
    "series": os.path.join(common.addon_data_dir, "cache_series_images"),
    "fav_series": os.path.join(common.addon_data_dir, "cache_fav_series_images"),
}

# ---------------------------------------------------------------------
# Payload Cache Cleanup (Favorites) – nur Cache, kann jederzeit neu gebaut werden
# ---------------------------------------------------------------------
PAYLOAD_GC_INTERVAL_HOURS = 6
PAYLOAD_GC_STATE_FILE = os.path.join(common.addon_data_dir, "payload_gc_state.json")
PAYLOAD_GC_PROP = f"{common.addon_id}.payload.gc.running"

# Payload-Cache-Verzeichnisse (aktuell nur Favorites)
PAYLOAD_CACHE_DIRS = {
    "fav_payload": os.path.join(common.addon_data_dir, "payload_cache_fav"),
}


IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp"}
_HASHED_CACHE_RE = re.compile(r"^[0-9a-fA-F]{32}(\.[A-Za-z0-9]{1,8})?$")  # tolerant


def _ensure_dir_slash(p):
    try:
        if p and not p.endswith(("/", "\\")):
            return p + "/"
    except Exception:
        pass
    return p


def _translate_path(p):
    try:
        return xbmcvfs.translatePath(p)
    except Exception:
        return p


def _exists_any(path):
    """
    exists() mit VFS + OS-Fallback (wichtig, weil xbmcvfs.exists manchmal "zickt")
    """
    if not path:
        return False
    try:
        if xbmcvfs.exists(path):
            return True
    except Exception:
        pass
    try:
        lp = _translate_path(path)
        return os.path.exists(lp)
    except Exception:
        return False


def _listdir_any(dir_path):
    """
    listdir() mit VFS + OS-Fallback.
    Gibt (dirs, files) zurück.
    """
    try:
        dirs, files = xbmcvfs.listdir(_ensure_dir_slash(dir_path))
        return (dirs or []), (files or [])
    except Exception:
        pass

    try:
        lp = _translate_path(dir_path)
        if os.path.isdir(lp):
            return [], os.listdir(lp)
    except Exception:
        pass

    return [], []


def _vfs_join(parent, child):
    """
    Join für VFS-Pfade (immer '/'), damit Windows-Backslashes nicht kaputt machen.
    """
    parent = (parent or "").rstrip("/").rstrip("\\")
    child = (child or "").lstrip("/").lstrip("\\")
    return parent + "/" + child


def _read_json_file(path):
    try:
        if not path or not _exists_any(path):
            return None
        with xbmcvfs.File(path, "r") as fh:
            raw = fh.read()
        if not raw:
            return None
        return json.loads(raw)
    except Exception:
        return None


def _extract_http_urls(obj, out_set):
    if obj is None:
        return
    if isinstance(obj, str):
        s = obj.strip()
        if s.startswith("http://") or s.startswith("https://"):
            out_set.add(s)
        return
    if isinstance(obj, dict):
        # URLs können bei einigen Cache-Dateien auch als dict-KEY vorkommen (z.B. alte watched_status oder URL->Meta Maps)
        for k, v in obj.items():
            if isinstance(k, str):
                ks = k.strip()
                if ks.startswith("http://") or ks.startswith("https://"):
                    out_set.add(ks)
            _extract_http_urls(v, out_set)
        return
    if isinstance(obj, list):
        for v in obj:
            _extract_http_urls(v, out_set)


def _url_to_keep_names(url):
    """Build all possible cache filenames for a given URL.

    IMPORTANT: Must cover ALL naming schemes used across the codebase:
      1. basename  – legacy schema (plain filename from URL path)
      2. MD5(raw)  – movies.py get_movie_cache_filename (MD5 of raw URL)
      3. SHA1(normalized) – gb_common.get_cache_filename (SHA1 of credentials-stripped URL)
         This is what _run_movies_image_prefetch / _run_series_image_prefetch write!
    Without scheme 3, the cleanup deletes ALL currently-cached images on every run.
    """
    keep = set()
    if not isinstance(url, str) or not url:
        return keep

    try:
        ext = os.path.splitext(urllib.parse.urlparse(url).path or "")[1] or ""
        if not ext or len(ext) > 8:
            ext = ".jpg"
    except Exception:
        ext = ".jpg"
    ext_variants = {ext.lower(), ".jpg", ".jpeg", ".png", ".webp"}

    # 1) basename (legacy)
    try:
        base = os.path.basename(urllib.parse.urlparse(url).path or "").strip()
        if base:
            keep.add(base)
    except Exception:
        pass

    # 2) MD5 of raw URL (movies.py: get_movie_cache_filename)
    try:
        h_md5 = hashlib.md5(url.encode("utf-8")).hexdigest()
        for e in ext_variants:
            keep.add(h_md5 + e)
    except Exception:
        pass

    # 3) SHA1 of normalized URL (gb_common.get_cache_filename)
    # gb_common strips credentials + unquotes before hashing with SHA1.
    # This is what _run_movies_image_prefetch and _run_series_image_prefetch use.
    try:
        u_norm = urllib.parse.unquote((url or "").strip())
        # strip credentials (user:pass@host -> host)
        p = urllib.parse.urlsplit(u_norm)
        if p.scheme and p.netloc:
            host = p.hostname or p.netloc.split("@")[-1]
            if p.port:
                host = f"{host}:{p.port}"
            u_norm = urllib.parse.urlunsplit((p.scheme, host, p.path, p.query, p.fragment))
        h_sha1 = hashlib.sha1(u_norm.encode("utf-8", errors="ignore")).hexdigest()
        for e in ext_variants:
            keep.add(h_sha1 + e)
    except Exception:
        pass

    return keep




def _save_json_atomic(path, obj, indent=2):
    try:
        return bool(common.atomic_write_json(path, obj, indent=indent, durable=True, backup=True))
    except Exception:
        try:
            with xbmcvfs.File(path, "w") as fh:
                fh.write(json.dumps(obj, ensure_ascii=False, indent=indent))
            return True
        except Exception:
            return False


def _load_active_series_details_keys() -> set:
    keys = set()
    for fp in (SERIES_CACHE_PATH, FAV_CACHE_PATH):
        try:
            data = common.safe_read_json(fp, default=[])
        except Exception:
            data = []
        if not isinstance(data, list):
            continue
        for it in data:
            if not isinstance(it, dict):
                continue
            durl = (it.get("details_url") or "").strip()
            if not durl.startswith(("http://", "https://")):
                continue
            try:
                k = _details_url_key(durl)
                if k:
                    keys.add(k)
            except Exception:
                pass
    return keys


def _prune_hashed_json_dir(dir_path: str, keep_hashes: set, max_delete: int = 3000, allowed_lengths=(40,)) -> tuple:
    removed = 0
    scanned = 0
    if not dir_path or not _exists_any(dir_path):
        return removed, scanned

    _dirs, files = _listdir_any(dir_path)
    for fn in (files or []):
        if not isinstance(fn, str):
            continue
        lfn = fn.lower().strip()
        if not lfn.endswith(".json"):
            continue
        stem = os.path.splitext(lfn)[0]
        if len(stem) not in set(int(x) for x in (allowed_lengths or (40,))) or not all(c in "0123456789abcdef" for c in stem):
            continue
        scanned += 1
        if stem in keep_hashes:
            continue

        p = _vfs_join(dir_path, fn)
        if _delete_file_any(p, with_sidecars=True):
            removed += 1
            if removed >= int(max_delete):
                break

    return removed, scanned


def _prune_series_payload_cache(active_keys: set, max_delete: int = 4000) -> tuple:
    """Remove stale series payload_cache entries that point to deleted details_url keys."""
    removed = 0
    scanned = 0
    if not SERIES_PAYLOAD_CACHE_DIR or not _exists_any(SERIES_PAYLOAD_CACHE_DIR):
        return removed, scanned

    _dirs, files = _listdir_any(SERIES_PAYLOAD_CACHE_DIR)
    for fn in (files or []):
        if not isinstance(fn, str) or not fn.lower().endswith('.json'):
            continue
        scanned += 1
        fp = _vfs_join(SERIES_PAYLOAD_CACHE_DIR, fn)
        try:
            data = _read_json_file(fp)
        except Exception:
            data = None
        if not isinstance(data, dict):
            continue

        cand = ''
        for key in ('details_url', 'online_serie_key', 'original_online_key'):
            v = (data.get(key) or '') if isinstance(data, dict) else ''
            v = str(v).strip() if v is not None else ''
            if v.startswith(('http://', 'https://')):
                cand = v
                break

        if not cand:
            continue

        try:
            dk = _details_url_key(cand)
        except Exception:
            dk = ''
        if not dk:
            continue
        if dk in active_keys:
            continue

        if _delete_file_any(fp, with_sidecars=True):
            removed += 1
            if removed >= int(max_delete):
                break

    return removed, scanned


def _prune_stale_series_details_artwork_sources():
    """Trim stale series-details caches/indexes so orphan artwork cleanup can delete old images."""
    active_keys = _load_active_series_details_keys()
    if not active_keys:
        log("Cache-Cleanup: details-stale-prune skip (keine aktiven details_keys)", xbmc.LOGINFO)
        return

    keep_hashes = set()
    for k in active_keys:
        try:
            keep_hashes.add(_hash_url(k).lower())
        except Exception:
            pass

    removed_cache, scanned_cache = _prune_hashed_json_dir(SERIES_DETAILS_CACHE_DIR, keep_hashes, allowed_lengths=(40, 32))
    removed_meta_dir, scanned_meta_dir = _prune_hashed_json_dir(SERIES_DETAILS_META_DIR, keep_hashes, allowed_lengths=(40, 32))
    removed_legacy, scanned_legacy = _prune_hashed_json_dir(LEGACY_SERIES_DETAILS_CACHE_DIR, set(), allowed_lengths=(40, 32))
    removed_payload, scanned_payload = _prune_series_payload_cache(active_keys)

    removed_seen = 0

    removed_prefetch_seen = 0
    try:
        st = _read_json_file(SERIES_PREFETCH_STATE_FILE)
        if isinstance(st, dict):
            seen_map = st.get('seen') if isinstance(st.get('seen'), dict) else {}
            if isinstance(seen_map, dict) and seen_map:
                new_seen = {}
                for k, v in seen_map.items():
                    if not isinstance(k, str):
                        continue
                    try:
                        if _details_url_key(k) in active_keys:
                            new_seen[k] = v
                    except Exception:
                        pass
                removed_prefetch_seen = max(0, len(seen_map) - len(new_seen))
                if removed_prefetch_seen:
                    st['seen'] = new_seen
                    _save_json_atomic(SERIES_PREFETCH_STATE_FILE, st, indent=None)
    except Exception:
        pass
    try:
        seen_raw = _read_json_file(SERIES_SEEN_URLS_FILE)
        seen_urls = []
        if isinstance(seen_raw, dict) and isinstance(seen_raw.get("urls"), list):
            seen_urls = seen_raw.get("urls") or []
        elif isinstance(seen_raw, list):
            seen_urls = seen_raw

        if seen_urls:
            filtered = []
            for u in seen_urls:
                if not isinstance(u, str):
                    continue
                try:
                    if _details_url_key(u) in active_keys:
                        filtered.append(u)
                except Exception:
                    pass
            removed_seen = max(0, len(seen_urls) - len(filtered))
            if removed_seen:
                _save_json_atomic(SERIES_SEEN_URLS_FILE, filtered, indent=2)
    except Exception:
        pass

    removed_meta_map = 0
    try:
        meta_map = _read_json_file(SERIES_DETAILS_META_FILE)
        if isinstance(meta_map, dict):
            new_meta = {k: v for k, v in meta_map.items() if isinstance(k, str) and k in active_keys}
            removed_meta_map = max(0, len(meta_map) - len(new_meta))
            if removed_meta_map:
                _save_json_atomic(SERIES_DETAILS_META_FILE, new_meta, indent=None)
    except Exception:
        pass

    removed_sha = 0
    try:
        sha_map = common.load_details_sha1_map() or {}
        if isinstance(sha_map, dict):
            new_sha = {}
            for k, v in sha_map.items():
                if not isinstance(k, str):
                    continue
                if k in active_keys:
                    new_sha[k] = v
                    continue
                if not (k.startswith("http://") or k.startswith("https://")):
                    # Drop legacy hashed keys (old sha1/md5 details-url keys),
                    # keep only non-url/non-hash custom entries if any.
                    lk = k.lower().strip()
                    is_legacy_hash = (
                        len(lk) in (40, 32) and all(c in "0123456789abcdef" for c in lk)
                    )
                    if not is_legacy_hash:
                        new_sha[k] = v
            removed_sha = max(0, len(sha_map) - len(new_sha))
            if removed_sha:
                common.save_details_sha1_map(new_sha)
    except Exception:
        pass

    log(
        "Cache-Cleanup: details-stale-prune done | active=%d cache_removed=%d/%d meta_dir_removed=%d/%d legacy_removed=%d/%d payload_removed=%d/%d seen_removed=%d prefetch_seen_removed=%d meta_map_removed=%d sha_removed=%d"
        % (
            len(active_keys),
            removed_cache,
            scanned_cache,
            removed_meta_dir,
            scanned_meta_dir,
            removed_legacy,
            scanned_legacy,
            removed_payload,
            scanned_payload,
            removed_seen,
            removed_prefetch_seen,
            removed_meta_map,
            removed_sha,
        ),
        xbmc.LOGINFO,
    )

def _load_last_cleanup_ts():
    data = _read_json_file(CACHE_CLEANUP_STATE_FILE)
    if isinstance(data, dict):
        try:
            return int(data.get("last_ts", 0))
        except Exception:
            return 0
    return 0


def _save_last_cleanup_ts(ts_int):
    return _save_json_atomic(CACHE_CLEANUP_STATE_FILE, {"last_ts": int(ts_int)}, indent=None)


def _iter_cache_files(cache_dir, max_depth=2):
    """
    Robust iterator:
    1) Try xbmcvfs.listdir (VFS)
    2) If it yields nothing but local path exists -> fallback to os.walk

    Yields tuples: (kind, root_dir, filename)
      kind: "vfs" or "os"
    """
    if not cache_dir:
        return

    vfs_dir = _ensure_dir_slash(cache_dir)
    local_dir = _translate_path(cache_dir)

    vfs_exists = False
    try:
        vfs_exists = xbmcvfs.exists(cache_dir)
    except Exception:
        vfs_exists = False

    os_exists = False
    try:
        os_exists = os.path.isdir(local_dir)
    except Exception:
        os_exists = False

    yielded_any = False

    # --- 1) xbmcvfs.listdir ---
    if vfs_exists:
        stack = [(vfs_dir, 0)]
        while stack:
            d, depth = stack.pop()
            try:
                dirs, files = xbmcvfs.listdir(_ensure_dir_slash(d))
            except Exception as e:
                log(f"Cache-Cleanup: xbmcvfs.listdir failed for {d}: {e}", xbmc.LOGWARNING)
                break

            if depth < max_depth:
                for sub in dirs or []:
                    if not sub:
                        continue
                    stack.append((_vfs_join(d, sub) + "/", depth + 1))

            for fn in files or []:
                if fn:
                    yielded_any = True
                    yield "vfs", d.rstrip("/"), fn

    # --- 2) os.walk fallback ---
    if (not yielded_any) and os_exists:
        try:
            base_depth = local_dir.rstrip("/").count(os.sep)
            for root, dirs, files in os.walk(local_dir):
                depth = root.count(os.sep) - base_depth
                if depth > max_depth:
                    dirs[:] = []
                    continue
                for fn in files:
                    if fn:
                        yield "os", root, fn
        except Exception as e:
            log(f"Cache-Cleanup: os.walk failed for {local_dir}: {e}", xbmc.LOGWARNING)


def _delete_targets(path: str, with_sidecars: bool = False):
    p = (path or "").strip()
    if not p:
        return []

    out = [p]
    if not with_sidecars:
        return out

    extras = [p + ".bak", p + ".seed", p + ".seed.bak"]
    lp = p.lower()
    if lp.endswith(".bak") or lp.endswith(".seed"):
        base = p.rsplit(".", 1)[0]
        extras.extend([base, base + ".bak", base + ".seed", base + ".seed.bak"])

    for cand in extras:
        if cand and cand not in out:
            out.append(cand)
    return out


def _delete_file_any(path, with_sidecars: bool = False):
    deleted_any = False
    for cand in _delete_targets(path, with_sidecars=with_sidecars):
        # Try OS delete first (fast) when local path exists, else xbmcvfs
        try:
            lp = _translate_path(cand)
            if os.path.isfile(lp):
                try:
                    os.remove(lp)
                    deleted_any = True
                    continue
                except Exception:
                    pass
        except Exception:
            pass

        # Avoid noisy Kodi error logs by skipping VFS delete when file doesn't exist.
        try:
            if not _exists_any(cand):
                continue
        except Exception:
            pass

        try:
            if xbmcvfs.delete(cand):
                deleted_any = True
                continue
        except Exception:
            pass

        # Final local fallback if VFS delete failed on a local path
        try:
            lp = _translate_path(cand)
            if os.path.isfile(lp):
                os.remove(lp)
                deleted_any = True
        except Exception:
            pass

    return deleted_any


def _prune_orphan_json_sidecars(max_delete: int = 4000) -> tuple:
    """Delete cache-sidecar files whose base .json no longer exists."""
    removed = 0
    scanned = 0
    dirs = [
        SERIES_PAYLOAD_CACHE_DIR,
        SERIES_DETAILS_CACHE_DIR,
        SERIES_DETAILS_META_DIR,
        LEGACY_SERIES_DETAILS_CACHE_DIR,
        PAYLOAD_CACHE_DIRS.get("fav_payload"),
    ]

    for d in dirs:
        if not d or not _exists_any(d):
            continue
        _subdirs, files = _listdir_any(d)
        for fn in (files or []):
            if not isinstance(fn, str):
                continue
            lfn = fn.lower().strip()
            base = None
            if lfn.endswith(".json.seed.bak"):
                base = fn[:-9]
            elif lfn.endswith(".json.seed"):
                base = fn[:-5]
            elif lfn.endswith(".json.bak"):
                base = fn[:-4]
            else:
                continue

            if not base:
                continue

            scanned += 1
            base_path = _vfs_join(d, base)
            if _exists_any(base_path):
                continue

            sidecar_path = _vfs_join(d, fn)
            if _delete_file_any(sidecar_path, with_sidecars=True):
                removed += 1
                if removed >= int(max_delete):
                    return removed, scanned

    return removed, scanned


def _cleanup_cache_dir(cache_dir, keep_names):
    removed = 0
    scanned = 0
    seen = 0

    vfs_exists = False
    try:
        vfs_exists = xbmcvfs.exists(cache_dir)
    except Exception:
        vfs_exists = False

    local_dir = _translate_path(cache_dir)
    os_exists = False
    try:
        os_exists = os.path.isdir(local_dir)
    except Exception:
        os_exists = False

    # Debug pro Cache-Dir (damit wir sehen warum files_seen=0 ist)
    log(f"Cache-Cleanup: dir='{cache_dir}' | vfs_exists={vfs_exists} | local='{local_dir}' | os_exists={os_exists}", xbmc.LOGINFO)

    for kind, d, fn in _iter_cache_files(cache_dir, max_depth=2):
        seen += 1

        ext = os.path.splitext(fn)[1].lower()
        is_image = ext in IMAGE_EXTS
        is_hashed = bool(_HASHED_CACHE_RE.match(fn))

        if not (is_image or is_hashed):
            continue

        scanned += 1
        if fn in keep_names:
            continue

        if kind == "os":
            p = os.path.join(d, fn)
        else:
            p = _vfs_join(d, fn)

        if _delete_file_any(p):
            removed += 1

    return removed, scanned, seen


def _iter_json_files_recursive(root_dir, max_depth=3, max_files=2500):
    """Yield absolute file paths (VFS-compatible) to .json files under root_dir up to max_depth.
    Skips image cache directories to keep it fast.
    """
    if not root_dir:
        return
    # Skip image cache dirs (contain many non-json files)
    skip_names = set()
    try:
        for _d in CACHE_DIRS.values():
            bn = os.path.basename(_translate_path(_d).rstrip("/"))
            if bn:
                skip_names.add(bn)
    except Exception:
        skip_names.update({"cache_movies_images", "cache_series_images", "cache_fav_series_images"})

    seen = 0
    stack = [(root_dir, 0)]
    visited = set()

    while stack and seen < int(max_files):
        d, depth = stack.pop()
        if not d or d in visited:
            continue
        visited.add(d)

        # VFS-first
        try:
            dirs, files = xbmcvfs.listdir(_ensure_dir_slash(d))
            for fn in files or []:
                if fn and fn.lower().endswith(".json"):
                    yield _vfs_join(d, fn)
                    seen += 1
                    if seen >= int(max_files):
                        return
            if depth < int(max_depth):
                for sub in dirs or []:
                    if not sub:
                        continue
                    if sub in skip_names:
                        continue
                    stack.append((_vfs_join(d, sub), depth + 1))
            continue
        except Exception:
            pass

        # OS fallback
        try:
            lp = _translate_path(d)
            if not os.path.isdir(lp):
                continue
            for fn in os.listdir(lp):
                if not fn:
                    continue
                fp = os.path.join(lp, fn)
                if os.path.isfile(fp) and fn.lower().endswith(".json"):
                    yield fp
                    seen += 1
                    if seen >= int(max_files):
                        return
            if depth < int(max_depth):
                for fn in os.listdir(lp):
                    if not fn:
                        continue
                    if fn in skip_names:
                        continue
                    fp = os.path.join(lp, fn)
                    if os.path.isdir(fp):
                        stack.append((fp, depth + 1))
        except Exception:
            pass



def _collect_keep_names_from_all_jsons(return_urls=False):
    """
    Wichtig: VFS + OS-Fallback, damit keep-set nicht leer bleibt,
    wenn xbmcvfs.exists/listdir mal spinnt.

    return_urls=False -> (keep_names, url_count)
    return_urls=True  -> (keep_names, url_count, urls_set)

    Hinweis: Scannt rekursiv auch Unterordner (z.B. series payload_cache),
    überspringt aber die großen Image-Cache-Verzeichnisse.
    """
    keep = set()
    urls = set()

    try:
        if not common.addon_data_dir or not _exists_any(common.addon_data_dir):
            return (keep, 0, urls) if return_urls else (keep, 0)
    except Exception:
        return (keep, 0, urls) if return_urls else (keep, 0)

    # Rekursiv JSON-Dateien sammeln (inkl. payload_cache)
    for p in _iter_json_files_recursive(common.addon_data_dir, max_depth=3, max_files=2500):
        data = _read_json_file(p)
        _extract_http_urls(data, urls)

    for u in urls:
        keep.update(_url_to_keep_names(u))

    if return_urls:
        return keep, len(urls), urls
    return keep, len(urls)


# ---------------------------------------------------------------------
# Watched-Status Cleanup (Hash-Dateien prunen wenn Medien verschwinden)
# ---------------------------------------------------------------------
_WATCHED_VIDEO_EXT_RE = re.compile(r"\.(mkv|mp4|avi|m4v|mov|ts|m2ts|wmv|webm|iso|strm)(\?|$)", re.IGNORECASE)
_WATCHED_EP_RE = re.compile(r"(?:^|[\\/])[^\\/]*S\d{1,2}E\d{1,2}[^\\/]*", re.IGNORECASE)


def _pick_existing_in_addon_data(candidates):
    for name in candidates:
        p = os.path.join(common.addon_data_dir, name)
        if _exists_any(p):
            return p
    return os.path.join(common.addon_data_dir, candidates[0])


def _load_watched_movies(path):
    if hasattr(common, "load_watched_status_movies"):
        return common.load_watched_status_movies(path)
    data = _read_json_file(path)
    return data if isinstance(data, dict) else {}


def _save_watched_movies(path, data):
    if hasattr(common, "save_watched_status_movies"):
        return common.save_watched_status_movies(path, data)
    # Fallback: compact dump
    try:
        with xbmcvfs.File(path, "w") as fh:
            fh.write(json.dumps(data, ensure_ascii=False, separators=(",", ":")))
        return True
    except Exception:
        return False


def _load_watched_series(path):
    if hasattr(common, "load_watched_status_series"):
        return common.load_watched_status_series(path)
    data = _read_json_file(path)
    return data if isinstance(data, dict) else {}


def _save_watched_series(path, data):
    if hasattr(common, "save_watched_status_series"):
        return common.save_watched_status_series(path, data)
    try:
        with xbmcvfs.File(path, "w") as fh:
            fh.write(json.dumps(data, ensure_ascii=False, separators=(",", ":")))
        return True
    except Exception:
        return False


def cleanup_watched_status_from_urls(urls_set):
    """
    Entfernt aus watched_status_series.json + watched_status_movies(.json)
    alle Hash-Keys, deren Medien-URL nicht mehr in den aktuellen JSONs vorkommt.

    Wichtig:
    - Hashes sind nicht umkehrbar, daher bestimmen wir die "keep"-Hashes,
      indem wir aus vorhandenen Medien-URLs die Hashes neu berechnen.
    - Failsafe: wenn wir keine keep-Hashes finden, löschen wir nichts.
    """
    try:
        if not urls_set:
            log("Watched-Cleanup: skip (urls_set leer).", xbmc.LOGINFO)
            return

        series_fp = os.path.join(common.addon_data_dir, "watched_status_series.json")
        movies_fp = _pick_existing_in_addon_data(["watched_status_movies.json", "watched_status_movie.json"])

        keep_movies = set()
        keep_series = set()

        # Build keep sets
        for u in urls_set:
            if not isinstance(u, str):
                continue
            url = u.strip()
            if not url or not (url.startswith("http://") or url.startswith("https://")):
                continue
            if not _WATCHED_VIDEO_EXT_RE.search(url):
                continue

            # Episode erkennen
            if _WATCHED_EP_RE.search(url):
                if hasattr(common, "series_watched_id"):
                    keep_series.add(common.series_watched_id(url))
            else:
                if hasattr(common, "movies_watched_id"):
                    keep_movies.add(common.movies_watched_id(url))

        removed_s = 0
        removed_m = 0

        # Series prune (nur wenn keep_series vorhanden)
        # Safety: wenn wir nur extrem wenige Episode-URLs finden (z.B. weil keine Episoden-Cachefiles vorhanden sind),
        # dann prunen wir NICHT, um keinen Watch-Status aus Versehen zu verlieren.
        if _exists_any(series_fp) and keep_series and len(keep_series) >= 3:
            watched_series = _load_watched_series(series_fp)
            before = len(watched_series)
            watched_series = {k: v for k, v in watched_series.items() if k in keep_series}
            removed_s = before - len(watched_series)
            if removed_s > 0:
                _save_watched_series(series_fp, watched_series)
        elif _exists_any(series_fp) and keep_series and len(keep_series) < 3:
            log(f"Watched-Cleanup: skip series prune (keep_series={len(keep_series)} zu klein).", xbmc.LOGINFO)

        # Movies prune (nur wenn keep_movies vorhanden)
        if _exists_any(movies_fp) and keep_movies:
            watched_movies = _load_watched_movies(movies_fp)
            before = len(watched_movies)
            watched_movies = {k: v for k, v in watched_movies.items() if k in keep_movies}
            removed_m = before - len(watched_movies)
            if removed_m > 0:
                _save_watched_movies(movies_fp, watched_movies)

        if (removed_s or removed_m):
            log(
                f"Watched-Cleanup: series_removed={removed_s} movies_removed={removed_m} "
                f"(keep_series={len(keep_series)} keep_movies={len(keep_movies)})",
                xbmc.LOGINFO
            )
        else:
            log(
                f"Watched-Cleanup: nothing to remove (keep_series={len(keep_series)} keep_movies={len(keep_movies)})",
                xbmc.LOGDEBUG
            )

    except Exception as e:
        log(f"Watched-Cleanup Fehler: {e}\n{traceback.format_exc()}", xbmc.LOGWARNING)


def run_orphan_artwork_cleanup(force=False, ignore_interval=False):
    try:
        if xbmc.getCondVisibility("Player.HasMedia"):
            log("Cache-Cleanup: skip (Player läuft)", xbmc.LOGINFO)
            return
    except Exception:
        pass

    now_ts = int(time.time())
    last_ts = _load_last_cleanup_ts()
    interval_sec = CACHE_CLEANUP_INTERVAL_HOURS * 3600

    if not force and not ignore_interval and last_ts and (now_ts - last_ts) < interval_sec:
        left = interval_sec - (now_ts - last_ts)
        log(f"Cache-Cleanup: skip (noch {left}s bis zum nächsten Lauf)", xbmc.LOGINFO)
        return

    win = None
    try:
        win = xbmcgui.Window(10000)
        if win.getProperty(CACHE_CLEANUP_PROP) == "1":
            log("Cache-Cleanup: skip (läuft bereits)", xbmc.LOGINFO)
            return
        win.setProperty(CACHE_CLEANUP_PROP, "1")
    except Exception:
        win = None

    # Startup-Schutz: Cleanup nicht innerhalb der ersten 3 Minuten nach Service-Start.
    # BG-Refresh-Threads schreiben JSONs asynchron – zu früh = keep_names leer = alles löschen!
    grace_sec = CACHE_CLEANUP_STARTUP_DELAY_SEC  # 3 Minuten
    uptime = int(time.time()) - SERVICE_START_TS
    if not force and uptime < grace_sec:
        log(f"Cache-Cleanup: skip (Service erst {uptime}s alt, grace={grace_sec}s)", xbmc.LOGINFO)
        if win:
            try:
                win.setProperty(CACHE_CLEANUP_PROP, "")
            except Exception:
                pass
        return

    log(f"Cache-Cleanup: gestartet (force={force}, last_ts={last_ts})", xbmc.LOGINFO)

    try:
        try:
            _prune_stale_series_details_artwork_sources()
        except Exception as e_prune:
            log(f"Cache-Cleanup: details-stale-prune failed: {e_prune}", xbmc.LOGWARNING)

        try:
            side_removed, side_scanned = _prune_orphan_json_sidecars()
            if side_scanned:
                log(
                    f"Cache-Cleanup: orphan-sidecars-prune done | removed={side_removed}/{side_scanned}",
                    xbmc.LOGINFO,
                )
        except Exception as e_side:
            log(f"Cache-Cleanup: orphan-sidecars-prune failed: {e_side}", xbmc.LOGWARNING)

        keep, url_count, urls = _collect_keep_names_from_all_jsons(return_urls=True)

        removed_total = 0
        scanned_total = 0
        seen_total = 0

        per_dir = {}

        for key, cdir in CACHE_DIRS.items():
            r, s, seen = _cleanup_cache_dir(cdir, keep)
            removed_total += r
            scanned_total += s
            seen_total += seen
            per_dir[key] = {"seen": seen, "scanned": s, "removed": r}

        try:
            cleanup_watched_status_from_urls(urls)
        except Exception as e:
            log(f"Watched-Cleanup failed: {e}", xbmc.LOGWARNING)

        _save_last_cleanup_ts(now_ts)

        log(
            f"Cache-Cleanup: done | urls_found={url_count} keep_names={len(keep)} "
            f"files_seen={seen_total} scanned={scanned_total} removed={removed_total} per_dir={per_dir}",
            xbmc.LOGINFO
        )

    except Exception as e:
        log(f"Cache-Cleanup Fehler: {e}\n{traceback.format_exc()}", xbmc.LOGWARNING)
    finally:
        try:
            if win:
                win.setProperty(CACHE_CLEANUP_PROP, "")
        except Exception:
            pass


# ---------------------------------------------------------------------
# Main loop
# ---------------------------------------------------------------------
def run_service():
    common.addon_handle = get_addon_handle_safe()

    try:
        if common.addon_data_dir and not xbmcvfs.exists(common.addon_data_dir):
            xbmcvfs.mkdirs(common.addon_data_dir)
    except Exception:
        pass

    # Cache dirs sicher anlegen (verhindert "exists false / walk nichts" Chaos)
    try:
        for _k, _d in CACHE_DIRS.items():
            if _d and not xbmcvfs.exists(_d):
                xbmcvfs.mkdirs(_d)
    except Exception:
        pass

    interval = _get_interval_seconds(3600)
    log(f"Service gestartet ({SERVICE_BUILD}), Intervall={interval}s")

    if common.addon_handle == -1:
        log("Service-Mode erkannt: kein Addon-Handle in sys.argv (ok).", xbmc.LOGDEBUG)

    try:
        changed, series_changed = background_refresh_all_once(do_series=True, do_movies=True, do_fav=True, head_series=False, head_movies=False)
        # Details/Artwork Prefetch im Hintergrund (ohne GUI Refresh)
        if series_changed:
            _start_series_prefetch_thread(force=True)
        else:
            details_cache_empty = (
                not os.path.isdir(SERIES_DETAILS_CACHE_DIR) or
                not any(f.endswith(".json") for f in os.listdir(SERIES_DETAILS_CACHE_DIR))
                if os.path.isdir(SERIES_DETAILS_CACHE_DIR) else True
            )
            if details_cache_empty:
                log("Service: Details-Cache leer -> details prefetch erzwungen (ignore_interval)", xbmc.LOGINFO)
                _start_series_prefetch_thread(force=False, ignore_interval=True)
            else:
                log("Service: Series list unchanged -> run details prefetch (throttled)", xbmc.LOGINFO)
                _start_series_prefetch_thread(force=False)

        # Movie- und Serien-Artwork proaktiv prefetchen (im Hintergrund)
        _start_movies_image_prefetch_thread()
        _start_series_image_prefetch_thread()

        if changed:
            _refresh_if_addon_visible()
    except Exception as e:
        log(f"Initial BG refresh failed: {e}", xbmc.LOGWARNING)

    # Cleanup beim Start: läuft NICHT sofort – der Main-Loop triggert ihn einmalig nach
    # CACHE_CLEANUP_STARTUP_DELAY_SEC (3 min), sobald die BG-Refresh-Threads fertig sind.
    # Der initiale Aufruf hier prüft nur ob der 24h-Intervall schon abgelaufen war
    # (z.B. nach sehr langem Kodi-Lauf) – grace_sec verhindert zu frühes Löschen.
    try:
        run_orphan_artwork_cleanup(force=False)
    except Exception as e:
        log(f"Initial Cache-Cleanup failed: {e}", xbmc.LOGWARNING)

    # Payload-Cache GC beim Start (force=True)
    try:
        run_payload_cache_gc(force=True)
    except Exception as e:
        log(f"Initial Payload-GC failed: {e}", xbmc.LOGWARNING)

    mon = xbmc.Monitor()
    next_periodic_ts = time.time() + max(5, int(interval))
    poll_s = 2
    # Track whether the one-shot startup cleanup has fired yet this session.
    _startup_cleanup_done = False

    while not mon.abortRequested():
        # Fast external trigger from series.py when list detects missing details.json.
        try:
            win = xbmcgui.Window(10000)
            trig = (win.getProperty(SERIES_PREFETCH_TRIGGER_PROP) or "").strip()
            if trig:
                win.clearProperty(SERIES_PREFETCH_TRIGGER_PROP)
                log("Service: external details prefetch trigger received", xbmc.LOGINFO)
                _start_series_prefetch_thread(force=False, ignore_interval=True)
        except Exception:
            pass

        # Mid-cycle trigger: movies.py detected a new movie and set this prop.
        try:
            win = xbmcgui.Window(10000)
            mv_trig = (win.getProperty(MOVIES_IMAGE_PREFETCH_TRIGGER_PROP) or "").strip()
            if mv_trig:
                win.clearProperty(MOVIES_IMAGE_PREFETCH_TRIGGER_PROP)
                log("Service: external movies image prefetch trigger received -> starting", xbmc.LOGINFO)
                _start_movies_image_prefetch_thread()
        except Exception:
            pass

        # One-shot startup cleanup: runs once ~3 min after boot (grace expires),
        # regardless of the 24h interval. After that, only the periodic loop fires it.
        if not _startup_cleanup_done:
            uptime = int(time.time()) - SERVICE_START_TS
            if uptime >= CACHE_CLEANUP_STARTUP_DELAY_SEC:
                _startup_cleanup_done = True
                log(f"Cache-Cleanup: Startup-Cleanup nach {uptime}s (einmalig, ignore_interval=True)", xbmc.LOGINFO)
                try:
                    run_orphan_artwork_cleanup(force=False, ignore_interval=True)
                except Exception as e:
                    log(f"Startup Cache-Cleanup failed: {e}", xbmc.LOGWARNING)

        now_ts = time.time()
        if now_ts >= next_periodic_ts:
            changed = False
            series_changed = False
            try:
                changed, series_changed = background_refresh_all_once(do_series=True, do_movies=True, do_fav=True, head_series=False, head_movies=False)
                # Prefetch gelegentlich laufen lassen (throttled)
                _start_series_prefetch_thread(force=False, ignore_interval=False)
                # Artwork-Prefetch wenn sich Daten geändert haben
                if changed:
                    _start_movies_image_prefetch_thread()
                if series_changed:
                    _start_series_image_prefetch_thread()
                if changed:
                    _refresh_if_addon_visible()
            except Exception as e:
                log(f"Periodic BG refresh failed: {e}\n{traceback.format_exc()}", xbmc.LOGWARNING)

            try:
                # Periodischer Cleanup läuft erst nachdem der Startup-Cleanup (nach 3 min)
                # schon einmal durch war – verhindert Doppellauf kurz nach dem Boot.
                if _startup_cleanup_done:
                    run_orphan_artwork_cleanup(force=bool(changed or series_changed))
            except Exception as e:
                log(f"Periodic Cache-Cleanup failed: {e}", xbmc.LOGWARNING)

            try:
                run_payload_cache_gc(force=False)
            except Exception as e:
                log(f"Periodic Payload-GC failed: {e}", xbmc.LOGWARNING)

            next_periodic_ts = now_ts + max(5, int(interval))

        if mon.waitForAbort(poll_s):
            break

    log("Service stopped")



def _load_last_payload_gc_ts():
    data = _read_json_file(PAYLOAD_GC_STATE_FILE)
    if isinstance(data, dict):
        try:
            return int(data.get("last_ts", 0))
        except Exception:
            return 0
    return 0


def _save_last_payload_gc_ts(ts_int):
    return _save_json_atomic(PAYLOAD_GC_STATE_FILE, {"last_ts": int(ts_int)}, indent=None)


def _mtime_any(path):
    """mtime für VFS + OS"""
    if not path:
        return 0
    try:
        st = xbmcvfs.Stat(path)
        mt_attr = getattr(st, "st_mtime", None)
        mt = mt_attr() if callable(mt_attr) else mt_attr
        return int(mt or 0)
    except Exception:
        pass
    try:
        lp = _translate_path(path)
        return int(os.path.getmtime(lp))
    except Exception:
        return 0

def _size_any(path):
    """Dateigröße in Bytes für VFS + OS; -1 wenn unbekannt."""
    if not path:
        return -1
    try:
        st = xbmcvfs.Stat(path)
        sz_attr = getattr(st, "st_size", None)
        sz = sz_attr() if callable(sz_attr) else sz_attr
        if sz is not None:
            return int(sz)
    except Exception:
        pass
    try:
        lp = _translate_path(path)
        return int(os.path.getsize(lp))
    except Exception:
        return -1




def _scrub_zero_byte_payload_dir(cache_dir):
    """Schneller Scrub: löscht 0-Byte *.json im payload_cache (ohne Alters-/Count-Regeln)."""
    removed = 0
    if not cache_dir or not _exists_any(cache_dir):
        return removed
    _dirs, files = _listdir_any(cache_dir)
    for fn in (files or []):
        if not fn:
            continue
        lfn = fn.lower()
        if not lfn.endswith(".json"):
            continue
        fp = _vfs_join(cache_dir, fn) if not os.path.isabs(fn) else fn
        try:
            if _size_any(fp) == 0:
                if _delete_file_any(fp, with_sidecars=True):
                    removed += 1
        except Exception:
            pass
    return removed

def _gc_payload_dir(cache_dir, now_ts, max_age_days=30, max_files=2000):
    """Löscht alte Payload-Cache-Dateien (.json) nach Alter und/oder count-limit."""
    removed = 0
    scanned = 0

    if not cache_dir or not _exists_any(cache_dir):
        return removed, scanned

    # VFS listdir first
    dirs, files = _listdir_any(cache_dir)
    candidates = []
    for fn in (files or []):
        if not fn:
            continue
        lfn = fn.lower()
        if not lfn.endswith(".json"):
            continue
        if lfn.endswith(".tmp") or lfn.endswith(".bak"):
            continue
        fp = _vfs_join(cache_dir, fn) if not os.path.isabs(fn) else fn
        # 0-Byte Files sofort entfernen (führen sonst zu "leeren Staffeln" via pid-cache)
        try:
            if _size_any(fp) == 0:
                if _delete_file_any(fp, with_sidecars=True):
                    removed += 1
                continue
        except Exception:
            pass

        mt = _mtime_any(fp)
        candidates.append((mt, fp))

    scanned = len(candidates)
    if not candidates:
        return removed, scanned

    max_age_sec = int(max_age_days) * 86400
    # 1) age-based
    keep = []
    for mt, fp in candidates:
        if mt and (now_ts - mt) > max_age_sec:
            if _delete_file_any(fp, with_sidecars=True):
                removed += 1
        else:
            keep.append((mt, fp))

    # 2) count-based (delete oldest)
    if int(max_files) > 0 and len(keep) > int(max_files):
        keep.sort(key=lambda t: (t[0] or 0))
        to_del = keep[: max(0, len(keep) - int(max_files))]
        for _mt, fp in to_del:
            if _delete_file_any(fp, with_sidecars=True):
                removed += 1

    return removed, scanned


def run_payload_cache_gc(force=False):
    """Payload-Cache GC (Favorites). Throttled: max alle PAYLOAD_GC_INTERVAL_HOURS."""
    # Player läuft? -> lieber skip (IO sparen)
    try:
        if xbmc.getCondVisibility("Player.HasMedia"):
            log("Payload-GC: skip (Player läuft)", xbmc.LOGDEBUG)
            return
    except Exception:
        pass


    # Immer (cheap): 0-Byte PID-Dateien scrubben, damit Navigation nicht "leere Staffel" zeigt
    try:
        scrubbed = 0
        for _k, _d in PAYLOAD_CACHE_DIRS.items():
            scrubbed += _scrub_zero_byte_payload_dir(_d)
        if scrubbed:
            log(f"Payload-Scrub: removed {scrubbed} 0-byte payload files", xbmc.LOGDEBUG)
    except Exception:
        pass

    now_ts = int(time.time())
    last_ts = _load_last_payload_gc_ts()
    interval_sec = int(PAYLOAD_GC_INTERVAL_HOURS) * 3600

    if not force and last_ts and (now_ts - last_ts) < interval_sec:
        return

    win = None
    try:
        win = xbmcgui.Window(10000)
        if win.getProperty(PAYLOAD_GC_PROP) == "1":
            return
        win.setProperty(PAYLOAD_GC_PROP, "1")
    except Exception:
        win = None

    removed_total = 0
    scanned_total = 0
    try:
        for key, pdir in PAYLOAD_CACHE_DIRS.items():
            r, s = _gc_payload_dir(pdir, now_ts, max_age_days=30, max_files=2000)
            removed_total += r
            scanned_total += s

        _save_last_payload_gc_ts(now_ts)

        log(f"Payload-GC: done | scanned={scanned_total} removed={removed_total}", xbmc.LOGINFO)
    except Exception as e:
        log(f"Payload-GC Fehler: {e}\n{traceback.format_exc()}", xbmc.LOGWARNING)
    finally:
        try:
            if win:
                win.setProperty(PAYLOAD_GC_PROP, "")
        except Exception:
            pass

if __name__ == "__main__":
    run_service()