# -*- coding: utf-8 -*-
"""Service leve do OnePlayVIP no startup do Kodi.

Mantém a limpeza conservadora de APKs e prepara o EPG XMLTV em background
quando o EPG está habilitado. A navegação do addon fica lendo o índice pronto,
sem baixar XMLTV pesado ao abrir categorias de canais.
Compatível com Kodi 19+ / Python 3.
"""
import hashlib
import html
import json
import os
import re
import time
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed

try:
    import xbmc
    import xbmcaddon
    import xbmcvfs
except Exception:
    xbmc = None
    xbmcaddon = None
    xbmcvfs = None

try:
    import requests
except Exception:
    requests = None

try:
    from resources.lib import epg_runtime, storage_runtime, proxy as proxy_runtime, dns as dns_runtime
    from resources.lib.server_config import DNS_LIST_B64, get_base_url as external_get_base_url, get_current_server_choice as external_get_current_server_choice
except Exception:
    epg_runtime = None
    storage_runtime = None
    proxy_runtime = None
    dns_runtime = None
    DNS_LIST_B64 = []
    external_get_base_url = None
    external_get_current_server_choice = None

try:
    from resources.lib import http_runtime
except Exception:
    http_runtime = None

ADDON_ID = 'plugin.video.oneplayvip'
USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36'
HEADERS = {'User-Agent': USER_AGENT}
TRANSIENT_HTTP_STATUS = {429, 500, 502, 503, 504}

APK_DOWNLOAD_DIRS = [
    '/storage/emulated/0/Download',
    '/sdcard/Download',
    '/storage/self/primary/Download',
]
APK_FILENAMES = [
    'OnePlayAlpha.apk',
    'OnePlayIbo.apk',
    'OnePlayXtream.apk',
]

EPG_CACHE_DAYS = 1
EPG_TTL = EPG_CACHE_DAYS * 24 * 3600
EPG_INDEX_VERSION = 'oneplayvip_epg_xcpro_window_v3_service_watchdog'
EPG_INDEX_WINDOW_DAYS = 2
MAX_EPG_INDEX_FILES = 8
XMLTV_DEFAULT_OFFSET_SECS = -(3 * 3600)
EPG_BOUNDARY_TOLERANCE_SECS = 300
# Vigia conservador: verifica de hora em hora; download real continua diário/condicional.
EPG_SERVICE_CHECK_INTERVAL_SECS = 60 * 60
# Pedido criado pela UI é consumido pelo service residente, que mantém o
# contexto do addon. Dois segundos deixam a ativação perceptível sem polling
# agressivo de disco/rede.
EPG_SERVICE_REQUEST_POLL_SECS = 2
# Alterações reais do EPG recebem uma única execução após a UI persistir os
# valores. O health-check do proxy não pode disparar este caminho.
EPG_SETTINGS_DEBOUNCE_SECS = 2.0
EPG_REFRESH_MARGIN_SECS = 2 * 3600
EPG_MIN_FUTURE_COVERAGE_SECS = 12 * 3600
EPG_MIN_REFRESH_INTERVAL_SECS = 60 * 60
# Servidor sem EPG é reavaliado periodicamente, mas nunca em loop curto.
EPG_FAILURE_RETRY_SECS = 60 * 60
# Exige cobertura do índice diário completo: Hoje + Amanhã.
EPG_REQUIRE_FULL_WINDOW = True

# Hidratação progressiva de sinopses VOD: controlada e fora da navegação.
VOD_DESC_HYDRATE_POLL_SECS = 5
VOD_DESC_HYDRATE_BATCH_SIZE = 10000
VOD_DESC_HYDRATE_MICROBATCH_MIN = 100
VOD_DESC_HYDRATE_MICROBATCH_DEFAULT = 200
VOD_DESC_HYDRATE_MICROBATCH_MAX = 300
VOD_DESC_HYDRATE_WORKERS_MIN = 2
VOD_DESC_HYDRATE_WORKERS_DEFAULT = 4
VOD_DESC_HYDRATE_WORKERS_MAX = 6
VOD_DESC_HYDRATE_TIMEOUT_SECS = 8
VOD_DESC_HYDRATE_MAX_ATTEMPTS = 2
# O service devolve o controle ao loop residente após poucos microblocos.
# Mantém workers/tamanho adaptativo aprovados, mas não segura EPG, proxy e
# séries durante uma fila VOD de milhares de itens.
VOD_DESC_HYDRATE_MAX_BLOCKS_PER_CYCLE = 2
VOD_DESC_EMPTY_RETRY_SECS = 6 * 3600
VOD_DESC_TTL_SECS = 7 * 24 * 3600
MAX_DESC_CACHE_ENTRIES = 50000
VOD_CATALOG_DISCOVERY_INTERVAL_SECS = 24 * 3600
VOD_CATALOG_DISCOVERY_RETRY_SECS = 30 * 60
VOD_CATALOG_QUEUE_TARGET = 30000
VOD_CATALOG_INDEX_MAX = 100000
# Regra deliberadamente binária e conservadora: qualquer sinopse válida no
# catálogo resumido comprova que o servidor já fornece metadados nativos.
# Somente catálogos com zero sinopses podem usar hidratação de detalhes.
METADATA_DETECTION_RULE = 'any_native_synopsis_v1'
NATIVE_METADATA_DISABLED_REASON = 'native_synopsis_detected_no_empty_lookup'
_TAG_RE = re.compile(r'<[^>]+>')

# Estado somente do processo service: evita repetição de mensagens e chamadas
# de stop quando ambos os proxies permanecem desligados.
_PROXY_SERVICE_STATE = {
    'enabled': None,
    'port': 0,
}

_VOD_HTTP_LOCAL = threading.local()


def _log(message, level=None):
    try:
        if xbmc is not None:
            xbmc.log('[{}] {}'.format(ADDON_ID, message), level or xbmc.LOGDEBUG)
    except Exception:
        pass


_SENSITIVE_QUERY_RE = re.compile(
    r'(?i)\b(username|password)=([^&\s]+)'
)
_URL_AUTH_RE = re.compile(
    r'(?i)(https?://)([^/@\s:]+):([^/@\s]+)@'
)


def _sanitize_log_error(error, username='', password=''):
    """Remove credenciais de exceções antes que alcancem o kodi.log."""
    try:
        text = str(error or '')
    except Exception:
        text = 'falha sem detalhe legível'
    text = _SENSITIVE_QUERY_RE.sub(lambda match: '{}=***'.format(match.group(1)), text)
    text = _URL_AUTH_RE.sub(r'\1***:***@', text)
    for secret in (username, password):
        try:
            secret = str(secret or '').strip()
        except Exception:
            secret = ''
        # Parâmetros curtos já são cobertos pelo padrão username/password.
        # Evita destruir palavras comuns do texto de diagnóstico.
        if len(secret) >= 4:
            text = text.replace(secret, '***')
    return text[:1600]


def _addon():
    if xbmcaddon is None:
        return None
    try:
        return xbmcaddon.Addon(ADDON_ID)
    except Exception:
        try:
            return xbmcaddon.Addon()
        except Exception:
            return None



def _log_i18n_runtime_probe(addon):
    """Registra a resolução real das labels pelo Kodi; não altera configurações."""
    try:
        if addon is None:
            _log('[I18N] Verificação não executada: addon indisponível.', getattr(xbmc, 'LOGWARNING', None))
            return
        category = str(addon.getLocalizedString(30000) or '').strip()
        proxy = str(addon.getLocalizedString(30001) or '').strip()
        stable = str(addon.getLocalizedString(30038) or '').strip()
        if category and proxy and stable:
            _log('[I18N] Labels resolvidas pelo Kodi: #30000="{}"; #30001="{}"; #30038="{}".'.format(category, proxy, stable), getattr(xbmc, 'LOGINFO', None))
        else:
            _log('[I18N] FALHA: uma ou mais labels de settings não foram resolvidas (#30000/#30001/#30038).', getattr(xbmc, 'LOGWARNING', None))
    except Exception as exc:
        _log('[I18N] Verificação finalizou com aviso: {}'.format(_sanitize_log_error(exc)), getattr(xbmc, 'LOGWARNING', None))

def _is_android():
    try:
        return bool(xbmc is not None and xbmc.getCondVisibility('System.Platform.Android'))
    except Exception:
        return False


def _path_exists(path):
    try:
        if xbmcvfs is not None and xbmcvfs.exists(path):
            return True
    except Exception:
        pass
    try:
        return os.path.exists(path)
    except Exception:
        return False


def _delete_file(path):
    try:
        if xbmcvfs is not None and xbmcvfs.exists(path):
            return bool(xbmcvfs.delete(path))
    except Exception:
        pass
    try:
        if os.path.exists(path):
            os.remove(path)
            return True
    except Exception:
        pass
    return False


def cleanup_downloaded_apks_on_startup():
    # Fora do Android/Android TV não existe instalador APK; evita varredura e log desnecessário.
    if not _is_android():
        return []

    removed = []
    seen = set()
    for folder in APK_DOWNLOAD_DIRS:
        if not folder or folder in seen:
            continue
        seen.add(folder)
        for filename in APK_FILENAMES:
            path = os.path.join(folder, filename)
            if _path_exists(path) and _delete_file(path):
                removed.append(path)
    if removed:
        _log('Limpeza de APK no startup: {} arquivo(s) removido(s).'.format(len(removed)), getattr(xbmc, 'LOGINFO', None))
    return removed


def _safe_requests_get(url, **kw):
    if requests is None:
        raise RuntimeError('requests indisponível no Kodi')
    kw.setdefault('headers', HEADERS)
    kw.setdefault('timeout', 30)
    retries = max(0, int(kw.pop('retries', 2)))
    backoff = max(0.2, float(kw.pop('backoff', 1.0)))
    if http_runtime is not None:
        try:
            session = http_runtime.get_thread_session(
                scope='service-api',
                headers=HEADERS,
                pool_connections=8,
                pool_maxsize=8,
            )
        except Exception:
            session = None
        if session is not None:
            return http_runtime.safe_get(
                url,
                session=session,
                retries=retries,
                backoff=backoff,
                retry_statuses=TRANSIENT_HTTP_STATUS,
                **kw,
            )

    # Fallback sem o runtime compartilhado: retenta somente falhas transitórias.
    last_exc = None
    for attempt in range(retries + 1):
        try:
            r = requests.get(url, **kw)
            if r.status_code in TRANSIENT_HTTP_STATUS and attempt < retries:
                try:
                    r.close()
                except Exception:
                    pass
                time.sleep(backoff * (2 ** attempt))
                continue
            r.raise_for_status()
            return r
        except Exception as exc:
            last_exc = exc
            status_code = None
            try:
                response = getattr(exc, 'response', None)
                if response is not None:
                    status_code = int(response.status_code)
            except Exception:
                status_code = None
            try:
                ssl_error = getattr(requests.exceptions, 'SSLError', ())
                connection_error = isinstance(exc, (requests.Timeout, requests.ConnectionError))
                if ssl_error and isinstance(exc, ssl_error):
                    connection_error = False
            except Exception:
                connection_error = False
            if attempt < retries and (
                status_code in TRANSIENT_HTTP_STATUS or connection_error
            ):
                time.sleep(backoff * (2 ** attempt))
                continue
            raise last_exc


def _get_profile_dir(addon):
    try:
        return xbmcvfs.translatePath(addon.getAddonInfo('profile'))
    except Exception:
        return os.path.join(os.path.expanduser('~'), '.kodi', 'userdata', 'addon_data', ADDON_ID)


def _get_username(addon):
    try:
        return (addon.getSetting('username') or '').strip()
    except Exception:
        return ''


def _get_password(addon):
    try:
        return (addon.getSetting('password') or '').strip()
    except Exception:
        return ''


def _get_enable_epg(addon):
    try:
        return (addon.getSetting('enable_epg') or 'false').lower() == 'true'
    except Exception:
        return False


def _epg_settings_signature(addon):
    """Assinatura mínima das configurações que realmente afetam o XMLTV."""
    if not _get_enable_epg(addon):
        return (False, '')
    try:
        return (True, _fingerprint(addon))
    except Exception:
        return (True, '')


def _epg_settings_change_requires_prepare(previous_signature, addon):
    """Retorna a assinatura atual e se a mudança exige preparar o EPG."""
    current_signature = _epg_settings_signature(addon)
    return current_signature, bool(current_signature != previous_signature and current_signature[0])


def _get_current_server_choice(addon):
    try:
        if external_get_current_server_choice:
            return external_get_current_server_choice(addon, dns_list=DNS_LIST_B64)
    except Exception:
        pass
    try:
        return max(0, int(addon.getSetting('server_choice') or '0'))
    except Exception:
        return 0


def _get_base_url(addon):
    try:
        if external_get_base_url:
            return (external_get_base_url(addon, dns_list=DNS_LIST_B64, log_fn=lambda message: _log(message, getattr(xbmc, 'LOGERROR', None))) or '').strip()
    except Exception:
        pass
    return ''


def _fingerprint(addon):
    base_url = (_get_base_url(addon) or '').strip().rstrip('/')
    username = _get_username(addon)
    password = _get_password(addon)
    raw = '{}|{}|{}'.format(base_url, username, password)
    return hashlib.sha1(raw.encode('utf-8')).hexdigest()


def _server_slot_scope_name(addon, server_idx=None):
    if storage_runtime is not None:
        try:
            return storage_runtime.server_slot_scope_name(lambda: _get_current_server_choice(addon), server_idx)
        except Exception:
            pass
    try:
        idx = _get_current_server_choice(addon) if server_idx is None else int(server_idx)
    except Exception:
        idx = 0
    if idx < 0:
        idx = 0
    return 'server_{}'.format(idx + 1)


def _server_slot_storage_path(profile_dir, addon, filename, server_idx=None):
    if storage_runtime is not None:
        try:
            return storage_runtime.server_slot_storage_path(profile_dir, filename, lambda: _get_current_server_choice(addon), server_idx)
        except Exception:
            pass
    return os.path.join(profile_dir, 'server_slot_storage', _server_slot_scope_name(addon, server_idx), filename)


def _load_json(path, default):
    try:
        if storage_runtime is not None:
            return storage_runtime.load_json_file(path, default)
    except Exception:
        pass
    try:
        with open(path, 'r', encoding='utf-8') as fh:
            return json.load(fh)
    except Exception:
        return default


def _atomic_write_json(path, data):
    try:
        if storage_runtime is not None:
            storage_runtime.ensure_parent_dir(path)
            return storage_runtime.atomic_write_json(path, data, ensure_ascii=False, indent=2)
    except Exception:
        pass
    folder = os.path.dirname(path)
    if folder and not os.path.isdir(folder):
        os.makedirs(folder, exist_ok=True)
    tmp = path + '.tmp'
    with open(tmp, 'w', encoding='utf-8') as fh:
        json.dump(data, fh, ensure_ascii=False, indent=2)
    os.replace(tmp, path)
    return True


def _clean_plot(text):
    value = html.unescape(str(text or ''))
    return _TAG_RE.sub('', value).strip()


def _dedupe_catalog_records(records):
    """Mantém um registro por ID e preserva a melhor evidência de sinopse.

    Alguns painéis Xtream repetem milhares de linhas com o mesmo stream_id.
    Contar essas linhas como itens reais infla a cobertura e cria backlogs
    descartáveis. A primeira posição continua estável; uma duplicata só a
    substitui quando acrescenta uma sinopse válida.
    """
    unique = []
    positions = {}
    duplicate_total = 0
    for record in records or []:
        if not isinstance(record, tuple) or len(record) < 3:
            continue
        item_id = str(record[1] or '').strip()
        if not item_id:
            continue
        if item_id not in positions:
            positions[item_id] = len(unique)
            unique.append(record)
            continue
        duplicate_total += 1
        old_pos = positions[item_id]
        old_record = unique[old_pos]
        if not str(old_record[2] or '').strip() and str(record[2] or '').strip():
            unique[old_pos] = record
    return unique, duplicate_total


def _search_result_url_is_rebuildable(item):
    if not isinstance(item, dict):
        return False
    kind = str(item.get('stream_kind') or '').strip().lower()
    stream_id = str(item.get('stream_id') or '').strip()
    if kind == 'live':
        return bool(stream_id)
    if kind == 'vod':
        return bool(str(item.get('vod_id') or stream_id).strip())
    if kind == 'episode':
        return bool(stream_id)
    return False


def _sanitize_persisted_search_state(data):
    """Retira URLs autenticadas legadas sem quebrar a retomada da busca."""
    if not isinstance(data, dict):
        return data, False
    clean = dict(data)
    results = clean.get('last_results')
    if not isinstance(results, list):
        return clean, False
    changed = False
    clean_results = []
    for entry in results:
        if not isinstance(entry, dict):
            clean_results.append(entry)
            continue
        safe_entry = dict(entry)
        if _search_result_url_is_rebuildable(safe_entry) and 'url' in safe_entry:
            safe_entry.pop('url', None)
            changed = True
        clean_results.append(safe_entry)
    clean['last_results'] = clean_results
    return clean, changed


def scrub_legacy_search_state_urls(addon=None, profile_dir=None):
    """Migra todos os slots locais e elimina cópias desnecessárias do login."""
    addon = addon or _addon()
    profile_dir = profile_dir or _get_profile_dir(addon)
    root = os.path.join(profile_dir, 'server_slot_storage')
    if not os.path.isdir(root):
        return 0
    changed_files = 0
    try:
        slot_names = os.listdir(root)
    except Exception:
        return 0
    for slot_name in slot_names:
        if not re.match(r'^server_\d+$', str(slot_name or '')):
            continue
        path = os.path.join(root, slot_name, 'search_state.json')
        if not os.path.isfile(path):
            continue
        original = _load_json(path, {})
        clean, changed = _sanitize_persisted_search_state(original)
        if changed:
            _atomic_write_json(path, clean)
            changed_files += 1
    if changed_files:
        _log(
            'Migração de privacidade concluída em {} estado(s) de pesquisa.'.format(changed_files),
            getattr(xbmc, 'LOGINFO', None),
        )
    return changed_files


def scrub_legacy_epg_meta_credentials(addon=None, profile_dir=None):
    """Higieniza o último erro EPG já gravado em todos os slots locais."""
    addon = addon or _addon()
    profile_dir = profile_dir or _get_profile_dir(addon)
    root = os.path.join(profile_dir, 'server_slot_storage')
    if not os.path.isdir(root):
        return 0
    username = _get_username(addon)
    password = _get_password(addon)
    changed_files = 0
    try:
        slot_names = os.listdir(root)
    except Exception:
        return 0
    for slot_name in slot_names:
        if not re.match(r'^server_\d+$', str(slot_name or '')):
            continue
        path = os.path.join(root, slot_name, 'epg_xmltv_meta.json')
        if not os.path.isfile(path):
            continue
        meta = _load_json(path, {})
        if not isinstance(meta, dict):
            continue
        old_error = str(meta.get('last_error') or '')
        safe_error = _sanitize_log_error(old_error, username, password)
        if safe_error != old_error:
            clean = dict(meta)
            clean['last_error'] = safe_error
            _atomic_write_json(path, clean)
            changed_files += 1
    if changed_files:
        _log(
            'Migração de privacidade higienizou {} diagnóstico(s) EPG.'.format(changed_files),
            getattr(xbmc, 'LOGINFO', None),
        )
    return changed_files



def _decode_json_response_conservative(response):
    """
    Decodifica JSON preservando UTF-8 quando possível e só tenta encodings
    alternativos quando os bytes brutos justificam. Nunca "adivinha" letras
    que já chegaram literalmente como '?' do servidor.
    """
    raw = response.content or b''
    candidates = []
    declared = ''
    try:
        declared = str(response.encoding or '').strip()
    except Exception:
        declared = ''
    if declared:
        candidates.append(declared)
    candidates.extend(['utf-8-sig', 'utf-8', 'cp1252', 'latin-1'])

    seen = set()
    best = None
    best_score = None
    for encoding in candidates:
        key = encoding.lower()
        if key in seen:
            continue
        seen.add(key)
        try:
            text = raw.decode(encoding, errors='strict')
            payload = json.loads(text)
        except Exception:
            continue
        # Penaliza mojibake clássico e replacement chars. '?' literal não é
        # alterado porque pode ter vindo assim do upstream.
        score = (
            text.count('\ufffd') * 1000
            + text.count('Ã') * 20
            + text.count('Â') * 20
            + text.count('â€') * 20
        )
        if best is None or score < best_score:
            best = payload
            best_score = score
    if best is not None:
        return best
    return response.json()


def _extract_vod_plot(payload):
    if not isinstance(payload, dict):
        return ''
    info = payload.get('info') or {}
    movie_data = payload.get('movie_data') or {}
    return _clean_plot(
        info.get('plot') or info.get('description') or
        movie_data.get('plot') or movie_data.get('description') or ''
    )


def _trim_vod_desc_cache(cache):
    if not isinstance(cache, dict):
        return {}
    now_ts = time.time()
    items = []
    for key, entry in cache.items():
        if not isinstance(entry, dict):
            continue
        try:
            fetched_at = float(entry.get('fetched_at') or 0)
        except Exception:
            fetched_at = 0
        if fetched_at and (now_ts - fetched_at) > VOD_DESC_TTL_SECS:
            continue
        items.append((str(key), {'plot': str(entry.get('plot') or ''), 'fetched_at': fetched_at}))
    items.sort(key=lambda item: item[1].get('fetched_at') or 0, reverse=True)
    return dict(items[:MAX_DESC_CACHE_ENTRIES])


def _vod_desc_cache_has_fresh_entry(cache, vod_id):
    entry = cache.get(str(vod_id)) if isinstance(cache, dict) else None
    if not isinstance(entry, dict):
        return False
    try:
        fetched_at = float(entry.get('fetched_at') or 0)
    except Exception:
        fetched_at = 0
    if not fetched_at:
        return False
    plot = str(entry.get('plot') or '')
    ttl = VOD_DESC_TTL_SECS if plot else VOD_DESC_EMPTY_RETRY_SECS
    return (time.time() - fetched_at) < ttl



def _vod_worker_session():
    """Mantém uma Session HTTP isolada por worker para reaproveitar keep-alive."""
    if http_runtime is not None:
        try:
            return http_runtime.get_thread_session(
                scope='service-vod-worker',
                headers=HEADERS,
                pool_connections=VOD_DESC_HYDRATE_WORKERS_MAX,
                pool_maxsize=VOD_DESC_HYDRATE_WORKERS_MAX,
            )
        except Exception:
            pass
    session = getattr(_VOD_HTTP_LOCAL, 'session', None)
    if session is None:
        session = requests.Session()
        session.headers.update(HEADERS)
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=VOD_DESC_HYDRATE_WORKERS_MAX,
            pool_maxsize=VOD_DESC_HYDRATE_WORKERS_MAX,
            max_retries=0,
            pool_block=True,
        )
        session.mount('http://', adapter)
        session.mount('https://', adapter)
        _VOD_HTTP_LOCAL.session = session
    return session


def _hydrate_vod_entry(entry, base_url, username, password):
    """Consulta um VOD; não altera fila/cache compartilhados dentro do worker."""
    vod_id = str(entry.get('vod_id') or '').strip()
    attempts = int(entry.get('attempts') or 0)
    if not vod_id:
        return {'status': 'invalid', 'entry': entry, 'vod_id': ''}
    url = '{}/player_api.php?username={}&password={}&action=get_vod_info&vod_id={}'.format(
        base_url, username, password, vod_id
    )
    request_started_at = time.time()
    try:
        session = _vod_worker_session()
        response = session.get(
            url,
            timeout=VOD_DESC_HYDRATE_TIMEOUT_SECS,
            allow_redirects=True,
        )
        try:
            response.raise_for_status()
            payload = _decode_json_response_conservative(response)
        finally:
            try:
                response.close()
            except Exception:
                pass
        return {
            'status': 'ok',
            'entry': entry,
            'vod_id': vod_id,
            'plot': _extract_vod_plot(payload),
            'fetched_at': time.time(),
            'duration': max(0.0, time.time() - request_started_at),
        }
    except Exception as exc:
        attempts += 1
        retry_entry = dict(entry)
        retry_entry['attempts'] = attempts
        retry_entry['last_attempt_at'] = time.time()
        return {
            'status': 'retry' if attempts < VOD_DESC_HYDRATE_MAX_ATTEMPTS else 'failed',
            'entry': retry_entry,
            'vod_id': vod_id,
            'error': _sanitize_log_error(exc, username, password),
            'duration': max(0.0, time.time() - request_started_at),
        }



def _clamp_int(value, minimum, maximum, default):
    try:
        value = int(value)
    except Exception:
        value = int(default)
    return max(int(minimum), min(int(maximum), value))


def _load_vod_tuning(path):
    data = _load_json(path, {})
    if not isinstance(data, dict):
        data = {}
    return {
        'workers': _clamp_int(
            data.get('workers'),
            VOD_DESC_HYDRATE_WORKERS_MIN,
            VOD_DESC_HYDRATE_WORKERS_MAX,
            VOD_DESC_HYDRATE_WORKERS_DEFAULT,
        ),
        'microbatch': _clamp_int(
            data.get('microbatch'),
            VOD_DESC_HYDRATE_MICROBATCH_MIN,
            VOD_DESC_HYDRATE_MICROBATCH_MAX,
            VOD_DESC_HYDRATE_MICROBATCH_DEFAULT,
        ),
        'stable_blocks': max(0, int(data.get('stable_blocks') or 0)),
        'updated_at': float(data.get('updated_at') or 0),
    }


def _adapt_vod_tuning(tuning, block_processed, block_failed, block_elapsed, durations):
    """Ajusta devagar, com histerese, evitando oscilações e sobrecarga da API."""
    workers = _clamp_int(
        tuning.get('workers'),
        VOD_DESC_HYDRATE_WORKERS_MIN,
        VOD_DESC_HYDRATE_WORKERS_MAX,
        VOD_DESC_HYDRATE_WORKERS_DEFAULT,
    )
    microbatch = _clamp_int(
        tuning.get('microbatch'),
        VOD_DESC_HYDRATE_MICROBATCH_MIN,
        VOD_DESC_HYDRATE_MICROBATCH_MAX,
        VOD_DESC_HYDRATE_MICROBATCH_DEFAULT,
    )
    stable_blocks = max(0, int(tuning.get('stable_blocks') or 0))
    failure_rate = (float(block_failed) / float(block_processed)) if block_processed else 0.0
    throughput = (float(block_processed) / max(0.001, float(block_elapsed))) if block_processed else 0.0
    avg_latency = (sum(durations) / float(len(durations))) if durations else 0.0

    # Qualquer sinal forte de pressão reduz imediatamente; subida exige 3 blocos estáveis.
    if failure_rate >= 0.03 or avg_latency >= 1.5:
        workers = max(VOD_DESC_HYDRATE_WORKERS_MIN, workers - 1)
        microbatch = max(VOD_DESC_HYDRATE_MICROBATCH_MIN, microbatch - 50)
        stable_blocks = 0
    elif failure_rate == 0 and avg_latency <= 0.55 and throughput >= max(8.0, workers * 2.5):
        stable_blocks += 1
        if stable_blocks >= 3:
            if workers < VOD_DESC_HYDRATE_WORKERS_MAX:
                workers += 1
            elif microbatch < VOD_DESC_HYDRATE_MICROBATCH_MAX:
                microbatch += 50
            stable_blocks = 0
    else:
        # Zona neutra: mantém perfil; não caça desempenho por ruído de um bloco.
        stable_blocks = max(0, stable_blocks - 1)

    return {
        'workers': workers,
        'microbatch': microbatch,
        'stable_blocks': stable_blocks,
        'last_failure_rate': failure_rate,
        'last_throughput': throughput,
        'last_avg_latency': avg_latency,
        'updated_at': time.time(),
    }


def _vod_priority_key(entry):
    """Maior prioridade primeiro; em empate, preserva a ordem mais antiga."""
    try:
        priority = int(entry.get('priority') or 0)
    except Exception:
        priority = 0
    try:
        queued_at = float(entry.get('queued_at') or 0)
    except Exception:
        queued_at = 0.0
    return (-priority, queued_at)




def _extract_catalog_year(item, title=''):
    raw = ' '.join([
        str((item or {}).get('year') or ''),
        str((item or {}).get('releasedate') or ''),
        str((item or {}).get('releaseDate') or ''),
        str(title or ''),
    ])
    match = re.search(r'\b((?:19|20)\d{2})\b', raw)
    return match.group(1) if match else ''


def _vod_discovery_paths(profile_dir, addon):
    return {
        'meta': _server_slot_storage_path(profile_dir, addon, 'vod_catalog_discovery_meta.json'),
        'backlog': _server_slot_storage_path(profile_dir, addon, 'vod_catalog_discovery_backlog.json'),
        'index': _server_slot_storage_path(profile_dir, addon, 'vod_catalog_index.json'),
        'queue': _server_slot_storage_path(profile_dir, addon, 'vod_desc_hydration_queue.json'),
        'cache': _server_slot_storage_path(profile_dir, addon, 'vod_desc_cache.json'),
        'profile': _server_slot_storage_path(profile_dir, addon, 'vod_metadata_profile.json'),
        'prefetch': _server_slot_storage_path(profile_dir, addon, 'vod_catalog_prefetch_request.json'),
    }


def _promote_profile_with_native_synopsis(profile_path):
    """Migra perfis antigos sem permitir que filas legadas alcancem a API."""
    profile = _load_json(profile_path, {})
    if not isinstance(profile, dict):
        profile = {}
    try:
        synopsis_total = int(profile.get('synopsis_total') or 0)
    except Exception:
        synopsis_total = 0

    changed = False
    if synopsis_total > 0 or profile.get('mode') == 'native_metadata':
        if profile.get('mode') != 'native_metadata':
            profile['mode'] = 'native_metadata'
            changed = True
        if profile.get('detection_rule') != METADATA_DETECTION_RULE:
            profile['detection_rule'] = METADATA_DETECTION_RULE
            changed = True
        if 'threshold' in profile:
            profile.pop('threshold', None)
            changed = True
        if changed:
            profile['migrated_at'] = time.time()
            _atomic_write_json(profile_path, profile)
    return profile, changed


def _profile_has_native_synopsis_evidence(profile):
    if not isinstance(profile, dict):
        return False
    if profile.get('mode') == 'native_metadata':
        return True
    try:
        return int(profile.get('synopsis_total') or 0) > 0
    except Exception:
        return False


def _clear_native_vod_hydration_state(addon, profile_dir):
    """Remove uma única vez resíduos incompatíveis com o modo nativo."""
    paths = _vod_discovery_paths(profile_dir, addon)
    now_ts = time.time()
    cleared = 0

    queue = _load_json(paths['queue'], {})
    queue_pending = queue.get('pending') if isinstance(queue, dict) else None
    if isinstance(queue_pending, list) and queue_pending:
        cleared += len(queue_pending)
        _atomic_write_json(paths['queue'], {'pending': [], 'updated_at': now_ts})

    backlog = _load_json(paths['backlog'], {})
    backlog_pending = backlog.get('pending') if isinstance(backlog, dict) else None
    backlog_reason = backlog.get('disabled_reason') if isinstance(backlog, dict) else ''
    if (isinstance(backlog_pending, list) and backlog_pending) or (
        isinstance(backlog, dict) and backlog and backlog_reason != NATIVE_METADATA_DISABLED_REASON
    ):
        if isinstance(backlog_pending, list):
            cleared += len(backlog_pending)
        _atomic_write_json(paths['backlog'], {
            'pending': [],
            'updated_at': now_ts,
            'disabled_reason': NATIVE_METADATA_DISABLED_REASON,
        })

    prefetch = _load_json(paths['prefetch'], {})
    categories = prefetch.get('categories') if isinstance(prefetch, dict) else None
    if isinstance(categories, list) and categories:
        cleared += len(categories)
        prefetch['categories'] = []
        prefetch['cursor'] = 0
        prefetch['updated_at'] = now_ts
        prefetch['disabled_reason'] = NATIVE_METADATA_DISABLED_REASON
        _atomic_write_json(paths['prefetch'], prefetch)
    return cleared


def _catalog_scan_due(meta, now_ts):
    if not isinstance(meta, dict):
        return True
    try:
        last_success = float(meta.get('last_success_at') or 0)
        last_attempt = float(meta.get('last_attempt_at') or 0)
    except Exception:
        return True
    if last_success and (now_ts - last_success) < VOD_CATALOG_DISCOVERY_INTERVAL_SECS:
        return False
    if last_attempt and (now_ts - last_attempt) < VOD_CATALOG_DISCOVERY_RETRY_SECS:
        return False
    return True


def _scan_full_vod_catalog(addon, profile_dir, base_url, username, password):
    """Classifica o servidor antes de ativar hidratação em massa."""
    paths = _vod_discovery_paths(profile_dir, addon)
    now_ts = time.time()
    meta = _load_json(paths['meta'], {})
    if not _catalog_scan_due(meta, now_ts):
        profile = _load_json(paths['profile'], {})
        return {
            'scanned': False,
            'reason': 'not_due',
            'mode': profile.get('mode') if isinstance(profile, dict) else '',
        }
    if _is_video_playing():
        return {'scanned': False, 'reason': 'playing'}

    meta = meta if isinstance(meta, dict) else {}
    meta['last_attempt_at'] = now_ts
    _atomic_write_json(paths['meta'], meta)

    url = '{}/player_api.php?username={}&password={}&action=get_vod_streams'.format(
        base_url, username, password
    )
    response = _safe_requests_get(url, timeout=30, retries=1, backoff=1.0)
    try:
        payload = _decode_json_response_conservative(response)
    finally:
        try:
            response.close()
        except Exception:
            pass
    if not isinstance(payload, list):
        raise RuntimeError('catálogo VOD resumido não retornou uma lista')

    cache = _load_json(paths['cache'], {})
    if not isinstance(cache, dict):
        cache = {}
    old_index = _load_json(paths['index'], {})
    if not isinstance(old_index, dict):
        old_index = {}

    valid_items = []
    synopsis_count = 0
    for item in payload:
        if not isinstance(item, dict):
            continue
        vod_id = str(item.get('stream_id') or item.get('vod_id') or '').strip()
        if not vod_id:
            continue
        info = item.get('info') or {}
        plot = _clean_plot(
            item.get('plot')
            or item.get('description')
            or item.get('overview')
            or info.get('plot')
            or info.get('description')
            or info.get('overview')
            or ''
        )
        valid_items.append((item, vod_id, plot))
        if plot:
            synopsis_count += 1

    valid_items, duplicate_total = _dedupe_catalog_records(valid_items)
    synopsis_count = sum(1 for _item, _vod_id, plot in valid_items if plot)
    sample_total = len(valid_items)
    coverage = (float(synopsis_count) / float(sample_total)) if sample_total else 0.0
    previous_profile = _load_json(paths['profile'], {})
    previous_evidence = _profile_has_native_synopsis_evidence(previous_profile)
    try:
        previous_synopsis_total = int((previous_profile or {}).get('synopsis_total') or 0)
    except Exception:
        previous_synopsis_total = 0
    native_mode = synopsis_count > 0 or previous_evidence
    mode = 'native_metadata' if native_mode else 'hydration_required'
    _atomic_write_json(paths['profile'], {
        'mode': mode,
        'coverage': coverage,
        'sample_total': sample_total,
        'synopsis_total': max(previous_synopsis_total, synopsis_count),
        'last_scan_synopsis_total': synopsis_count,
        'duplicate_rows_ignored': duplicate_total,
        'detection_rule': METADATA_DETECTION_RULE,
        'classified_at': now_ts,
    })

    catalog_index = {}
    total = 0
    new_count = 0

    # Uma única sinopse nativa já desativa toda consulta de detalhe em massa.
    # Vazios do resumo são tratados como vazios reais nesse servidor.
    if native_mode:
        for item, vod_id, plot in valid_items:
            total += 1
            title = html.unescape(str(item.get('name') or item.get('title') or '')).strip()
            year = _extract_catalog_year(item, title)
            old_entry = old_index.get(vod_id) if isinstance(old_index.get(vod_id), dict) else {}
            if not old_entry:
                new_count += 1
            catalog_index[vod_id] = {
                'title': title[:160],
                'identity_title': _normalize_vod_identity_title(title),
                'year': year,
                'first_seen_at': float(old_entry.get('first_seen_at') or now_ts),
                'last_seen_at': now_ts,
                'native_plot_available': bool(plot),
            }

        _atomic_write_json(paths['index'], catalog_index)
        _atomic_write_json(paths['queue'], {'pending': [], 'updated_at': now_ts})
        _atomic_write_json(paths['backlog'], {
            'pending': [],
            'updated_at': now_ts,
            'catalog_total': total,
            'disabled_reason': NATIVE_METADATA_DISABLED_REASON,
        })
        _clear_native_vod_hydration_state(addon, profile_dir)
        meta.update({
            'last_success_at': now_ts,
            'mode': mode,
            'coverage': coverage,
            'catalog_total': total,
            'cached_total': 0,
            'already_queued_total': 0,
            'backlog_total': 0,
            'new_total': new_count,
            'removed_total': max(0, len(old_index) - len(catalog_index)),
            'duplicate_rows_ignored': duplicate_total,
        })
        _atomic_write_json(paths['meta'], meta)
        return {
            'scanned': True,
            'mode': mode,
            'coverage': coverage,
            'synopsis_total': synopsis_count,
            'total': total,
            'cached': 0,
            'queued': 0,
            'backlog': 0,
            'new': new_count,
            'duplicates_ignored': duplicate_total,
        }

    queue = _load_json(paths['queue'], {})
    pending = queue.get('pending') if isinstance(queue, dict) else []
    if not isinstance(pending, list):
        pending = []
    queued_ids = {
        str(entry.get('vod_id') or '').strip()
        for entry in pending if isinstance(entry, dict)
    }

    backlog = []
    cached = queued = 0
    for item, vod_id, plot in valid_items:
        total += 1
        title = html.unescape(str(item.get('name') or item.get('title') or '')).strip()
        year = _extract_catalog_year(item, title)
        old_entry = old_index.get(vod_id) if isinstance(old_index.get(vod_id), dict) else {}
        is_new = not bool(old_entry)
        if is_new:
            new_count += 1
        catalog_index[vod_id] = {
            'title': title[:160],
            'identity_title': _normalize_vod_identity_title(title),
            'year': year,
            'first_seen_at': float(old_entry.get('first_seen_at') or now_ts),
            'last_seen_at': now_ts,
        }
        if plot:
            continue
        if _vod_desc_cache_has_fresh_entry(cache, vod_id):
            cached += 1
            continue
        if vod_id in queued_ids:
            queued += 1
            continue
        backlog.append({
            'vod_id': vod_id,
            'title': title[:160],
            'identity_title': _normalize_vod_identity_title(title),
            'year': year,
            'queued_at': now_ts,
            'attempts': 0,
            'priority': 80 if is_new else 30,
            'source': 'catalog_new' if is_new else 'catalog_discovery',
        })

    backlog.sort(key=_vod_priority_key)
    if len(catalog_index) > VOD_CATALOG_INDEX_MAX:
        catalog_index = dict(sorted(
            catalog_index.items(),
            key=lambda kv: float((kv[1] or {}).get('last_seen_at') or 0),
            reverse=True
        )[:VOD_CATALOG_INDEX_MAX])

    _atomic_write_json(paths['index'], catalog_index)
    _atomic_write_json(paths['backlog'], {
        'pending': backlog,
        'updated_at': now_ts,
        'catalog_total': total,
    })
    meta.update({
        'last_success_at': now_ts,
        'mode': mode,
        'coverage': coverage,
        'catalog_total': total,
        'cached_total': cached,
        'already_queued_total': queued,
        'backlog_total': len(backlog),
        'new_total': new_count,
        'removed_total': max(0, len(old_index) - len(catalog_index)),
        'duplicate_rows_ignored': duplicate_total,
    })
    _atomic_write_json(paths['meta'], meta)
    return {
        'scanned': True,
        'mode': mode,
        'coverage': coverage,
        'total': total,
        'cached': cached,
        'queued': queued,
        'backlog': len(backlog),
        'new': new_count,
        'duplicates_ignored': duplicate_total,
    }

def _refill_vod_queue_from_backlog(addon, profile_dir):
    paths = _vod_discovery_paths(profile_dir, addon)
    queue = _load_json(paths['queue'], {})
    if not isinstance(queue, dict):
        queue = {}
    pending = queue.get('pending') if isinstance(queue.get('pending'), list) else []
    backlog_data = _load_json(paths['backlog'], {})
    backlog = backlog_data.get('pending') if isinstance(backlog_data, dict) else []
    if not isinstance(backlog, list) or not backlog:
        return 0

    existing = {
        str(entry.get('vod_id') or '').strip()
        for entry in pending if isinstance(entry, dict)
    }
    capacity = max(0, VOD_CATALOG_QUEUE_TARGET - len(pending))
    if capacity <= 0:
        return 0

    moved, remaining = [], []
    backlog_pruned = False
    for entry in backlog:
        if not isinstance(entry, dict):
            backlog_pruned = True
            continue
        vod_id = str(entry.get('vod_id') or '').strip()
        if not vod_id or vod_id in existing:
            backlog_pruned = True
            continue
        if len(moved) < capacity:
            moved.append(entry)
            existing.add(vod_id)
        else:
            remaining.append(entry)

    if not moved:
        if backlog_pruned:
            backlog_data = backlog_data if isinstance(backlog_data, dict) else {}
            backlog_data['pending'] = remaining
            backlog_data['updated_at'] = time.time()
            _atomic_write_json(paths['backlog'], backlog_data)
        return 0
    pending.extend(moved)
    pending.sort(key=_vod_priority_key)
    queue['pending'] = pending[:VOD_CATALOG_QUEUE_TARGET]
    queue['updated_at'] = time.time()
    backlog_data = backlog_data if isinstance(backlog_data, dict) else {}
    backlog_data['pending'] = remaining
    backlog_data['updated_at'] = time.time()
    _atomic_write_json(paths['queue'], queue)
    _atomic_write_json(paths['backlog'], backlog_data)
    return len(moved)


def ensure_vod_catalog_discovery(addon, profile_dir, base_url, username, password):
    result = {'scanned': False}
    try:
        result = _scan_full_vod_catalog(
            addon, profile_dir, base_url, username, password
        )
        if result.get('scanned'):
            mode = result.get('mode') or 'hydration_required'
            coverage = float(result.get('coverage') or 0) * 100.0
            if mode == 'native_metadata':
                _log(
                    'VOD modo nativo: sinopse nativa detectada neste servidor; '
                    'nenhum item vazio será consultado.',
                    getattr(xbmc, 'LOGINFO', None)
                )
            else:
                _log(
                    'Descoberta automática VOD: {} total, cobertura {:.1f}%, '
                    '{} novo(s), {} aguardando hidratação.'.format(
                        result.get('total', 0),
                        coverage,
                        result.get('new', 0),
                        result.get('backlog', 0),
                    ),
                    getattr(xbmc, 'LOGINFO', None)
                )
    except Exception as exc:
        _log(
            'Descoberta automática VOD finalizou com aviso: {}.'.format(
                _sanitize_log_error(exc, username, password)
            ),
            getattr(xbmc, 'LOGDEBUG', None)
        )

    profile_path = _server_slot_storage_path(
        profile_dir, addon, 'vod_metadata_profile.json'
    )
    profile, promoted = _promote_profile_with_native_synopsis(profile_path)
    mode = profile.get('mode') if isinstance(profile, dict) else ''
    if mode == 'native_metadata':
        cleared = _clear_native_vod_hydration_state(addon, profile_dir)
        if cleared:
            _log(
                'Modo nativo VOD removeu {} pendência(s) legada(s) sem consultar a API.'.format(cleared),
                getattr(xbmc, 'LOGDEBUG', None)
            )
        result['moved'] = 0
        result['cleared'] = cleared
        result['profile_promoted'] = promoted
        return result
    moved = _refill_vod_queue_from_backlog(addon, profile_dir)
    if moved:
        _log(
            'Fila VOD reabastecida automaticamente: {} item(ns).'.format(moved),
            getattr(xbmc, 'LOGDEBUG', None)
        )
    result['moved'] = moved
    return result


def _normalize_vod_identity_title(value):
    value = html.unescape(str(value or '')).strip().lower()
    value = re.sub(r'\[[^\]]+\]', ' ', value)
    value = re.sub(r'\([^)]*(?:dublado|legendado|dual|4k|hd|fhd|uhd)[^)]*\)', ' ', value, flags=re.I)
    value = re.sub(r'[^a-z0-9\u00c0-\u024f]+', ' ', value, flags=re.I)
    return re.sub(r'\s+', ' ', value).strip()


def _shared_vod_identity(entry):
    title = str(entry.get('identity_title') or '').strip() or _normalize_vod_identity_title(entry.get('title') or '')
    year = str(entry.get('year') or '').strip()
    return '{}|{}'.format(title, year) if title and year else ''


def _shared_vod_index_path(profile_dir):
    return os.path.join(profile_dir, 'shared_vod_plot_index.json')


def _process_catalog_prefetch(addon, profile_dir, base_url, username, password):
    profile_path = _server_slot_storage_path(profile_dir, addon, 'vod_metadata_profile.json')
    profile = _load_json(profile_path, {})
    if isinstance(profile, dict) and profile.get('mode') == 'native_metadata':
        _clear_native_vod_hydration_state(addon, profile_dir)
        return 0

    request_path = _server_slot_storage_path(profile_dir, addon, 'vod_catalog_prefetch_request.json')
    request_data = _load_json(request_path, {})
    categories = request_data.get('categories') if isinstance(request_data, dict) else None
    if not isinstance(categories, list) or not categories:
        return 0
    cursor = max(0, int(request_data.get('cursor') or 0))
    if cursor >= len(categories):
        request_data['categories'] = []
        _atomic_write_json(request_path, request_data)
        return 0
    category = categories[cursor] if isinstance(categories[cursor], dict) else {}
    cid = str(category.get('category_id') or '').strip()
    if not cid:
        request_data['cursor'] = cursor + 1
        _atomic_write_json(request_path, request_data)
        return 0
    url = '{}/player_api.php?username={}&password={}&action=get_vod_streams&category_id={}'.format(
        base_url, username, password, cid
    )
    response = _safe_requests_get(url, timeout=VOD_DESC_HYDRATE_TIMEOUT_SECS, retries=0)
    try:
        payload = _decode_json_response_conservative(response)
    finally:
        try:
            response.close()
        except Exception:
            pass
    queue_path = _server_slot_storage_path(profile_dir, addon, 'vod_desc_hydration_queue.json')
    cache_path = _server_slot_storage_path(profile_dir, addon, 'vod_desc_cache.json')
    queue = _load_json(queue_path, {})
    if not isinstance(queue, dict):
        queue = {}
    pending = queue.get('pending') if isinstance(queue.get('pending'), list) else []
    cache = _load_json(cache_path, {})
    if not isinstance(cache, dict):
        cache = {}
    existing = {str(e.get('vod_id') or '') for e in pending if isinstance(e, dict)}
    now_ts = time.time()
    added = 0
    for item in payload if isinstance(payload, list) else []:
        if not isinstance(item, dict):
            continue
        vod_id = str(item.get('stream_id') or '').strip()
        if not vod_id or vod_id in existing:
            continue
        native_plot = _clean_plot(item.get('plot') or item.get('description') or '')
        if native_plot or _vod_desc_cache_has_fresh_entry(cache, vod_id):
            continue
        title = html.unescape(str(item.get('name') or '')).strip()
        year_match = re.search(r'(19|20)\d{2}', str(item.get('year') or item.get('releasedate') or title))
        pending.append({
            'vod_id': vod_id,
            'title': title[:160],
            'identity_title': _normalize_vod_identity_title(title),
            'year': year_match.group(0) if year_match else '',
            'queued_at': now_ts,
            'attempts': 0,
            'priority': 15,
            'source': 'prefetch',
        })
        existing.add(vod_id)
        added += 1
        if len(pending) >= 30000:
            break
    if added:
        queue['pending'] = pending[:30000]
        queue['updated_at'] = now_ts
        _atomic_write_json(queue_path, queue)
    request_data['cursor'] = cursor + 1
    _atomic_write_json(request_path, request_data)
    return added



def _series_paths(profile_dir, addon):
    return {
        'profile': _server_slot_storage_path(profile_dir, addon, 'series_metadata_profile.json'),
        'meta': _server_slot_storage_path(profile_dir, addon, 'series_catalog_discovery_meta.json'),
        'index': _server_slot_storage_path(profile_dir, addon, 'series_catalog_index.json'),
        'queue': _server_slot_storage_path(profile_dir, addon, 'series_desc_hydration_queue.json'),
        'backlog': _server_slot_storage_path(profile_dir, addon, 'series_catalog_discovery_backlog.json'),
        'cache': _server_slot_storage_path(profile_dir, addon, 'series_desc_cache.json'),
        'tuning': _server_slot_storage_path(profile_dir, addon, 'series_desc_hydration_tuning.json'),
    }


def _clear_native_series_hydration_state(addon, profile_dir):
    """Remove resíduos de hidratação de séries herdados de versões anteriores."""
    paths = _series_paths(profile_dir, addon)
    now_ts = time.time()
    cleared = 0

    queue = _load_json(paths['queue'], {})
    queue_pending = queue.get('pending') if isinstance(queue, dict) else None
    if isinstance(queue_pending, list) and queue_pending:
        cleared += len(queue_pending)
        _atomic_write_json(paths['queue'], {'pending': [], 'updated_at': now_ts})

    backlog = _load_json(paths['backlog'], {})
    backlog_pending = backlog.get('pending') if isinstance(backlog, dict) else None
    backlog_reason = backlog.get('disabled_reason') if isinstance(backlog, dict) else ''
    if (isinstance(backlog_pending, list) and backlog_pending) or (
        isinstance(backlog, dict) and backlog and backlog_reason != NATIVE_METADATA_DISABLED_REASON
    ):
        if isinstance(backlog_pending, list):
            cleared += len(backlog_pending)
        _atomic_write_json(paths['backlog'], {
            'pending': [],
            'updated_at': now_ts,
            'disabled_reason': NATIVE_METADATA_DISABLED_REASON,
        })
    return cleared


def _extract_series_plot_payload(payload):
    if not isinstance(payload, dict):
        return ''
    info = payload.get('info') or {}
    return _clean_plot(
        info.get('plot')
        or info.get('description')
        or info.get('overview')
        or ''
    )


def _hydrate_series_entry(entry, base_url, username, password):
    series_id = str(entry.get('series_id') or '').strip()
    attempts = int(entry.get('attempts') or 0)
    if not series_id:
        return {'status': 'invalid', 'entry': entry, 'series_id': ''}
    url = '{}/player_api.php?username={}&password={}&action=get_series_info&series_id={}'.format(
        base_url, username, password, series_id
    )
    started = time.time()
    try:
        session = _vod_worker_session()
        response = session.get(url, timeout=VOD_DESC_HYDRATE_TIMEOUT_SECS, allow_redirects=True)
        try:
            response.raise_for_status()
            payload = _decode_json_response_conservative(response)
        finally:
            try:
                response.close()
            except Exception:
                pass
        return {
            'status': 'ok',
            'entry': entry,
            'series_id': series_id,
            'plot': _extract_series_plot_payload(payload),
            'fetched_at': time.time(),
            'duration': max(0.0, time.time() - started),
        }
    except Exception as exc:
        attempts += 1
        retry_entry = dict(entry)
        retry_entry['attempts'] = attempts
        retry_entry['last_attempt_at'] = time.time()
        return {
            'status': 'retry' if attempts < VOD_DESC_HYDRATE_MAX_ATTEMPTS else 'failed',
            'entry': retry_entry,
            'series_id': series_id,
            'error': _sanitize_log_error(exc, username, password),
            'duration': max(0.0, time.time() - started),
        }


def _series_cache_has_fresh_entry(cache, series_id):
    entry = cache.get(str(series_id)) if isinstance(cache, dict) else None
    if not isinstance(entry, dict):
        return False
    try:
        fetched_at = float(entry.get('fetched_at') or 0)
    except Exception:
        fetched_at = 0
    if not fetched_at:
        return False
    plot = str(entry.get('plot') or '')
    ttl = VOD_DESC_TTL_SECS if plot else VOD_DESC_EMPTY_RETRY_SECS
    return (time.time() - fetched_at) < ttl


def _scan_series_catalog(addon, profile_dir, base_url, username, password):
    paths = _series_paths(profile_dir, addon)
    now_ts = time.time()
    meta = _load_json(paths['meta'], {})
    if not _catalog_scan_due(meta, now_ts):
        profile = _load_json(paths['profile'], {})
        return {'scanned': False, 'mode': profile.get('mode') if isinstance(profile, dict) else ''}
    if _is_video_playing():
        return {'scanned': False, 'reason': 'playing'}

    meta = meta if isinstance(meta, dict) else {}
    meta['last_attempt_at'] = now_ts
    _atomic_write_json(paths['meta'], meta)

    url = '{}/player_api.php?username={}&password={}&action=get_series'.format(
        base_url, username, password
    )
    response = _safe_requests_get(url, timeout=30, retries=1, backoff=1.0)
    try:
        payload = _decode_json_response_conservative(response)
    finally:
        try:
            response.close()
        except Exception:
            pass
    if not isinstance(payload, list):
        raise RuntimeError('catálogo de séries não retornou uma lista')

    valid = []
    with_plot = 0
    for item in payload:
        if not isinstance(item, dict):
            continue
        series_id = str(item.get('series_id') or '').strip()
        if not series_id:
            continue
        info = item.get('info') or {}
        plot = _clean_plot(
            item.get('plot')
            or item.get('description')
            or info.get('plot')
            or info.get('description')
            or info.get('overview')
            or ''
        )
        valid.append((item, series_id, plot))
        if plot:
            with_plot += 1

    valid, duplicate_total = _dedupe_catalog_records(valid)
    with_plot = sum(1 for _item, _series_id, plot in valid if plot)
    total = len(valid)
    coverage = (float(with_plot) / float(total)) if total else 0.0
    previous_profile = _load_json(paths['profile'], {})
    previous_evidence = _profile_has_native_synopsis_evidence(previous_profile)
    try:
        previous_synopsis_total = int((previous_profile or {}).get('synopsis_total') or 0)
    except Exception:
        previous_synopsis_total = 0
    native = with_plot > 0 or previous_evidence
    mode = 'native_metadata' if native else 'hydration_required'
    _atomic_write_json(paths['profile'], {
        'mode': mode,
        'coverage': coverage,
        'sample_total': total,
        'synopsis_total': max(previous_synopsis_total, with_plot),
        'last_scan_synopsis_total': with_plot,
        'duplicate_rows_ignored': duplicate_total,
        'detection_rule': METADATA_DETECTION_RULE,
        'classified_at': now_ts,
    })

    index = {}
    cache = _load_json(paths['cache'], {})
    if not isinstance(cache, dict):
        cache = {}
    backlog = []

    for item, series_id, plot in valid:
        title = html.unescape(str(item.get('name') or item.get('title') or '')).strip()
        index[series_id] = {
            'title': title[:160],
            'last_seen_at': now_ts,
            'native_plot_available': bool(plot),
        }
        if native or plot or _series_cache_has_fresh_entry(cache, series_id):
            continue
        backlog.append({
            'series_id': series_id,
            'title': title[:160],
            'queued_at': now_ts,
            'attempts': 0,
            'priority': 30,
            'source': 'series_discovery',
        })

    _atomic_write_json(paths['index'], index)
    if native:
        _atomic_write_json(paths['queue'], {'pending': [], 'updated_at': now_ts})
        _atomic_write_json(paths['backlog'], {
            'pending': [],
            'updated_at': now_ts,
            'disabled_reason': NATIVE_METADATA_DISABLED_REASON,
        })
        _clear_native_series_hydration_state(addon, profile_dir)
    else:
        _atomic_write_json(paths['backlog'], {'pending': backlog, 'updated_at': now_ts})
    meta.update({
        'last_success_at': now_ts,
        'mode': mode,
        'coverage': coverage,
        'catalog_total': total,
        'backlog_total': len(backlog),
        'duplicate_rows_ignored': duplicate_total,
    })
    _atomic_write_json(paths['meta'], meta)
    return {
        'scanned': True,
        'mode': mode,
        'coverage': coverage,
        'synopsis_total': with_plot,
        'total': total,
        'backlog': len(backlog),
        'duplicates_ignored': duplicate_total,
    }


def _refill_series_queue(addon, profile_dir):
    paths = _series_paths(profile_dir, addon)
    queue = _load_json(paths['queue'], {})
    if not isinstance(queue, dict):
        queue = {}
    pending = queue.get('pending') if isinstance(queue.get('pending'), list) else []
    backlog_data = _load_json(paths['backlog'], {})
    backlog = backlog_data.get('pending') if isinstance(backlog_data, dict) else []
    if not isinstance(backlog, list) or not backlog:
        return 0
    existing = {str(e.get('series_id') or '') for e in pending if isinstance(e, dict)}
    capacity = max(0, VOD_CATALOG_QUEUE_TARGET - len(pending))
    if capacity <= 0:
        return 0
    moved, remaining = [], []
    backlog_pruned = False
    for entry in backlog:
        if not isinstance(entry, dict):
            backlog_pruned = True
            continue
        sid = str(entry.get('series_id') or '').strip()
        if not sid or sid in existing:
            backlog_pruned = True
            continue
        if len(moved) < capacity:
            moved.append(entry)
            existing.add(sid)
        else:
            remaining.append(entry)
    if not moved:
        if backlog_pruned:
            backlog_data = backlog_data if isinstance(backlog_data, dict) else {}
            backlog_data['pending'] = remaining
            backlog_data['updated_at'] = time.time()
            _atomic_write_json(paths['backlog'], backlog_data)
        return 0
    pending.extend(moved)
    pending.sort(key=lambda e: (-int(e.get('priority') or 0), float(e.get('queued_at') or 0)))
    queue['pending'] = pending[:VOD_CATALOG_QUEUE_TARGET]
    queue['updated_at'] = time.time()
    backlog_data['pending'] = remaining
    backlog_data['updated_at'] = time.time()
    _atomic_write_json(paths['queue'], queue)
    _atomic_write_json(paths['backlog'], backlog_data)
    return len(moved)


def process_series_description_hydration(monitor=None):
    if _is_video_playing() or requests is None:
        return False
    addon = _addon()
    if addon is None:
        return False
    username = _get_username(addon)
    password = _get_password(addon)
    base_url = (_get_base_url(addon) or '').rstrip('/')
    if not (base_url and username and password):
        return False

    profile_dir = _get_profile_dir(addon)
    try:
        result = _scan_series_catalog(addon, profile_dir, base_url, username, password)
        if result.get('scanned'):
            coverage = float(result.get('coverage') or 0) * 100.0
            if result.get('mode') == 'native_metadata':
                _log(
                    'Séries modo nativo: sinopse nativa detectada neste servidor; nenhum item vazio será consultado.',
                    getattr(xbmc, 'LOGINFO', None)
                )
            else:
                _log(
                    'Descoberta automática de séries: {} total, cobertura {:.1f}%, {} aguardando hidratação.'.format(
                        result.get('total', 0), coverage, result.get('backlog', 0)
                    ),
                    getattr(xbmc, 'LOGINFO', None)
                )
    except Exception as exc:
        _log(
            'Descoberta de séries finalizou com aviso: {}.'.format(
                _sanitize_log_error(exc, username, password)
            ),
            getattr(xbmc, 'LOGDEBUG', None)
        )

    paths = _series_paths(profile_dir, addon)
    profile, promoted = _promote_profile_with_native_synopsis(paths['profile'])
    if isinstance(profile, dict) and profile.get('mode') == 'native_metadata':
        cleared = _clear_native_series_hydration_state(addon, profile_dir)
        if cleared:
            _log(
                'Modo nativo de séries removeu {} pendência(s) legada(s) sem consultar a API.'.format(cleared),
                getattr(xbmc, 'LOGDEBUG', None)
            )
        elif promoted:
            _log(
                'Perfil antigo de séries promovido para modo nativo sem consultar detalhes.',
                getattr(xbmc, 'LOGDEBUG', None)
            )
        return False

    _refill_series_queue(addon, profile_dir)
    queue = _load_json(paths['queue'], {})
    pending = queue.get('pending') if isinstance(queue, dict) else []
    if not isinstance(pending, list) or not pending:
        return False
    cache = _load_json(paths['cache'], {})
    if not isinstance(cache, dict):
        cache = {}
    pending = [
        e for e in pending
        if isinstance(e, dict)
        and str(e.get('series_id') or '').strip()
        and not _series_cache_has_fresh_entry(cache, str(e.get('series_id') or '').strip())
    ]

    tuning = _load_vod_tuning(paths['tuning'])
    workers = _clamp_int(
        tuning.get('workers'),
        VOD_DESC_HYDRATE_WORKERS_MIN,
        VOD_DESC_HYDRATE_WORKERS_MAX,
        VOD_DESC_HYDRATE_WORKERS_DEFAULT,
    )
    microbatch = _clamp_int(
        tuning.get('microbatch'),
        VOD_DESC_HYDRATE_MICROBATCH_MIN,
        VOD_DESC_HYDRATE_MICROBATCH_MAX,
        VOD_DESC_HYDRATE_MICROBATCH_DEFAULT,
    )
    block = pending[:microbatch]
    remaining = pending[microbatch:]
    failed = 0
    found = 0
    empty = 0
    durations = []
    block_started = time.time()

    with ThreadPoolExecutor(max_workers=workers, thread_name_prefix='OnePlayVIP-Series') as executor:
        futures = [
            executor.submit(_hydrate_series_entry, entry, base_url, username, password)
            for entry in block
        ]
        for future in as_completed(futures):
            result = future.result()
            durations.append(float(result.get('duration') or 0))
            status = result.get('status')
            sid = result.get('series_id') or ''
            if status == 'ok':
                plot = result.get('plot') or ''
                cache[sid] = {
                    'plot': plot,
                    'fetched_at': result.get('fetched_at') or time.time(),
                }
                if plot:
                    found += 1
                else:
                    empty += 1
            elif status == 'retry':
                remaining.append(result.get('entry') or {})
                failed += 1
            elif status == 'failed':
                failed += 1

    queue['pending'] = remaining[:VOD_CATALOG_QUEUE_TARGET]
    queue['updated_at'] = time.time()
    cache = _trim_vod_desc_cache(cache)
    elapsed = max(0.001, time.time() - block_started)
    tuning = _adapt_vod_tuning(tuning, len(block), failed, elapsed, durations)
    _atomic_write_json(paths['queue'], queue)
    _atomic_write_json(paths['cache'], cache)
    _atomic_write_json(paths['tuning'], tuning)
    _refill_series_queue(addon, profile_dir)

    _log(
        'Hidratação de séries: {} consultada(s), {} sinopse(s) encontrada(s), {} vazia(s), {} falha(s), {} pendente(s), {:.1f} item(ns)/s, workers={}, microbloco={}.'.format(
            len(block), found, empty, failed, len(remaining), len(block) / elapsed,
            tuning.get('workers'), tuning.get('microbatch')
        ),
        getattr(xbmc, 'LOGDEBUG', None)
    )
    return bool(block)


def process_vod_description_hydration(monitor=None):
    """Hidrata a fila com auto-tuning conservador e checkpoints seguros.

    Parte do perfil aprovado de 4 workers/200 itens. A cada microbloco mede
    latência, throughput e falhas por servidor. Só sobe após três blocos
    estáveis; reduz imediatamente quando detecta pressão. Reprodução bloqueia
    o próximo microbloco, e o encerramento do Kodi preserva o restante. Cada
    chamada processa poucos microblocos e devolve o controle ao service.
    """
    if _is_video_playing() or requests is None:
        return False
    addon = _addon()
    if addon is None:
        return False
    username = _get_username(addon)
    password = _get_password(addon)
    base_url = (_get_base_url(addon) or '').rstrip('/')
    if not (base_url and username and password):
        return False

    profile_dir = _get_profile_dir(addon)
    ensure_vod_catalog_discovery(addon, profile_dir, base_url, username, password)
    metadata_profile_path = _server_slot_storage_path(
        profile_dir, addon, 'vod_metadata_profile.json'
    )
    metadata_profile = _load_json(metadata_profile_path, {})
    if isinstance(metadata_profile, dict) and metadata_profile.get('mode') == 'native_metadata':
        return False
    try:
        prefetched = _process_catalog_prefetch(addon, profile_dir, base_url, username, password)
        if prefetched:
            _log('Pré-agendamento VOD inteligente: {} item(ns) descoberto(s).'.format(prefetched), getattr(xbmc, 'LOGDEBUG', None))
    except Exception as exc:
        _log(
            'Pré-agendamento VOD ignorado com segurança: {}.'.format(
                _sanitize_log_error(exc, username, password)
            ),
            getattr(xbmc, 'LOGDEBUG', None)
        )
    queue_path = _server_slot_storage_path(profile_dir, addon, 'vod_desc_hydration_queue.json')
    cache_path = _server_slot_storage_path(profile_dir, addon, 'vod_desc_cache.json')
    tuning_path = _server_slot_storage_path(profile_dir, addon, 'vod_desc_hydration_tuning.json')

    queue = _load_json(queue_path, {})
    pending = queue.get('pending') if isinstance(queue, dict) else None
    if not isinstance(pending, list) or not pending:
        moved = _refill_vod_queue_from_backlog(addon, profile_dir)
        if moved:
            queue = _load_json(queue_path, {})
            pending = queue.get('pending') if isinstance(queue, dict) else None
        if not isinstance(pending, list) or not pending:
            return False
    cache = _load_json(cache_path, {})
    if not isinstance(cache, dict):
        cache = {}
    shared_index_path = _shared_vod_index_path(profile_dir)
    shared_index = _load_json(shared_index_path, {})
    if not isinstance(shared_index, dict):
        shared_index = {}

    original_pending_count = len(pending)

    # Reaproveitamento entre servidores somente com título normalizado + ano.
    reused = 0
    filtered_pending = []
    for entry in pending:
        if not isinstance(entry, dict):
            continue
        identity = _shared_vod_identity(entry)
        shared_entry = shared_index.get(identity) if identity else None
        if isinstance(shared_entry, dict) and str(shared_entry.get('plot') or '').strip():
            vod_id = str(entry.get('vod_id') or '').strip()
            cache[vod_id] = {
                'plot': str(shared_entry.get('plot') or ''),
                'fetched_at': time.time(),
            }
            reused += 1
        else:
            filtered_pending.append(entry)
    pending = filtered_pending
    if reused:
        cache = _trim_vod_desc_cache(cache)
        _atomic_write_json(cache_path, cache)
        _log('Reaproveitamento seguro entre servidores: {} sinopse(s).'.format(reused), getattr(xbmc, 'LOGDEBUG', None))

    pending = [
        entry for entry in pending
        if isinstance(entry, dict)
        and str(entry.get('vod_id') or '').strip()
        and not _vod_desc_cache_has_fresh_entry(cache, str(entry.get('vod_id') or '').strip())
    ]
    pending.sort(key=_vod_priority_key)

    filtered_out = max(0, original_pending_count - len(pending))
    if filtered_out:
        queue['pending'] = pending[:VOD_CATALOG_QUEUE_TARGET]
        queue['updated_at'] = time.time()
        _atomic_write_json(queue_path, queue)
    if not pending:
        # Evita reler indefinidamente entradas já reaproveitadas ou cacheadas.
        _refill_vod_queue_from_backlog(addon, profile_dir)
        return False

    tuning = _load_vod_tuning(tuning_path)
    total_processed = 0
    total_success = 0
    total_empty = 0
    total_failed = 0
    total_blocks = 0
    started_at = time.time()

    while (
        pending
        and total_processed < VOD_DESC_HYDRATE_BATCH_SIZE
        and total_blocks < VOD_DESC_HYDRATE_MAX_BLOCKS_PER_CYCLE
    ):
        try:
            if monitor is not None and monitor.abortRequested():
                break
        except Exception:
            pass
        if _is_video_playing():
            break

        workers = _clamp_int(
            tuning.get('workers'),
            VOD_DESC_HYDRATE_WORKERS_MIN,
            VOD_DESC_HYDRATE_WORKERS_MAX,
            VOD_DESC_HYDRATE_WORKERS_DEFAULT,
        )
        microbatch = _clamp_int(
            tuning.get('microbatch'),
            VOD_DESC_HYDRATE_MICROBATCH_MIN,
            VOD_DESC_HYDRATE_MICROBATCH_MAX,
            VOD_DESC_HYDRATE_MICROBATCH_DEFAULT,
        )
        capacity = min(
            microbatch,
            VOD_DESC_HYDRATE_BATCH_SIZE - total_processed,
            len(pending),
        )
        block = pending[:capacity]
        pending = pending[capacity:]
        retry_entries = []
        changed_cache = False
        block_failed = 0
        block_processed = 0
        durations = []
        block_started_at = time.time()

        with ThreadPoolExecutor(
            max_workers=workers,
            thread_name_prefix='OnePlayVIP-VOD'
        ) as executor:
            futures = [
                executor.submit(_hydrate_vod_entry, entry, base_url, username, password)
                for entry in block
            ]
            for future in as_completed(futures):
                result = future.result()
                block_processed += 1
                total_processed += 1
                try:
                    durations.append(float(result.get('duration') or 0))
                except Exception:
                    pass
                status = result.get('status')
                vod_id = result.get('vod_id') or ''
                if status == 'ok':
                    cache[vod_id] = {
                        'plot': result.get('plot') or '',
                        'fetched_at': result.get('fetched_at') or time.time(),
                    }
                    if result.get('plot'):
                        total_success += 1
                    else:
                        total_empty += 1
                    changed_cache = True
                    identity = _shared_vod_identity(result.get('entry') or {})
                    if identity and str(result.get('plot') or '').strip():
                        shared_index[identity] = {
                            'plot': str(result.get('plot') or ''),
                            'updated_at': time.time(),
                        }
                elif status == 'retry':
                    retry_entries.append(result.get('entry') or {})
                    block_failed += 1
                    total_failed += 1
                    _log(
                        'Hidratação VOD {} será repetida sem afetar a navegação: {}.'.format(
                            vod_id,
                            _sanitize_log_error(
                                result.get('error') or 'falha temporária', username, password
                            )
                        ),
                        getattr(xbmc, 'LOGDEBUG', None)
                    )
                elif status == 'failed':
                    block_failed += 1
                    total_failed += 1
                    _log(
                        'Hidratação VOD {} esgotou tentativas: {}.'.format(
                            vod_id,
                            _sanitize_log_error(
                                result.get('error') or 'falha', username, password
                            )
                        ),
                        getattr(xbmc, 'LOGDEBUG', None)
                    )

        total_blocks += 1

        pending.extend(retry_entries)
        pending.sort(key=_vod_priority_key)

        queue['pending'] = pending[:30000]
        queue['updated_at'] = time.time()
        block_elapsed = max(0.001, time.time() - block_started_at)
        tuning = _adapt_vod_tuning(
            tuning,
            block_processed,
            block_failed,
            block_elapsed,
            durations,
        )
        history = tuning.get('history') if isinstance(tuning.get('history'), list) else []
        history.append({
            'at': time.time(),
            'workers': int(tuning.get('workers') or 0),
            'microbatch': int(tuning.get('microbatch') or 0),
            'throughput': float(tuning.get('last_throughput') or 0),
            'latency': float(tuning.get('last_avg_latency') or 0),
            'failure_rate': float(tuning.get('last_failure_rate') or 0),
        })
        tuning['history'] = history[-30:]
        if tuning['history']:
            tuning['rolling_throughput_30'] = sum(x.get('throughput', 0) for x in tuning['history']) / len(tuning['history'])
            tuning['rolling_latency_30'] = sum(x.get('latency', 0) for x in tuning['history']) / len(tuning['history'])

        try:
            _atomic_write_json(queue_path, queue)
            if changed_cache:
                cache = _trim_vod_desc_cache(cache)
                _atomic_write_json(cache_path, cache)
            _atomic_write_json(tuning_path, tuning)
            if shared_index:
                if len(shared_index) > 100000:
                    shared_index = dict(sorted(
                        shared_index.items(),
                        key=lambda kv: float((kv[1] or {}).get('updated_at') or 0),
                        reverse=True
                    )[:100000])
                _atomic_write_json(shared_index_path, shared_index)
        except Exception as exc:
            _log(
                'Falha ao persistir checkpoint da hidratação VOD: {}.'.format(
                    _sanitize_log_error(exc, username, password)
                ),
                getattr(xbmc, 'LOGDEBUG', None)
            )
            return False

        elapsed = max(0.001, time.time() - started_at)
        _log(
            'Hidratação VOD adaptativa: {} sinopse(s) encontrada(s), {} vazia(s), {} falha(s), {} pendente(s), '
            '{:.1f} item(ns)/s, workers={}, microbloco={}, latência={:.3f}s.'.format(
                total_success,
                total_empty,
                total_failed,
                len(pending),
                total_processed / elapsed,
                tuning.get('workers'),
                tuning.get('microbatch'),
                float(tuning.get('last_avg_latency') or 0),
            ),
            getattr(xbmc, 'LOGDEBUG', None)
        )

    try:
        _refill_vod_queue_from_backlog(addon, profile_dir)
    except Exception as exc:
        _log(
            'Reabastecimento final da fila VOD ignorado: {}.'.format(
                _sanitize_log_error(exc, username, password)
            ),
            getattr(xbmc, 'LOGDEBUG', None)
        )
    return bool(total_processed)

def prepare_epg_xmltv_cache_on_startup(monitor=None):
    if epg_runtime is None or storage_runtime is None:
        _log('EPG service ignorado: runtime indisponível.', getattr(xbmc, 'LOGDEBUG', None))
        return False

    addon = _addon()
    if addon is None:
        return False
    if not _get_enable_epg(addon):
        return False

    username = _get_username(addon)
    password = _get_password(addon)
    base_url = _get_base_url(addon)
    if not (username and password and base_url):
        return False

    if monitor is not None:
        try:
            if monitor.abortRequested():
                return False
        except Exception:
            pass

    profile_dir = _get_profile_dir(addon)

    def ensure_profile_dir():
        return storage_runtime.ensure_profile_dir(profile_dir)

    def ensure_parent_dir(path):
        return storage_runtime.ensure_parent_dir(path)

    def load_json_file(path, default=None):
        return storage_runtime.load_json_file(path, default=default, profile_dir=profile_dir, log_fn=_log)

    def atomic_write_json(path, data, ensure_ascii=False, indent=None):
        return storage_runtime.atomic_write_json(path, data, profile_dir=profile_dir, ensure_ascii=ensure_ascii, indent=indent, log_fn=_log)

    def atomic_write_text(path, text, encoding='utf-8'):
        return storage_runtime.atomic_write_text(path, text, profile_dir=profile_dir, encoding=encoding)

    def server_slot_storage_path(filename, server_idx=None):
        return _server_slot_storage_path(profile_dir, addon, filename, server_idx)

    def epg_xml_storage_path():
        return server_slot_storage_path('epg_xmltv.xml')

    def epg_meta_storage_path():
        return server_slot_storage_path('epg_xmltv_meta.json')

    ctx = {
        'profile_dir': profile_dir,
        'username': username,
        'password': password,
        'epg_cache_days': EPG_CACHE_DAYS,
        'epg_ttl': EPG_TTL,
        'epg_index_version': EPG_INDEX_VERSION,
        'epg_index_window_days': EPG_INDEX_WINDOW_DAYS,
        'max_epg_index_files': MAX_EPG_INDEX_FILES,
        'xmltv_default_offset_secs': XMLTV_DEFAULT_OFFSET_SECS,
        'epg_boundary_tolerance_secs': EPG_BOUNDARY_TOLERANCE_SECS,
        'epg_refresh_margin_secs': EPG_REFRESH_MARGIN_SECS,
        'epg_min_future_coverage_secs': EPG_MIN_FUTURE_COVERAGE_SECS,
        'epg_min_refresh_interval_secs': EPG_MIN_REFRESH_INTERVAL_SECS,
        'epg_failure_retry_secs': EPG_FAILURE_RETRY_SECS,
        'epg_require_full_window': EPG_REQUIRE_FULL_WINDOW,
        'honor_refresh_request': True,
        'log_fn': _log,
        'load_json_fn': load_json_file,
        'atomic_write_json_fn': atomic_write_json,
        'atomic_write_text_fn': atomic_write_text,
        'epg_meta_storage_path_fn': epg_meta_storage_path,
        'epg_xml_storage_path_fn': epg_xml_storage_path,
        'server_slot_storage_path_fn': server_slot_storage_path,
        'safe_delete_file_fn': storage_runtime.safe_delete_file,
        'ensure_profile_dir_fn': ensure_profile_dir,
        'ensure_parent_dir_fn': ensure_parent_dir,
        'safe_requests_get_fn': _safe_requests_get,
        'get_base_url_fn': lambda: base_url,
        'fingerprint_fn': lambda: _fingerprint(addon),
        'clear_desc_caches_fn': lambda: None,
        'current_server_scope_identity_fn': lambda: _server_slot_scope_name(addon),
        # Aqui o download é permitido porque service.py roda fora da navegação do usuário.
        'download_on_demand': True,
    }

    try:
        ensure_profile_dir()
        epg_runtime.epg_load_parsed(ctx, force=False)
        meta = load_json_file(epg_meta_storage_path(), {})
        retry_after = int((meta or {}).get('retry_after') or 0)
        if str((meta or {}).get('last_attempt_status') or '') == 'failed' and retry_after > int(time.time()):
            remaining = max(1, retry_after - int(time.time()))
            _log('EPG indisponível neste servidor; nova tentativa automática em aproximadamente {} minuto(s).'.format(max(1, remaining // 60)), getattr(xbmc, 'LOGWARNING', None))
        else:
            _log('EPG XMLTV preparado em background pelo service.py.', getattr(xbmc, 'LOGINFO', None))
        return True
    except Exception as exc:
        _log(
            'Falha ao preparar EPG XMLTV no service.py: {}'.format(
                _sanitize_log_error(exc, username, password)
            ),
            getattr(xbmc, 'LOGWARNING', None)
        )
        return False


def _epg_refresh_request_pending(addon):
    try:
        if addon is None or not _get_enable_epg(addon):
            return False
        profile_dir = _get_profile_dir(addon)
        request_path = _server_slot_storage_path(profile_dir, addon, 'epg_refresh_request.json')
        return bool(_path_exists(request_path))
    except Exception:
        return False


def _is_video_playing():
    """Evita disputar rede/CPU com reprodução ativa; o pedido fica pendente."""
    try:
        if xbmc is None:
            return False
        return bool(xbmc.Player().isPlayingVideo())
    except Exception:
        return False


_EPG_SERVICE_STATE = {
    'last_defer_log_at': 0.0,
}


def _prepare_pending_epg(monitor=None, reason='pedido_pendente'):
    """Executa EPG somente no service residente, que possui contexto do addon.

    A UI apenas grava o pedido no storage por servidor. Assim não usamos
    RunScript(service.py), que no Kodi 21 é iniciado sem addon id e perde os
    imports/resources do addon.
    """
    addon = _addon()
    if addon is None or not _get_enable_epg(addon):
        return False

    if _is_video_playing():
        now = time.time()
        if (now - float(_EPG_SERVICE_STATE.get('last_defer_log_at') or 0)) >= 60:
            _log('EPG pendente aguardando o fim da reprodução para não disputar buffer/rede.', getattr(xbmc, 'LOGDEBUG', None))
            _EPG_SERVICE_STATE['last_defer_log_at'] = now
        return False

    _log('EPG background processado pelo service residente: {}.'.format(reason), getattr(xbmc, 'LOGDEBUG', None))
    return bool(prepare_epg_xmltv_cache_on_startup(monitor=monitor))


def _wait_for_abort(monitor, seconds):
    try:
        if monitor is not None:
            return bool(monitor.waitForAbort(float(seconds)))
    except Exception:
        pass
    try:
        if xbmc is not None:
            xbmc.sleep(int(float(seconds) * 1000))
    except Exception:
        pass
    return False


def _is_custom_dns_enabled(addon):
    try:
        return bool(addon and (addon.getSetting('enable_custom_dns') or 'false').lower() == 'true')
    except Exception:
        return False


def _is_any_proxy_enabled(addon):
    """Respeita os dois controles independentes; sem chave, o proxy fica opt-in."""
    try:
        live = (addon.getSetting('proxy_live_enabled') or 'false').lower() == 'true'
        vod = (addon.getSetting('proxy_vod_enabled') or 'false').lower() == 'true'
        return bool(live or vod)
    except Exception:
        return False


def prepare_proxy_runtime_on_startup():
    """Mantém o listener local no service apenas quando Live ou VOD estão ativos.

    O plugin de navegação termina logo após setResolvedUrl; hospedar o proxy
    nele deixa VOD sujeito a corrida de encerramento. O service é o dono único
    do listener e publica a porta validada para as invocações de main.py.
    """
    addon = _addon()
    try:
        if dns_runtime is not None:
            if _is_custom_dns_enabled(addon):
                dns_runtime.install(cache_ttl=1800, mode_logger=False)
            else:
                dns_runtime.uninstall()
    except Exception as exc:
        _log('Configuração DNS do proxy finalizou com aviso: {}'.format(_sanitize_log_error(exc)), getattr(xbmc, 'LOGDEBUG', None))

    try:
        if proxy_runtime is None:
            return False

        enabled = _is_any_proxy_enabled(addon)
        if not enabled:
            # Com default opt-in, não fica chamando stop/logando em cada ciclo.
            if _PROXY_SERVICE_STATE.get('enabled') is not False:
                try:
                    proxy_runtime.stop_proxy()
                except Exception:
                    pass
                _log('Proxy nativo em espera: VOD e canais ao vivo estão desativados.', getattr(xbmc, 'LOGDEBUG', None))
            _PROXY_SERVICE_STATE['enabled'] = False
            _PROXY_SERVICE_STATE['port'] = 0
            return True

        started = bool(proxy_runtime.start_proxy())
        if started:
            port = int(getattr(proxy_runtime, 'get_proxy_port', lambda: 0)() or 0)
            if 1024 <= port <= 65535:
                # O health-check é feito periodicamente, mas a confirmação é
                # registrada só no start/alteração de porta, não a cada 15 s.
                if (not _PROXY_SERVICE_STATE.get('enabled')) or int(_PROXY_SERVICE_STATE.get('port') or 0) != port:
                    _log('Proxy nativo OnePlayVIP pronto no service na porta dinâmica {}.'.format(port), getattr(xbmc, 'LOGINFO', None))
                _PROXY_SERVICE_STATE['enabled'] = True
                _PROXY_SERVICE_STATE['port'] = port
            else:
                _log('Proxy nativo OnePlayVIP iniciou sem publicar porta válida.', getattr(xbmc, 'LOGWARNING', None))
                _PROXY_SERVICE_STATE['enabled'] = False
                _PROXY_SERVICE_STATE['port'] = 0
                return False
        return started
    except Exception as exc:
        _log('Falha ao preparar proxy nativo OnePlayVIP: {}'.format(_sanitize_log_error(exc)), getattr(xbmc, 'LOGWARNING', None))
        return False



def _create_service_monitor():
    """Monitor que acelera a reação a mudanças nas configurações do addon."""
    if xbmc is None:
        return None
    try:
        class _OnePlayServiceMonitor(xbmc.Monitor):
            def __init__(self):
                try:
                    super(_OnePlayServiceMonitor, self).__init__()
                except Exception:
                    pass
                # A mesma notificação do Kodi serve para proxy e EPG. O loop
                # decide qual subsistema precisa agir; não há polling de UI.
                self.settings_changed = False
                # Alias interno/legado: evita quebrar consumidores já auditados
                # que apenas observavam a reação do monitor do proxy.
                self.proxy_settings_changed = False

            def onSettingsChanged(self):
                self.settings_changed = True
                self.proxy_settings_changed = True

        return _OnePlayServiceMonitor()
    except Exception:
        try:
            return xbmc.Monitor()
        except Exception:
            return None

def run_service_loop():
    monitor = _create_service_monitor()

    _log_i18n_runtime_probe(_addon())

    # Executa uma única varredura local no startup. Nenhuma chamada de rede é
    # feita; somente URLs autenticadas antigas da Pesquisa Global são removidas.
    try:
        scrub_legacy_search_state_urls()
        scrub_legacy_epg_meta_credentials()
    except Exception as exc:
        _log(
            'Migração local do estado de pesquisa finalizou com aviso: {}'.format(
                _sanitize_log_error(exc)
            ),
            getattr(xbmc, 'LOGDEBUG', None),
        )

    # O proxy precisa estar pronto antes da primeira navegação/reprodução; EPG
    # continua com pausa leve para não disputar I/O no boot do Kodi.
    prepare_proxy_runtime_on_startup()
    if _wait_for_abort(monitor, 2.5):
        return

    cleanup_downloaded_apks_on_startup()
    if _wait_for_abort(monitor, 1.5):
        return

    try:
        prepare_epg_xmltv_cache_on_startup(monitor=monitor)
    except Exception as exc:
        _log('Service EPG inicial finalizou com aviso: {}'.format(_sanitize_log_error(exc)), getattr(xbmc, 'LOGDEBUG', None))

    last_epg_check = time.time()
    last_epg_request_poll = 0.0
    last_proxy_check = time.time()
    last_vod_hydration_poll = 0.0
    # Guarda somente configurações que afetam o XMLTV. Assim, o health-check
    # periódico do proxy e alterações visuais não recriam o caminho do EPG.
    last_epg_settings_signature = _epg_settings_signature(_addon())
    epg_settings_due_at = 0.0
    try:
        while True:
            # Espera curta, sem polling de rede: quando a pessoa ativa Live/VOD
            # nas configurações, o listener do service fica disponível em até 1 s.
            if _wait_for_abort(monitor, 1):
                break
            now = time.time()
            settings_changed = bool(getattr(monitor, 'settings_changed', False))
            if settings_changed or (now - last_proxy_check) >= 15:
                prepare_proxy_runtime_on_startup()
                last_proxy_check = now

                # O mesmo monitor notifica qualquer ajuste do addon. Só agenda
                # EPG quando a configuração que o afeta mudou: habilitação ou
                # identidade do acesso/servidor. Um tick periódico do proxy, ou
                # mudança de DNS/tema/PIN, não deve reprocessar XMLTV.
                if settings_changed:
                    addon = _addon()
                    current_signature, should_prepare_epg = _epg_settings_change_requires_prepare(
                        last_epg_settings_signature, addon
                    )
                    last_epg_settings_signature = current_signature
                    if should_prepare_epg:
                        epg_settings_due_at = now + EPG_SETTINGS_DEBOUNCE_SECS
                    elif not current_signature[0]:
                        epg_settings_due_at = 0.0
                try:
                    monitor.settings_changed = False
                    monitor.proxy_settings_changed = False
                except Exception:
                    pass

            # Ativação/alteração vinda de settings: executa uma única vez após
            # a curta janela de persistência, sempre no contexto do service.
            if epg_settings_due_at and now >= epg_settings_due_at:
                try:
                    _prepare_pending_epg(monitor=monitor, reason='settings_changed')
                except Exception as exc:
                    _log('Service EPG por alteração de configuração finalizou com aviso: {}'.format(_sanitize_log_error(exc)), getattr(xbmc, 'LOGDEBUG', None))
                last_epg_check = now
                epg_settings_due_at = 0.0

            # A navegação grava epg_refresh_request.json; o service residente
            # consome em até dois segundos. Não há RunScript avulso, portanto o
            # addon id e seus recursos permanecem válidos no Kodi 19-21.
            if (now - last_epg_request_poll) >= EPG_SERVICE_REQUEST_POLL_SECS:
                addon = _addon()
                if _epg_refresh_request_pending(addon):
                    processed_request = False
                    try:
                        processed_request = bool(_prepare_pending_epg(monitor=monitor, reason='request_file'))
                    except Exception as exc:
                        _log('Service EPG por pedido pendente finalizou com aviso: {}'.format(_sanitize_log_error(exc)), getattr(xbmc, 'LOGDEBUG', None))
                    # Ao ativar o EPG, a UI cria o pedido de arquivo e o Monitor
                    # também notifica a mesma mudança. Se o pedido já foi atendido
                    # neste ciclo, não releia o XML/índice pela agenda atrasada.
                    # Em falha ou durante reprodução, mantém a agenda para não
                    # perder a recuperação posterior.
                    if processed_request:
                        epg_settings_due_at = 0.0
                    last_epg_check = now
                last_epg_request_poll = now

            if (now - last_epg_check) >= EPG_SERVICE_CHECK_INTERVAL_SECS:
                try:
                    prepare_epg_xmltv_cache_on_startup(monitor=monitor)
                except Exception as exc:
                    _log('Service EPG periódico finalizou com aviso: {}'.format(_sanitize_log_error(exc)), getattr(xbmc, 'LOGDEBUG', None))
                last_epg_check = now

            if (now - last_vod_hydration_poll) >= VOD_DESC_HYDRATE_POLL_SECS:
                try:
                    process_vod_description_hydration(monitor=monitor)
                except Exception as exc:
                    _log('Service de sinopses VOD finalizou com aviso: {}'.format(_sanitize_log_error(exc)), getattr(xbmc, 'LOGDEBUG', None))
                try:
                    process_series_description_hydration(monitor=monitor)
                except Exception as exc:
                    _log('Service de sinopses de séries finalizou com aviso: {}'.format(_sanitize_log_error(exc)), getattr(xbmc, 'LOGDEBUG', None))
                last_vod_hydration_poll = now
    finally:
        try:
            if proxy_runtime is not None:
                proxy_runtime.stop_proxy()
        except Exception:
            pass
        try:
            if dns_runtime is not None:
                dns_runtime.uninstall()
        except Exception:
            pass
        try:
            if http_runtime is not None:
                http_runtime.close_thread_sessions()
        except Exception:
            pass


if __name__ == '__main__':
    try:
        run_service_loop()
    except Exception as exc:
        _log('Service finalizou com aviso: {}'.format(_sanitize_log_error(exc)), getattr(xbmc, 'LOGDEBUG', None))
