# -*- coding: utf-8 -*-
import base64
import json
import urllib.parse
import requests
import binascii
import os
import re
import time
import logging
import threading
import socket
import select
import sys
import tempfile
from collections import deque

try:
    import xbmc
    import xbmcaddon
except ImportError:
    from kodi_six import xbmc, xbmcaddon

from urllib.parse import urljoin
try:
    from urllib.request import Request as UrlRequest, urlopen as urlopen_url
    from urllib.error import HTTPError as UrlHTTPError, URLError as UrlURLError
except ImportError:
    from urllib2 import Request as UrlRequest, urlopen as urlopen_url, HTTPError as UrlHTTPError, URLError as UrlURLError  # type: ignore
from requests.exceptions import ConnectionError, RequestException, ReadTimeout, ChunkedEncodingError
try:
    from urllib3.exceptions import IncompleteRead
except ImportError:
    from urllib2 import HTTPError as IncompleteRead  # type: ignore

try:
    from http.server import BaseHTTPRequestHandler, HTTPServer
    from socketserver import ThreadingMixIn
except ImportError:
    from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer  # type: ignore
    from SocketServer import ThreadingMixIn  # type: ignore

# Porta dinâmica exclusiva do OnePlayVIP.
# O bind em porta 0 deixa o próprio sistema operacional reservar, de forma
# atômica, uma porta local livre e alta a cada início do service. Isso evita
# colisões com portas conhecidas de outros addons (por exemplo 8088/8089) e
# também elimina a corrida entre "testar se está livre" e efetivamente abrir.
# A porta escolhida é publicada no profile e validada pelo health-check antes
# de qualquer URL de reprodução ser montada.
AUTO_BIND_PORT = 0
MIN_DYNAMIC_PORT = 20000
MAX_DYNAMIC_PORT = 65535
DEFAULT_PORT = AUTO_BIND_PORT
PORT = None
PORT_CANDIDATES = ()
PROXY_HEALTH_PATH = '/_oneplayvip_health'
PROXY_SERVICE_ID = 'oneplayvip-proxy'
PROXY_RUNTIME_FILE = 'oneplayvip_proxy_runtime.json'
PROXY_TARGET_FILE_PREFIX = 'oneplayvip_proxy_target_'
PROXY_TARGET_TTL_SECS = 24 * 3600
DEFAULT_USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36"

IP_CACHE_TS = {}
IP_CACHE_MP4 = {}
AGENT_OF_CHAOS = {}
COUNT_CLEAR = {}
_SERVER_THREAD = None
_SERVER_STARTED = False
_SERVER = None
_SERVER_LOCK = threading.Lock()
_TARGET_LOCK = threading.Lock()
_TARGET_REGISTRY = {}
_TARGET_CLEANUP_AT = 0.0

_TS_STATUS_LOG = {}
_ORIGIN_HEALTH = {}
_LOG_BURSTS = {}
_HLS_PLAYLIST_CACHE = {}
_SESSION_CACHE = {
    'good_hosts': {},
    'bad_hosts': {},
    'working_user_agent': {},
    'last_working_url': {},
    'content_type': {},
    'source_profile': {},
}
# O handle do addon e o snapshot das preferências têm relógios independentes.
# Compartilhar um único timestamp fazia o service renovar o handle a cada ciclo
# e o proxy devolver o snapshot padrão antes de reler proxy_mode/toggles.
_SETTINGS_CACHE = {
    'addon_ts': 0,
    'settings_ts': 0,
    'addon': None,
    'mode': 'stable',
    'xtream_hardening': True,
    'host_memory': False,
    'debug': False,
}
_USER_AGENT_POOL = [
    DEFAULT_USER_AGENT,
    'Mozilla/5.0 (Linux; Android 10; SM-A505G) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Mobile Safari/537.36',
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:133.0) Gecko/20100101 Firefox/133.0',
    'ExoPlayerLib/2.18.7',
]
_PROXY_POLICIES = {
    # Cada fluxo usa a mesma seleção operacional, mas tem orçamento próprio:
    # VOD direto retoma bytes; VOD HLS repete segmento íntegro; Live HLS mantém
    # segmento contínuo via Range; Live TS reconecta mantendo a conexão local.
    'fast': {
        'hls_max_retries': 5, 'live_hls_max_retries': 5, 'live_segment_max_retries': 2,
        # Segmento Live que responde 404: 1 retry adicional (2 tentativas totais).
        # Perfil Rápido prioriza voltar imediatamente para a playlist atual.
        'live_segment_404_retries': 1,
        'vod_max_retries': 1, 'vod_segment_max_retries': 2, 'vod_playlist_max_retries': 5,
        'ts_max_retries': 999999, 'live_ts_reprobe_every': 4,
        'timeout_connect': 4, 'timeout_read': 10, 'cooldown_base': 0.10,
    },
    'balance': {
        'hls_max_retries': 7, 'live_hls_max_retries': 7, 'live_segment_max_retries': 3,
        # Segmento Live 404: 2 retries adicionais (3 tentativas totais).
        'live_segment_404_retries': 2,
        'vod_max_retries': 2, 'vod_segment_max_retries': 3, 'vod_playlist_max_retries': 7,
        'ts_max_retries': 999999, 'live_ts_reprobe_every': 4,
        'timeout_connect': 5, 'timeout_read': 15, 'cooldown_base': 0.15,
    },
    'stable': {
        'hls_max_retries': 9, 'live_hls_max_retries': 9, 'live_segment_max_retries': 4,
        # Segmento Live 404: 3 retries adicionais (4 tentativas totais).
        # É o teto deliberado para não atrasar a renovação da playlist.
        'live_segment_404_retries': 3,
        'vod_max_retries': 3, 'vod_segment_max_retries': 4, 'vod_playlist_max_retries': 9,
        'ts_max_retries': 999999, 'live_ts_reprobe_every': 3,
        'timeout_connect': 6, 'timeout_read': 18, 'cooldown_base': 0.20,
        'ts_keepalive_window': 12.0, 'ts_null_packets': 64, 'ts_fill_sleep': 0.08,
    },
}

_RUNTIME_CACHE_LIMITS = {
    'stream_cache_keys': 24,
    'agent_chaos': 128,
    'status_log': 256,
    'log_bursts': 256,
    'origin_health': 128,
}

# completa defaults legados sem forçar mudança de comportamento em outros pontos
for _mode_name, _mode_policy in _PROXY_POLICIES.items():
    _mode_policy.setdefault('ts_keepalive_window', 4.0 if _mode_name == 'fast' else 5.5 if _mode_name == 'balance' else 8.0)
    _mode_policy.setdefault('ts_null_packets', 12 if _mode_name == 'fast' else 20 if _mode_name == 'balance' else 32)
    _mode_policy.setdefault('ts_fill_sleep', 0.04 if _mode_name == 'fast' else 0.05 if _mode_name == 'balance' else 0.06)
    _mode_policy.setdefault('ts_startup_keepalive_window', 0.0)
    _mode_policy.setdefault('ts_startup_read_timeout', 4.0 if _mode_name == 'fast' else 5.0 if _mode_name == 'balance' else 6.0)
    _mode_policy.setdefault('ts_startup_handoff_window', 4.0 if _mode_name == 'fast' else 5.0 if _mode_name == 'balance' else 6.0)
    _mode_policy.setdefault('ts_startup_handoff_bytes', 262144 if _mode_name == 'fast' else 393216 if _mode_name == 'balance' else 524288)
    _mode_policy.setdefault('hls_playlist_hold', 8.0 if _mode_name == 'fast' else 10.0 if _mode_name == 'balance' else 12.0)
    _mode_policy.setdefault('live_hls_max_retries', _mode_policy.get('hls_max_retries', 5))
    _mode_policy.setdefault('live_segment_max_retries', 2 if _mode_name == 'fast' else 3 if _mode_name == 'balance' else 4)
    # Retentativas adicionais de 404 em segmento Live (original + retries).
    _mode_policy.setdefault('live_segment_404_retries', 1 if _mode_name == 'fast' else 2 if _mode_name == 'balance' else 3)
    _mode_policy.setdefault('vod_playlist_max_retries', _mode_policy.get('hls_max_retries', 5))
    _mode_policy.setdefault('live_ts_reprobe_every', 4 if _mode_name != 'stable' else 3)
    _mode_policy.setdefault('hls_chunk_size', 32768 if _mode_name == 'fast' else 49152 if _mode_name == 'balance' else 65536)
    _mode_policy.setdefault('vod_max_retries', 1 if _mode_name == 'fast' else 2 if _mode_name == 'balance' else 3)
    _mode_policy.setdefault('vod_segment_max_retries', 2 if _mode_name == 'fast' else 3 if _mode_name == 'balance' else 4)

# Silencia o root logger para evitar spam no kodi.log e usa xbmc.log para eventos úteis.
_root_logger = logging.getLogger()
try:
    _root_logger.handlers = []
except Exception:
    pass
_root_logger.setLevel(logging.CRITICAL)
for _name in ('werkzeug', 'urllib3', 'requests'):
    try:
        _lg = logging.getLogger(_name)
        _lg.handlers = []
        _lg.propagate = False
        _lg.setLevel(logging.CRITICAL)
    except Exception:
        pass

for _mod in list(sys.modules):
    if _mod.startswith('flask') or _mod.startswith('werkzeug') or _mod.startswith('netunblock'):
        try:
            del sys.modules[_mod]
        except Exception:
            pass


def _safe_text(value):
    try:
        text = str(value)
    except Exception:
        try:
            text = repr(value)
        except Exception:
            return 'erro_desconhecido'
    text = re.sub(r'(?i)\b(username|password)=([^&\s]+)', lambda m: '%s=***' % m.group(1), text)
    text = re.sub(r'(?i)(https?://)([^/@\s:]+):([^/@\s]+)@', r'\1***:***@', text)
    text = re.sub(r'(?i)(/(?:live|movie|series)/)([^/\s?#]+)/([^/\s?#]+)(/)', r'\1***/***\4', text)
    return text[:1600]



def _addon_profile_path():
    """Retorna um caminho local estável para coordenar proxy entre service e plugin.

    Cada invocação Kodi tem seu próprio interpretador Python; portanto a porta
    ativa não pode existir somente em variável de módulo. O arquivo runtime
    permite que o plugin encontre o proxy que vive no service.
    """
    try:
        addon = _addon_handle()
        raw = addon.getAddonInfo('profile') if addon else ''
        if raw:
            try:
                import xbmcvfs
                translated = xbmcvfs.translatePath(raw)
            except Exception:
                try:
                    translated = xbmc.translatePath(raw)
                except Exception:
                    translated = raw
            if translated:
                try:
                    os.makedirs(translated, exist_ok=True)
                except Exception:
                    pass
                return translated
    except Exception:
        pass
    fallback = os.path.join(os.path.expanduser('~'), '.kodi', 'userdata', 'addon_data', 'plugin.video.oneplayvip')
    try:
        os.makedirs(fallback, exist_ok=True)
    except Exception:
        pass
    return fallback


def _runtime_state_path():
    return os.path.join(_addon_profile_path(), PROXY_RUNTIME_FILE)


def _proxy_target_path(token):
    return os.path.join(_addon_profile_path(), PROXY_TARGET_FILE_PREFIX + str(token) + '.json')


def _valid_proxy_target_token(token):
    return bool(re.match(r'^[a-f0-9]{32}$', str(token or '')))


def _cleanup_proxy_targets(force=False):
    """Limita memória e remove somente registros locais vencidos."""
    global _TARGET_CLEANUP_AT
    now = time.time()
    with _TARGET_LOCK:
        expired = [
            token for token, entry in _TARGET_REGISTRY.items()
            if (now - float((entry or {}).get('last_used_at') or (entry or {}).get('created_at') or 0)) > PROXY_TARGET_TTL_SECS
        ]
        for token in expired:
            _TARGET_REGISTRY.pop(token, None)
        while len(_TARGET_REGISTRY) > 8192:
            try:
                _TARGET_REGISTRY.pop(next(iter(_TARGET_REGISTRY)))
            except Exception:
                break
        if not force and (now - _TARGET_CLEANUP_AT) < 300:
            return
        _TARGET_CLEANUP_AT = now

    root = _addon_profile_path()
    try:
        names = os.listdir(root)
    except Exception:
        return
    for name in names:
        if not name.startswith(PROXY_TARGET_FILE_PREFIX) or not name.endswith('.json'):
            continue
        path = os.path.join(root, name)
        try:
            if (now - os.path.getmtime(path)) > PROXY_TARGET_TTL_SECS:
                os.remove(path)
        except Exception:
            pass


def register_proxy_target(url, header_blob='', kind='', persist=False):
    """Registra uma origem e devolve token opaco seguro para o kodi.log."""
    if not _is_valid_upstream_url(url):
        return ''
    try:
        token = binascii.hexlify(os.urandom(16)).decode('ascii')
    except Exception:
        return ''
    entry = {
        'url': str(url),
        'headers': str(header_blob or ''),
        'kind': 'vod' if str(kind or '').lower() == 'vod' else 'live',
        'created_at': time.time(),
    }
    with _TARGET_LOCK:
        _TARGET_REGISTRY[token] = entry
    if persist:
        path = _proxy_target_path(token)
        tmp = path + '.tmp'
        try:
            with open(tmp, 'w', encoding='utf-8') as fh:
                json.dump(entry, fh, ensure_ascii=False, separators=(',', ':'))
            try:
                os.chmod(tmp, 0o600)
            except Exception:
                pass
            os.replace(tmp, path)
        except Exception:
            try:
                if os.path.exists(tmp):
                    os.remove(tmp)
            except Exception:
                pass
            with _TARGET_LOCK:
                _TARGET_REGISTRY.pop(token, None)
            return ''
    _cleanup_proxy_targets()
    return token


def resolve_proxy_target(token):
    """Resolve token criado pelo plugin, inclusive entre processos do Kodi."""
    token = str(token or '').strip().lower()
    if not _valid_proxy_target_token(token):
        return None
    with _TARGET_LOCK:
        entry = _TARGET_REGISTRY.get(token)
    if not isinstance(entry, dict):
        try:
            with open(_proxy_target_path(token), 'r', encoding='utf-8') as fh:
                entry = json.load(fh)
        except Exception:
            return None
        if not isinstance(entry, dict):
            return None
        with _TARGET_LOCK:
            _TARGET_REGISTRY[token] = entry
        # Depois do primeiro handoff entre plugin e service, a memória do
        # listener é a dona do token; elimina a cópia transitória do disco.
        try:
            os.remove(_proxy_target_path(token))
        except Exception:
            pass
    try:
        now = time.time()
        if (now - float(entry.get('last_used_at') or entry.get('created_at') or 0)) > PROXY_TARGET_TTL_SECS:
            return None
    except Exception:
        return None
    if not _is_valid_upstream_url(entry.get('url')):
        return None
    entry['last_used_at'] = now
    with _TARGET_LOCK:
        _TARGET_REGISTRY[token] = entry
    return dict(entry)


def _target_from_query_params(params):
    token = (params.get('token') or [''])[0]
    entry = resolve_proxy_target(token) if token else None
    if isinstance(entry, dict):
        return (
            entry.get('url') or '',
            (params.get('headers') or [entry.get('headers') or ''])[0],
            (params.get('kind') or [entry.get('kind') or 'live'])[0],
        )
    url = (params.get('url') or [None])[0]
    if url:
        url = urllib.parse.unquote(url)
    return (
        url or '',
        (params.get('headers') or [''])[0],
        (params.get('kind') or ['live'])[0],
    )


def _write_runtime_state(port):
    data = {
        'service': PROXY_SERVICE_ID,
        'port': int(port),
        'version': 2,
        'pid': os.getpid(),
        'updated': time.time(),
    }
    path = _runtime_state_path()
    tmp = path + '.tmp'
    try:
        with open(tmp, 'w', encoding='utf-8') as fh:
            json.dump(data, fh, separators=(',', ':'))
        try:
            os.replace(tmp, path)
        except AttributeError:
            if os.path.exists(path):
                os.remove(path)
            os.rename(tmp, path)
        return True
    except Exception:
        try:
            if os.path.exists(tmp):
                os.remove(tmp)
        except Exception:
            pass
        return False


def _read_runtime_port():
    try:
        with open(_runtime_state_path(), 'r', encoding='utf-8') as fh:
            data = json.load(fh)
        if not isinstance(data, dict) or data.get('service') != PROXY_SERVICE_ID:
            return None
        port = int(data.get('port') or 0)
        return port if 1 <= port <= 65535 else None
    except Exception:
        return None


def _clear_runtime_state(port=None):
    path = _runtime_state_path()
    try:
        if port is not None:
            current = _read_runtime_port()
            if current is not None and int(current) != int(port):
                return False
        if os.path.exists(path):
            os.remove(path)
        return True
    except Exception:
        return False


def _probe_proxy_identity(port, timeout=0.40):
    """Confirma que uma porta local é realmente do OnePlayVIP.

    Não basta conectar na porta: o antigo código aceitava qualquer serviço em
    8088 como se fosse próprio, e isso enviava VOD ao proxy do Matrix.
    """
    try:
        port = int(port)
        if not (1 <= port <= 65535):
            return False
        req = UrlRequest('http://127.0.0.1:%d%s' % (port, PROXY_HEALTH_PATH), headers={'Connection': 'close'})
        response = urlopen_url(req, timeout=max(0.10, float(timeout)))
        try:
            body = response.read(4096)
            status = int(response.getcode() or 0)
        finally:
            try:
                response.close()
            except Exception:
                pass
        if status != 200:
            return False
        if isinstance(body, bytes):
            body = body.decode('utf-8', 'replace')
        data = json.loads(body or '{}')
        return bool(isinstance(data, dict) and data.get('service') == PROXY_SERVICE_ID)
    except Exception:
        return False


def _candidate_ports():
    """Portas já conhecidas do proxy em execução.

    Não existe mais uma faixa fixa para tentar. A porta nova é atribuída pelo
    sistema operacional com bind(0); aqui apenas consultamos o estado publicado
    ou o listener deste próprio processo.
    """
    seen = set()
    ordered = []
    for candidate in (_read_runtime_port(), PORT):
        try:
            candidate = int(candidate)
        except Exception:
            continue
        if candidate in seen or not (1 <= candidate <= 65535):
            continue
        seen.add(candidate)
        ordered.append(candidate)
    return ordered


def _discover_running_proxy():
    global PORT, _SERVER_STARTED
    for candidate in _candidate_ports():
        if _probe_proxy_identity(candidate):
            PORT = candidate
            _SERVER_STARTED = True
            _write_runtime_state(candidate)
            return candidate
    return None


def get_proxy_port():
    """Retorna somente uma porta real, identificada como OnePlayVIP.

    Zero é usado como sentinela de "ainda não iniciado". Quem montar uma URL
    deve chamar start_proxy() antes e recusar o proxy se uma porta válida não
    estiver disponível; nunca voltar para uma porta fixa de outro addon.
    """
    port = _discover_running_proxy()
    if port:
        return int(port)
    try:
        current = int(PORT or 0)
    except Exception:
        current = 0
    return current if 1 <= current <= 65535 else 0

def _emit_log(message, level=logging.INFO):
    try:
        if level >= logging.ERROR:
            xbmc.log('[OnePlayProxy] ' + message, level=xbmc.LOGERROR)
        elif level >= logging.WARNING:
            xbmc.log('[OnePlayProxy] ' + message, level=xbmc.LOGWARNING)
        elif level >= logging.INFO:
            xbmc.log('[OnePlayProxy] ' + message, level=xbmc.LOGINFO)
        else:
            xbmc.log('[OnePlayProxy] ' + message, level=xbmc.LOGDEBUG)
    except Exception:
        pass


def _throttled_log(key, interval, level, message):
    if level < logging.INFO and not bool(_SETTINGS_CACHE.get('debug')):
        return
    now = time.time()
    last = _TS_STATUS_LOG.get(key, 0)
    if (now - last) >= interval:
        _TS_STATUS_LOG[key] = now
        _trim_mapping(_TS_STATUS_LOG, max_items=_RUNTIME_CACHE_LIMITS.get('status_log', 256))
        _emit_log(message, level)


def _burst_log(key, window, level, template):
    if level < logging.INFO and not bool(_SETTINGS_CACHE.get('debug')):
        return
    now = time.time()
    item = _LOG_BURSTS.get(key)
    if not item or (now - item['start']) > window:
        item = {'start': now, 'count': 0}
    item['count'] += 1
    _LOG_BURSTS[key] = item
    _trim_mapping(_LOG_BURSTS, max_items=_RUNTIME_CACHE_LIMITS.get('log_bursts', 256))
    if item['count'] == 1:
        _emit_log(template % 1, level)
    elif item['count'] in (5, 15, 30) or (now - item['start']) >= window:
        _emit_log(template % item['count'], level)
        item['start'] = now
        item['count'] = 0


def _log_ts_status_once(status_code):
    _burst_log('status:%s' % status_code, 45, logging.INFO, '[TS Downloader] HTTP {} recorrente; %sx no período.'.format(status_code))


def _log(msg, level=None):
    try:
        if level is None:
            level = xbmc.LOGINFO
        # evita poluir produção com debug sem perder mensagens úteis
        if level == xbmc.LOGDEBUG and not _settings_snapshot().get('debug'):
            return
        xbmc.log('[OnePlayProxy] ' + msg, level=level)
    except Exception:
        pass


def _decode_header_blob(value):
    value = (value or '').strip()
    if not value:
        return {}
    try:
        padded = value + ('=' * (-len(value) % 4))
        payload = base64.urlsafe_b64decode(padded.encode('ascii')).decode('utf-8')
        data = json.loads(payload)
        if not isinstance(data, dict):
            return {}
        blocked = {'host', 'connection', 'content-length', 'accept-encoding'}
        clean = {}
        for key, val in data.items():
            lk = str(key or '').strip().lower()
            if not lk or lk in blocked:
                continue
            clean[str(key)] = str(val)
        return clean
    except Exception:
        return {}


def _addon_handle():
    now = time.time()
    if _SETTINGS_CACHE.get('addon') is not None and (now - _SETTINGS_CACHE.get('addon_ts', 0)) < 30:
        return _SETTINGS_CACHE.get('addon')
    try:
        _SETTINGS_CACHE['addon'] = xbmcaddon.Addon('plugin.video.oneplayvip')  # type: ignore[name-defined]
    except Exception:
        _SETTINGS_CACHE['addon'] = None
    _SETTINGS_CACHE['addon_ts'] = now
    return _SETTINGS_CACHE.get('addon')


def _settings_snapshot(force=False):
    now = time.time()
    if not force and (now - _SETTINGS_CACHE.get('settings_ts', 0)) < 10:
        return _SETTINGS_CACHE

    addon = _addon_handle()
    mode = 'stable'
    xtream_hardening = True
    host_memory = False
    debug = False
    try:
        if addon:
            raw = (addon.getSetting('proxy_mode') or '').strip().lower()
            enum_map = {'0': 'fast', '1': 'balance', '2': 'stable', 'rápido': 'fast', 'rapido': 'fast', 'balanceado': 'balance', 'estável': 'stable', 'estavel': 'stable'}
            raw = enum_map.get(raw, raw)
            if raw in _PROXY_POLICIES:
                mode = raw
            xtream_hardening = (addon.getSetting('proxy_xtream_hardening') or 'false').lower() == 'true'
            host_memory = (addon.getSetting('proxy_host_memory') or 'false').lower() == 'true'
            debug = (addon.getSetting('proxy_debug') or 'false').lower() == 'true'
    except Exception:
        pass

    _SETTINGS_CACHE.update({
        'settings_ts': now,
        'mode': mode,
        'xtream_hardening': xtream_hardening,
        'host_memory': host_memory,
        'debug': debug,
    })
    _policy_diag = _PROXY_POLICIES.get(mode, _PROXY_POLICIES['stable'])
    diagnostic_signature = (mode, bool(xtream_hardening), bool(host_memory))
    if _SETTINGS_CACHE.get('diagnostic_signature') != diagnostic_signature:
        _SETTINGS_CACHE['diagnostic_signature'] = diagnostic_signature
        _emit_log(
            '[Proxy] Configuração efetiva: perfil %s; tolerância Xtream=%s; memória de host=%s; '
            'VOD Range=%s%s, VOD HLS=%s%s, Live HLS=%s%s, Live seg.404=%s total, Live TS keepalive=%.1fs.' % (
                {'fast': 'Rápido', 'balance': 'Balanceado', 'stable': 'Estável'}.get(mode, 'Estável'),
                'ligada' if xtream_hardening else 'desligada',
                'ligada' if host_memory else 'desligada',
                int(_policy_diag.get('vod_max_retries', 1)), '+1 Xtream' if xtream_hardening else '',
                int(_policy_diag.get('vod_segment_max_retries', 2)), '+1 Xtream' if xtream_hardening else '',
                int(_policy_diag.get('live_hls_max_retries', 5)), '+1 Xtream' if xtream_hardening else '',
                min(4, int(_policy_diag.get('live_segment_404_retries', 2) or 0) + 1 + (1 if xtream_hardening else 0)),
                float(_policy_diag.get('ts_keepalive_window', 8.0) or 8.0) + (2.0 if xtream_hardening else 0.0),
            ),
            logging.INFO,
        )
    return _SETTINGS_CACHE


def _get_mode():
    mode = _settings_snapshot().get('mode', 'stable')
    return mode if mode in _PROXY_POLICIES else 'stable'


def _policy():
    return _PROXY_POLICIES.get(_get_mode(), _PROXY_POLICIES['stable'])


def _is_xtream_source(url):
    return _source_profile(url) in ('xtream_good', 'xtream_bad')


def _xtream_hardening_enabled(url):
    try:
        return bool(_settings_snapshot().get('xtream_hardening')) and _is_xtream_source(url)
    except Exception:
        return False


def _classify_source(url):
    try:
        parsed = urllib.parse.urlparse(url or '')
        host = (parsed.netloc or '').lower()
        path = (parsed.path or '').lower()
        query = (parsed.query or '').lower()
    except Exception:
        host = path = query = ''
    joined = '%s%s?%s' % (host, path, query)
    is_ts = path.endswith('.ts') or '/hls/' in path or '/hl' in path
    is_m3u8 = path.endswith('.m3u8') or 'mpegurl' in joined
    tokenized = 'token=' in query or 'wmsauth' in query or 'auth=' in query
    is_xtream = '/live/' in path or '/hls/' in path or '/movie/' in path or '/series/' in path or tokenized
    bad_xtream = is_xtream and (is_ts or is_m3u8) and tokenized
    if bad_xtream:
        return 'xtream_bad'
    if is_xtream:
        return 'xtream_good'
    if is_m3u8:
        return 'hls_generic'
    if is_ts:
        return 'ts_generic'
    return 'generic'


def _source_profile(url):
    profile = _classify_source(url)
    if _settings_snapshot().get('host_memory'):
        host = _host_from_url(url)
        cached = _SESSION_CACHE['source_profile'].get(host)
        if cached:
            profile = cached
    return profile


def _remember_source_profile(url, profile):
    host = _host_from_url(url)
    if host and profile:
        _SESSION_CACHE['source_profile'][host] = profile
        _trim_mapping(_SESSION_CACHE['source_profile'])


def _host_from_url(url):
    try:
        return urllib.parse.urlparse(url).netloc.lower()
    except Exception:
        return ''


def _trim_mapping(mapping, max_items=256):
    try:
        if not isinstance(mapping, dict) or len(mapping) <= max_items:
            return
        overflow = len(mapping) - max_items
        for key in list(mapping.keys())[:overflow]:
            mapping.pop(key, None)
    except Exception:
        pass


def _host_stats(host):
    item = _SESSION_CACHE['good_hosts'].get(host)
    if item is None:
        item = {'success_count': 0, 'fail_count': 0, 'cooldown_until': 0, 'last_error': '', 'last_success': 0}
        _SESSION_CACHE['good_hosts'][host] = item
    return item


def _host_is_available(host):
    if not host:
        return True
    return time.time() >= _host_stats(host).get('cooldown_until', 0)


def _host_backoff(host):
    if not host:
        return
    item = _host_stats(host)
    wait = item.get('cooldown_until', 0) - time.time()
    if wait > 0:
        time.sleep(min(0.8, wait))


def _mark_host_ok(host, final_url=None, user_agent=None, content_type=None):
    if not host or not _settings_snapshot().get('host_memory'):
        return
    item = _host_stats(host)
    item['success_count'] = item.get('success_count', 0) + 1
    item['fail_count'] = 0
    item['cooldown_until'] = 0
    item['last_error'] = ''
    item['last_success'] = time.time()
    _SESSION_CACHE['bad_hosts'].pop(host, None)
    if final_url:
        _SESSION_CACHE['last_working_url'][host] = final_url
    if user_agent:
        _SESSION_CACHE['working_user_agent'][host] = user_agent
    if content_type:
        _SESSION_CACHE['content_type'][host] = content_type
    _trim_mapping(_SESSION_CACHE['good_hosts'])
    _trim_mapping(_SESSION_CACHE['bad_hosts'])
    _trim_mapping(_SESSION_CACHE['last_working_url'])
    _trim_mapping(_SESSION_CACHE['working_user_agent'])
    _trim_mapping(_SESSION_CACHE['content_type'])


def _mark_host_fail(host, err='unknown', soft=False):
    if not host or not _settings_snapshot().get('host_memory'):
        return
    item = _host_stats(host)
    item['fail_count'] = item.get('fail_count', 0) + 1
    item['last_error'] = str(err)
    base = _policy().get('cooldown_base', 0.15)
    threshold = 5 if soft else 3
    if item['fail_count'] >= threshold:
        item['cooldown_until'] = time.time() + min(3.0, base * item['fail_count'])
        _SESSION_CACHE['bad_hosts'][host] = item['cooldown_until']
        _trim_mapping(_SESSION_CACHE['good_hosts'])
        _trim_mapping(_SESSION_CACHE['bad_hosts'])


def _smart_retry_delay(err_kind, attempt):
    if err_kind == '406':
        return min(0.45, 0.08 * max(1, attempt))
    if err_kind == 'timeout':
        return min(1.0, 0.20 * max(1, attempt))
    if err_kind == 'connection':
        return min(0.8, 0.16 * max(1, attempt))
    return min(0.75, 0.14 * max(1, attempt))


def _pick_user_agent(host, reconnects=0, prefer_stream=False):
    if host and reconnects <= 1:
        cached = _SESSION_CACHE['working_user_agent'].get(host)
        if cached:
            return cached
    idx = reconnects % len(_USER_AGENT_POOL)
    ua = _USER_AGENT_POOL[idx]
    if prefer_stream and reconnects >= 3:
        ua = 'Lavf/60.3.100'
    return ua


def _validate_upstream_response(response, expect_playlist=False, expect_stream=False, url=''):
    try:
        ctype = (response.headers.get('content-type') or '').lower()
    except Exception:
        ctype = ''
    if response.status_code not in (200, 206):
        return False, ctype
    if 'text/html' in ctype and not expect_playlist:
        return False, ctype
    if expect_stream and ('json' in ctype or 'xml' in ctype):
        return False, ctype
    profile = _source_profile(url)
    if expect_stream and profile == 'xtream_bad':
        # origem ruim costuma mentir no content-type; aceita octet/text genérico se a URL é claramente stream
        if not ctype or 'application/octet-stream' in ctype or 'video/' in ctype or 'audio/' in ctype or 'text/plain' in ctype:
            return True, ctype or 'application/octet-stream'
    return True, ctype


def _dns_override():
    """Aplica DNS alternativo somente quando o usuário habilitou a opção.

    O código anterior apenas instanciava ``customdns`` e não instalava o patch;
    no proxy persistente isso significava que o toggle não alterava a resolução.
    """
    try:
        addon = _addon_handle()
        enabled = bool(addon and (addon.getSetting('enable_custom_dns') or 'false').lower() == 'true')
    except Exception:
        enabled = False
    try:
        from resources.lib import dns as dns_runtime
        if enabled:
            return bool(dns_runtime.install(cache_ttl=1800, mode_logger=False))
        return bool(dns_runtime.uninstall())
    except Exception:
        return False


def _get_client_ip(headers, client_address):
    forwarded_for = headers.get('X-Forwarded-For')
    real_ip = headers.get('X-Real-IP')
    if forwarded_for:
        return forwarded_for.split(',')[0].strip()
    if real_ip:
        return real_ip
    return client_address[0] if client_address else '127.0.0.1'


def _get_cache_key(client_ip, url):
    return '%s:%s' % (client_ip, url)


def _is_valid_upstream_url(url):
    try:
        parsed = urllib.parse.urlparse(url or '')
        if (parsed.scheme or '').lower() not in ('http', 'https'):
            return False
        host = (parsed.hostname or '').strip().lower()
        if not host:
            return False
        if host in ('127.0.0.1', 'localhost', '::1'):
            return False
        if host.startswith('169.254.') or host.startswith('0.'):
            return False
        return True
    except Exception:
        return False


def _origin_key(url):
    try:
        p = urllib.parse.urlparse(url)
        return '%s://%s' % (p.scheme or 'http', p.netloc or '')
    except Exception:
        return url or 'unknown'


def _origin_state(url):
    key = _origin_key(url)
    state = _ORIGIN_HEALTH.get(key)
    now = time.time()
    if not state or (now - state.get('updated', 0)) > 180:
        state = {
            'updated': now,
            'fail_count': 0,
            '406_count': 0,
            'timeout_count': 0,
            'good_count': 0,
            'cooldown_until': 0,
            'last_error': ''
        }
        _ORIGIN_HEALTH[key] = state
        _trim_mapping(_ORIGIN_HEALTH, max_items=_RUNTIME_CACHE_LIMITS.get('origin_health', 128))
    return state


def _mark_origin_ok(url):
    state = _origin_state(url)
    state['updated'] = time.time()
    state['good_count'] = state.get('good_count', 0) + 1
    state['fail_count'] = 0
    state['406_count'] = 0
    state['timeout_count'] = 0
    state['last_error'] = ''
    state['cooldown_until'] = 0


def _mark_origin_fail(url, code=None, exc_name=None):
    state = _origin_state(url)
    state['updated'] = time.time()
    state['fail_count'] = state.get('fail_count', 0) + 1
    if code == 406:
        state['406_count'] = state.get('406_count', 0) + 1
    if exc_name in ('ReadTimeout', 'ConnectionError', 'ChunkedEncodingError'):
        state['timeout_count'] = state.get('timeout_count', 0) + 1
    state['last_error'] = str(code or exc_name or 'unknown')
    fail_count = state.get('fail_count', 0)
    if fail_count >= 6:
        cooldown = min(2.0, 0.15 * fail_count)
        state['cooldown_until'] = time.time() + cooldown


def _apply_origin_cooldown(url):
    state = _origin_state(url)
    now = time.time()
    cooldown_until = state.get('cooldown_until', 0)
    if cooldown_until > now:
        time.sleep(min(0.75, cooldown_until - now))


def _build_upstream_headers(original_headers, reconnects=0, response_code=None, prefer_stream=False, url='', force_range=None):
    headers = dict(original_headers or {})
    profile = _source_profile(url)
    headers.pop('Host', None)
    headers.pop('Connection', None)
    headers.pop('Content-Length', None)

    if reconnects > 0:
        headers.setdefault('User-Agent', DEFAULT_USER_AGENT)
        headers['Connection'] = 'close'

    if prefer_stream:
        headers.setdefault('Accept', '*/*')
        headers['Icy-MetaData'] = '1'
        headers['Accept-Encoding'] = 'identity'

    if _xtream_hardening_enabled(url):
        headers.pop('Range', None)
        headers.pop('If-None-Match', None)
        headers.pop('If-Modified-Since', None)
        headers['Accept'] = '*/*'
        headers['Accept-Encoding'] = 'identity'
        headers['Connection'] = 'close'

    if response_code == 406:
        for noisy in ('Accept', 'Accept-Language', 'Origin', 'Referer', 'Sec-Fetch-Dest', 'Sec-Fetch-Mode', 'Sec-Fetch-Site'):
            headers.pop(noisy, None)
        headers['Accept'] = '*/*'
        headers['Connection'] = 'close'
        if reconnects >= 2:
            headers.pop('Range', None)
        if reconnects >= 3 and profile != 'xtream_bad':
            headers['User-Agent'] = binascii.b2a_hex(os.urandom(20))[:32].decode('ascii')

    # Um 404 de segmento Live pode ocorrer antes de sua publicação. Nas poucas
    # retentativas específicas desse caso, sempre consulta a origem novamente e
    # evita reaproveitar uma resposta negativa de cache intermediário.
    if response_code == 404 and prefer_stream and _is_hls_media_like(url):
        headers['Cache-Control'] = 'no-cache'
        headers['Pragma'] = 'no-cache'
        headers['Connection'] = 'close'

    if reconnects >= 4:
        headers.pop('Accept-Encoding', None)
        headers['Cache-Control'] = 'no-cache'
        headers['Pragma'] = 'no-cache'

    # Range de retomada é uma decisão do proxy, nunca um header herdado.
    if force_range:
        headers['Range'] = str(force_range)
    return headers



def _timeout_tuple(url, prefer_stream=False):
    policy = _policy()
    connect = policy.get('timeout_connect', 5)
    read = policy.get('timeout_read', 15)
    profile = _source_profile(url)
    if _xtream_hardening_enabled(url):
        return (min(connect, 4), max(read, 20 if prefer_stream else 12))
    if profile == 'xtream_good':
        return (connect, max(read, 16 if prefer_stream else read))
    if profile == 'hls_generic' and prefer_stream:
        return (connect, max(read, 18))
    return (connect, read)


def _is_probably_live_segment(url):
    u = (url or '').lower()
    return '.ts' in u or '.m4s' in u or '.aac' in u or '.m4a' in u or '/hls/' in u or '/hl' in u


def _is_hls_media_like(url):
    u = (url or '').lower()
    return '.ts' in u or '.m4s' in u or '.aac' in u or '.m4a' in u or '/hls/' in u or '/hl' in u or '.mp4' in u


def _is_hls_media_response(url, content_type=''):
    """Reconhece segmentos Live/HLS mesmo quando a URL não tem extensão.

    Alguns Xtream expõem `/live/user/pass/id` sem `.ts`. Nesses casos o
    Content-Type é a fonte segura para escolher a rota de segmento e impedir
    que a antiga cauda em cache seja usada.
    """
    if _is_hls_media_like(url):
        return True
    ctype = str(content_type or '').split(';', 1)[0].strip().lower()
    return ctype in (
        'video/mp2t', 'video/mp4', 'audio/mp4', 'audio/aac',
        'application/mp4', 'application/vnd.apple.mpegurl-segment',
        'video/iso.segment',
    )


def _is_hls_playlist_url(url):
    try:
        path = (urllib.parse.urlparse(url or '').path or '').lower()
        return path.endswith('.m3u8') or '.m3u8' in (url or '').lower()
    except Exception:
        return '.m3u8' in (url or '').lower()


def _hls_playlist_cache_key(url, kind='live'):
    try:
        parsed = urllib.parse.urlparse(url or '')
        base = urllib.parse.urlunparse((parsed.scheme, parsed.netloc, parsed.path, '', parsed.query, ''))
        return '%s|%s' % ('vod' if str(kind).lower() == 'vod' else 'live', base)
    except Exception:
        return '%s|%s' % ('vod' if str(kind).lower() == 'vod' else 'live', url or '')


def _hls_playlist_hold_seconds(raw_playlist=None):
    hold = float(_policy().get('hls_playlist_hold', 8.0) or 8.0)
    try:
        if raw_playlist:
            m = re.search(r'#EXT-X-TARGETDURATION\s*:\s*(\d+)', raw_playlist)
            if m:
                target = int(m.group(1))
                hold = max(hold, min(16.0, target * 2.0))
    except Exception:
        pass
    return max(4.0, min(18.0, hold))


def _store_hls_playlist(url, raw_playlist, base_url, content_type='application/vnd.apple.mpegurl', kind='live'):
    try:
        if not url or not raw_playlist or '#EXTM3U' not in raw_playlist[:256].upper():
            return
        key = _hls_playlist_cache_key(url, kind=kind)
        _HLS_PLAYLIST_CACHE[key] = {
            'ts': time.time(),
            'raw': raw_playlist,
            'base_url': base_url,
            'content_type': content_type or 'application/vnd.apple.mpegurl',
            'hold': _hls_playlist_hold_seconds(raw_playlist),
        }
        if len(_HLS_PLAYLIST_CACHE) > 32:
            oldest = sorted(_HLS_PLAYLIST_CACHE.items(), key=lambda kv: kv[1].get('ts', 0))[:8]
            for old_key, _ in oldest:
                _HLS_PLAYLIST_CACHE.pop(old_key, None)
    except Exception:
        pass


def _get_hls_playlist_cache(url, kind='live'):
    try:
        item = _HLS_PLAYLIST_CACHE.get(_hls_playlist_cache_key(url, kind=kind))
        if not item:
            return None
        age = time.time() - item.get('ts', 0)
        hold = float(item.get('hold') or _hls_playlist_hold_seconds(item.get('raw')))
        if age <= hold:
            return item
    except Exception:
        return None
    return None


def _hls_chunk_size(url):
    try:
        size = int(_policy().get('hls_chunk_size', 32768) or 32768)
    except Exception:
        size = 32768
    return max(8192, min(131072, size))


def _ts_stream_chunk_size():
    try:
        base = int(_policy().get('hls_chunk_size', 32768) or 32768)
    except Exception:
        base = 32768
    # TS puro se beneficia de blocos um pouco maiores que 4 KB,
    # mas sem exagerar para não piorar a responsividade do retry.
    return max(8192, min(65536, base // 2 if base > 16384 else 16384))


def _live_hls_chunk_size(url=''):
    """Bloco curto para Live HLS entregar bytes antes de uma microqueda.

    O VOD pode ler blocos maiores porque só expõe segmento íntegro. No Live,
    blocos de 16 KiB evitam que uma queda antes de 64 KiB faça o requests
    esconder bytes já recebidos e obrigue a retomada do zero.
    """
    try:
        base = int(_policy().get('hls_chunk_size', 32768) or 32768)
    except Exception:
        base = 32768
    return max(8192, min(16384, base))


def _is_local_proxy_uri(uri):
    try:
        p = urllib.parse.urlparse(uri or '')
        host = (p.hostname or '').lower()
        return host in ('127.0.0.1', 'localhost', '::1') and p.path in ('/hlsretry', '/tsdownloader')
    except Exception:
        return False


def _proxify_hls_uri(uri, base_url, host, header_blob='', kind='live'):
    uri = (uri or '').strip()
    if not uri or uri.startswith('#'):
        return uri
    low = uri.lower()
    if low.startswith(('data:', 'skd:', 'urn:', 'about:', 'javascript:')):
        return uri
    if _is_local_proxy_uri(uri):
        return uri
    try:
        absolute_url = urljoin(base_url + '/', uri)
        if not _is_valid_upstream_url(absolute_url):
            return uri
        target_kind = 'vod' if str(kind or '').lower() == 'vod' else 'live'
        token = register_proxy_target(
            absolute_url,
            header_blob=header_blob,
            kind=target_kind,
            persist=False,
        )
        if not token:
            return uri
        try:
            path_lower = (urllib.parse.urlparse(absolute_url).path or '').lower()
            if '.m3u8' in path_lower:
                media_type = 'm3u8'
            else:
                media_type = path_lower.rsplit('.', 1)[-1] if '.' in path_lower else 'ts'
                if media_type not in ('ts', 'm4s', 'mp4', 'aac', 'm4a'):
                    media_type = 'ts'
        except Exception:
            media_type = 'ts'
        proxied_url = 'http://%s/hlsretry?token=%s&type=%s' % (host, token, media_type)
        return proxied_url
    except Exception:
        return uri


def _rewrite_m3u8_urls(playlist_content, base_url, host, header_blob='', kind='live'):
    """Reescreve URIs HLS sem alterar tags.

    Cobre playlists IPTV comuns e HLS mais completo: segmentos em linha,
    variantes, EXT-X-KEY, EXT-X-MAP, EXT-X-MEDIA e I-FRAME via URI="...".
    """
    def rewrite_uri_attr(match):
        prefix, uri, suffix = match.group(1), match.group(2), match.group(3)
        return prefix + _proxify_hls_uri(uri, base_url, host, header_blob=header_blob, kind=kind) + suffix

    out_lines = []
    for raw_line in (playlist_content or '').splitlines():
        line = raw_line.strip()
        if not line:
            out_lines.append(raw_line)
            continue
        if line.startswith('#'):
            out_lines.append(re.sub(r'(URI=")([^"]+)(")', rewrite_uri_attr, raw_line))
            continue
        leading = raw_line[:len(raw_line) - len(raw_line.lstrip())]
        trailing = raw_line[len(raw_line.rstrip()):]
        out_lines.append(leading + _proxify_hls_uri(line, base_url, host, header_blob=header_blob, kind=kind) + trailing)
    suffix = '\n' if (playlist_content or '').endswith(('\n', '\r')) else ''
    return '\n'.join(out_lines) + suffix


def _stream_cache_has(client_ip, url):
    if not url:
        return False
    cache_key = _get_cache_key(client_ip, url) if any(ext in url.lower() for ext in ['.mp4', '.m3u8']) else client_ip
    if '.mp4' in url.lower():
        return bool(IP_CACHE_MP4.get(cache_key))
    if _is_hls_media_like(url):
        return bool(IP_CACHE_TS.get(cache_key))
    return False


def _stream_response(response, client_ip, url, sess):
    cache_key = _get_cache_key(client_ip, url) if any(ext in url.lower() for ext in ['.mp4', '.m3u8']) else client_ip
    mode_ts = [_is_hls_media_like(url) and '.mp4' not in (url or '').lower()]

    def generate_chunks():
        bytes_read = 0
        try:
            for chunk in response.iter_content(chunk_size=_hls_chunk_size(url)):
                if not chunk:
                    continue
                bytes_read += len(chunk)
                if '.mp4' in url.lower():
                    if cache_key not in IP_CACHE_MP4:
                        _trim_mapping(IP_CACHE_MP4, max_items=_RUNTIME_CACHE_LIMITS.get('stream_cache_keys', 24))
                        IP_CACHE_MP4[cache_key] = []
                    IP_CACHE_MP4[cache_key].append(chunk)
                    if len(IP_CACHE_MP4[cache_key]) > 20:
                        IP_CACHE_MP4[cache_key].pop(0)
                elif _is_hls_media_like(url):
                    mode_ts[0] = True
                    if cache_key not in IP_CACHE_TS:
                        _trim_mapping(IP_CACHE_TS, max_items=_RUNTIME_CACHE_LIMITS.get('stream_cache_keys', 24))
                        IP_CACHE_TS[cache_key] = []
                    IP_CACHE_TS[cache_key].append(chunk)
                    if len(IP_CACHE_TS[cache_key]) > 20:
                        IP_CACHE_TS[cache_key].pop(0)
                yield chunk
        except (IncompleteRead, ConnectionError, ChunkedEncodingError, ReadTimeout) as e:
            _throttled_log('hls-tail-fallback:%s' % _origin_key(url), 15, logging.INFO, '[HLS Proxy] Segmento interrompido; reaproveitando cauda recente para atravessar microfalha.')
            cache = IP_CACHE_TS if mode_ts[0] else IP_CACHE_MP4
            for chunk in cache.get(cache_key, [])[-5:]:
                yield chunk
        finally:
            try:
                sess.close()
            except Exception:
                pass
    return generate_chunks()


def _response_header(response, name, default=''):
    """Lê header sem depender da implementação (urllib/requests)."""
    try:
        headers = getattr(response, 'headers', {}) or {}
        return headers.get(name) or headers.get(name.lower()) or default
    except Exception:
        return default


def _parse_requested_byte_range(value):
    """Aceita somente Range simples; múltiplos ranges não são seguros para relay."""
    try:
        match = re.match(r'^\s*bytes=(\d+)-(\d*)\s*$', str(value or ''), re.I)
        if not match:
            return None, None
        start = int(match.group(1))
        end = int(match.group(2)) if match.group(2) else None
        return start, end
    except Exception:
        return None, None


def _parse_content_range(value):
    """Retorna start, end e total de um Content-Range válido, ou Nones."""
    try:
        match = re.match(r'^\s*bytes\s+(\d+)-(\d+)/(\d+|\*)\s*$', str(value or ''), re.I)
        if not match:
            return None, None, None
        start = int(match.group(1))
        end = int(match.group(2))
        total = None if match.group(3) == '*' else int(match.group(3))
        if start > end:
            return None, None, None
        return start, end, total
    except Exception:
        return None, None, None


def _positive_int(value):
    try:
        number = int(str(value or '').strip())
        return number if number >= 0 else None
    except Exception:
        return None


def _vod_retry_delay(attempt):
    """Backoff curto: oculta microqueda, mas não transforma falha definitiva em espera sem fim."""
    try:
        index = max(1, int(attempt))
    except Exception:
        index = 1
    schedule = (0.30, 0.80, 1.50, 2.50)
    return schedule[min(index - 1, len(schedule) - 1)]


def _vod_recovery_settings(url):
    policy = _policy()
    try:
        retries = int(policy.get('vod_max_retries', 2) or 0)
    except Exception:
        retries = 2
    try:
        segment_retries = int(policy.get('vod_segment_max_retries', 3) or 0)
    except Exception:
        segment_retries = 3
    retries = max(0, min(4, retries))
    segment_retries = max(0, min(5, segment_retries))
    if _xtream_hardening_enabled(url):
        retries = min(4, retries + 1)
        segment_retries = min(5, segment_retries + 1)
    try:
        timeout = max(float(policy.get('timeout_connect', 5) or 5), float(policy.get('timeout_read', 15) or 15))
    except Exception:
        timeout = 15.0
    return {
        'max_retries': retries,
        'segment_max_retries': segment_retries,
        'timeout': max(5.0, min(25.0, timeout)),
    }


def _hls_retry_settings(url, kind='live'):
    """Orçamento de abertura de playlist/segmento antes de expor erro ao player."""
    policy = _policy()
    is_vod = str(kind or '').lower() == 'vod'
    key = 'vod_playlist_max_retries' if is_vod else 'live_hls_max_retries'
    try:
        retries = int(policy.get(key, policy.get('hls_max_retries', 5)) or 0)
    except Exception:
        retries = 5
    if _xtream_hardening_enabled(url):
        retries += 1
    return {'max_retries': max(1, min(12, retries))}


def _live_hls_segment_settings(url):
    policy = _policy()
    try:
        retries = int(policy.get('live_segment_max_retries', 3) or 0)
    except Exception:
        retries = 3
    if _xtream_hardening_enabled(url):
        retries += 1
    return {'max_retries': max(0, min(5, retries))}


def _live_hls_segment_404_settings(url):
    """Orçamento curto para um segmento Live que responde HTTP 404.

    404 de segmento não usa o orçamento longo da playlist. O segmento pode ainda
    estar sendo publicado, portanto recebe poucas retentativas curtas; persistindo
    o 404, devolvemos o status ao player para que ele reabra a playlist e siga a
    janela atual, em vez de insistir numa URL que já expirou.
    """
    policy = _policy()
    try:
        retries = int(policy.get('live_segment_404_retries', 2) or 0)
    except Exception:
        retries = 2
    # A tolerância Xtream acrescenta no máximo uma tentativa, sem ultrapassar o
    # teto operacional de quatro tentativas totais (original + três retries).
    if _xtream_hardening_enabled(url):
        retries = min(3, retries + 1)
    retries = max(0, min(3, retries))
    return {
        'retries': retries,
        'total_attempts': retries + 1,
    }


def _live_hls_segment_404_delay(retry_number):
    """Atraso progressivo curto: 0,30s, 0,70s e 1,20s."""
    try:
        index = max(1, int(retry_number))
    except Exception:
        index = 1
    schedule = (0.30, 0.70, 1.20)
    return schedule[min(index - 1, len(schedule) - 1)]


def _live_ts_settings(url):
    policy = _policy()
    try:
        window = float(policy.get('ts_keepalive_window', 8.0) or 8.0)
    except Exception:
        window = 8.0
    try:
        packets = int(policy.get('ts_null_packets', 32) or 32)
    except Exception:
        packets = 32
    try:
        fill_sleep = float(policy.get('ts_fill_sleep', 0.06) or 0.06)
    except Exception:
        fill_sleep = 0.06
    try:
        reprobe_every = int(policy.get('live_ts_reprobe_every', 4) or 4)
    except Exception:
        reprobe_every = 4
    if _xtream_hardening_enabled(url):
        window = min(16.0, window + 2.0)
        packets = min(96, packets + 12)
        fill_sleep = min(0.14, fill_sleep + 0.01)
        reprobe_every = min(reprobe_every, 3)
    return {
        'keepalive_window': max(2.0, window),
        'null_packets': max(6, min(96, packets)),
        'fill_sleep': max(0.02, min(0.18, fill_sleep)),
        'reprobe_every': max(2, min(6, reprobe_every)),
        'startup_keepalive_window': float(policy.get('ts_startup_keepalive_window', 0.0) or 0.0),
        'startup_read_timeout': float(policy.get('ts_startup_read_timeout', 5.0) or 5.0),
        'startup_handoff_window': float(policy.get('ts_startup_handoff_window', 5.0) or 5.0),
        'startup_handoff_bytes': int(policy.get('ts_startup_handoff_bytes', 393216) or 393216),
    }


def _read_hls_segment_to_spool(response, url):
    """Lê um segmento por inteiro antes de o expor no VOD HLS.

    Isso é intencionalmente diferente do live: em VOD, integridade vale mais do
    que despejar bytes cedo e depois tentar compensar com dados repetidos.
    """
    expected = _positive_int(_response_header(response, 'Content-Length', ''))
    spool = tempfile.SpooledTemporaryFile(max_size=2 * 1024 * 1024, mode='w+b')
    written = 0
    try:
        for chunk in response.iter_content(chunk_size=_hls_chunk_size(url)):
            if not chunk:
                continue
            spool.write(chunk)
            written += len(chunk)
        if expected is not None and written != expected:
            raise IOError('segmento_hls_incompleto:%d/%d' % (written, expected))
        if written <= 0:
            raise IOError('segmento_hls_vazio')
        spool.seek(0)
        return spool, written
    except Exception:
        try:
            spool.close()
        except Exception:
            pass
        raise


def _buffer_vod_hls_segment(session, initial_response, url, original_headers, max_retries):
    """Baixa/rebaixa um segmento VOD HLS com integridade antes do player vê-lo.

    Nunca usa cauda de cache e nunca concatena tentativa parcial com segmento
    reiniciado. Se a origem não entregar um corpo completo, o segmento inteiro é
    solicitado de novo antes da resposta local ser iniciada.
    """
    response = initial_response
    attempt = 0  # 0 = resposta já aberta; os demais são novas tentativas.
    last_status = None
    while attempt <= max_retries:
        if response is None:
            try:
                _apply_origin_cooldown(url)
                headers = _build_upstream_headers(
                    original_headers,
                    reconnects=attempt,
                    response_code=last_status,
                    prefer_stream=True,
                    url=url
                )
                headers.setdefault('Accept-Encoding', 'identity')
                candidate = session.get(
                    url,
                    headers=headers,
                    allow_redirects=True,
                    stream=True,
                    timeout=_timeout_tuple(url, prefer_stream=True)
                )
                ok_response, _ctype = _validate_upstream_response(candidate, expect_stream=True, url=url)
                if not ok_response:
                    code = getattr(candidate, 'status_code', 0)
                    try:
                        candidate.close()
                    except Exception:
                        pass
                    raise IOError('segmento_hls_http_%s' % code)
                response = candidate
                _mark_origin_ok(getattr(candidate, 'url', '') or url)
            except Exception as open_exc:
                last_status = open_exc.__class__.__name__
                _mark_origin_fail(url, exc_name=last_status)
                _mark_host_fail(_host_from_url(url), err=last_status, soft=True)
                if attempt >= max_retries:
                    break
                attempt += 1
                _throttled_log(
                    'vod-hls-segment-open-retry:%s' % _origin_key(url),
                    15,
                    logging.INFO,
                    '[VOD HLS] Origem oscilou antes do segmento; retry seguro (%d/%d).' % (attempt, max_retries)
                )
                time.sleep(_vod_retry_delay(attempt))
                continue

        try:
            spool, size = _read_hls_segment_to_spool(response, url)
            result = {
                'spool': spool,
                'size': size,
                'status': int(getattr(response, 'status_code', 200) or 200),
                'headers': dict(getattr(response, 'headers', {}) or {}),
                'url': getattr(response, 'url', '') or url,
            }
            try:
                response.close()
            except Exception:
                pass
            return result
        except Exception as exc:
            last_status = exc.__class__.__name__
            _mark_origin_fail(url, exc_name=last_status)
            _mark_host_fail(_host_from_url(url), err=last_status, soft=True)
            try:
                response.close()
            except Exception:
                pass
            response = None
            if attempt >= max_retries:
                break
            attempt += 1
            _throttled_log(
                'vod-hls-segment-retry:%s' % _origin_key(url),
                15,
                logging.INFO,
                '[VOD HLS] Microqueda no segmento; novo download íntegro em curso (%d/%d).' % (attempt, max_retries)
            )
            time.sleep(_vod_retry_delay(attempt))

    _throttled_log(
        'vod-hls-segment-fail:%s' % _origin_key(url),
        15,
        logging.WARNING,
        '[VOD HLS] Segmento incompleto após retomadas seguras (%s).' % _safe_text(last_status)
    )
    return None


def _relay_live_hls_segment(handler, session, initial_response, url, original_headers, media_type, response_headers, head_only=False):
    """Relaya um segmento HLS Live sem repetir cauda já enviada.

    Live não pode usar o mesmo spool de VOD, porque isso adicionaria latência a
    cada segmento. Quando a origem corta durante o corpo, o proxy solicita só a
    faixa faltante e a anexa exclusivamente se receber ``206 Content-Range`` no
    byte exato. Sem suporte confiável a Range, encerra o segmento sem fabricar
    pacotes; o player segue para a próxima requisição de playlist/segmento.
    """
    response = initial_response
    active_url = getattr(response, 'url', '') or url
    status = int(getattr(response, 'status_code', 200) or 200)
    content_length = _positive_int(_response_header(response, 'Content-Length', ''))
    content_range = _response_header(response, 'Content-Range', '')
    range_start, range_end, _range_total = _parse_content_range(content_range)
    stream_start = range_start if range_start is not None else 0
    expected_bytes = content_length
    if expected_bytes is None and range_start is not None and range_end is not None:
        expected_bytes = (range_end - range_start) + 1
    stream_end = (stream_start + expected_bytes - 1) if expected_bytes is not None else None

    handler.send_response(206 if status == 206 else 200)
    handler.send_header('Content-Type', media_type)
    if expected_bytes is not None:
        handler.send_header('Content-Length', str(expected_bytes))
    for key, value in (response_headers or {}).items():
        lower = str(key).lower()
        if lower in ('content-type', 'content-length', 'transfer-encoding', 'connection'):
            continue
        if lower in ('accept-ranges', 'content-range') and status != 206:
            continue
        if lower in ('accept-ranges', 'content-range'):
            handler.send_header(key, value)
    handler.send_header('Cache-Control', 'no-store')
    handler.send_header('Connection', 'close')
    handler.end_headers()
    if head_only:
        try:
            response.close()
        except Exception:
            pass
        return

    delivered = 0
    retries = 0
    max_retries = int(_live_hls_segment_settings(active_url).get('max_retries', 0) or 0)
    try:
        while True:
            failure = None
            try:
                for chunk in response.iter_content(chunk_size=_live_hls_chunk_size(active_url)):
                    if not chunk:
                        continue
                    if expected_bytes is not None:
                        remaining = expected_bytes - delivered
                        if remaining <= 0:
                            return
                        if len(chunk) > remaining:
                            chunk = chunk[:remaining]
                    try:
                        _write_client_chunk(handler, chunk, flush=(delivered == 0 or delivered < 131072))
                    except (BrokenPipeError, ConnectionResetError, OSError):
                        return
                    delivered += len(chunk)
                    if expected_bytes is not None and delivered >= expected_bytes:
                        return
                if expected_bytes is None or delivered >= expected_bytes:
                    return
                failure = IOError('live_hls_eof_antes_do_content_length')
            except Exception as exc:
                failure = exc

            if failure is None:
                return
            _mark_origin_fail(active_url, exc_name=failure.__class__.__name__)
            _mark_host_fail(_host_from_url(active_url), err=failure.__class__.__name__, soft=True)
            if expected_bytes is None:
                _throttled_log(
                    'live-hls-no-length:%s' % _origin_key(active_url), 20, logging.INFO,
                    '[Live HLS] Segmento caiu sem Content-Length; não há retomada por byte segura.'
                )
                return
            if retries >= max_retries:
                _throttled_log(
                    'live-hls-retry-exhausted:%s' % _origin_key(active_url), 15, logging.INFO,
                    '[Live HLS] Segmento não retomou dentro do orçamento; seguindo sem bytes duplicados.'
                )
                return

            retries += 1
            resume_at = stream_start + delivered
            resume_range = 'bytes=%d-' % resume_at if stream_end is None else 'bytes=%d-%d' % (resume_at, stream_end)
            _throttled_log(
                'live-hls-range-resume:%s' % _origin_key(active_url), 12, logging.INFO,
                '[Live HLS] Microqueda detectada; retomando segmento por Range (%d/%d).' % (retries, max_retries)
            )
            try:
                response.close()
            except Exception:
                pass
            time.sleep(_vod_retry_delay(retries))
            _apply_origin_cooldown(active_url)
            headers = _build_upstream_headers(
                original_headers,
                reconnects=retries,
                prefer_stream=True,
                url=active_url,
                force_range=resume_range
            )
            candidate = None
            try:
                candidate = session.get(
                    active_url,
                    headers=headers,
                    allow_redirects=True,
                    stream=True,
                    timeout=_timeout_tuple(active_url, prefer_stream=True)
                )
                candidate_status = int(getattr(candidate, 'status_code', 0) or 0)
                candidate_range = _response_header(candidate, 'Content-Range', '')
                candidate_start, candidate_end, _candidate_total = _parse_content_range(candidate_range)
                remaining = expected_bytes - delivered
                candidate_length = _positive_int(_response_header(candidate, 'Content-Length', ''))
                range_ok = (
                    candidate_status == 206 and
                    candidate_start == resume_at and
                    candidate_end is not None and
                    candidate_end >= (resume_at + max(0, remaining - 1)) and
                    (candidate_length is None or candidate_length >= remaining)
                )
                if not range_ok:
                    _mark_origin_fail(active_url, code=candidate_status)
                    _throttled_log(
                        'live-hls-range-rejected:%s' % _origin_key(active_url), 15, logging.WARNING,
                        '[Live HLS] Origem não confirmou Range exato; evitando repetir pacotes do segmento.'
                    )
                    return
                response = candidate
                candidate = None
                try:
                    active_url = response.url or active_url
                except Exception:
                    pass
                _mark_origin_ok(active_url)
            except Exception as open_exc:
                _mark_origin_fail(active_url, exc_name=open_exc.__class__.__name__)
                if retries >= max_retries:
                    _throttled_log(
                        'live-hls-range-open-fail:%s' % _origin_key(active_url), 15, logging.INFO,
                        '[Live HLS] Retomada do segmento não respondeu; mantendo integridade do stream.'
                    )
                    return
                continue
            finally:
                if candidate is not None:
                    try:
                        candidate.close()
                    except Exception:
                        pass
    finally:
        try:
            response.close()
        except Exception:
            pass


_TS_NULL_PACKET = b"\x47\x1f\xff\x10" + (b"\xff" * 184)


def _ts_null_chunk(packet_count=None, url=''):
    try:
        packets = int(packet_count or _live_ts_settings(url).get('null_packets', 40))
    except Exception:
        packets = 40
    packets = max(1, min(256, packets))
    return _TS_NULL_PACKET * packets


def _should_ts_keepalive(last_good_at, startup_keepalive_until=0, url=''):
    # Conservador: só injeta null packets depois que já houve tráfego real.
    # Antes do primeiro chunk bom, filler tende a piorar continuidade/PTS em algumas origens.
    if last_good_at:
        window = float(_live_ts_settings(url).get('keepalive_window', 5.5) or 5.5)
        return (time.time() - last_good_at) <= max(1.5, window)
    return False


def _ts_keepalive_once(last_good_at, reconnects=0, startup_keepalive_until=0, url=''):
    if not _should_ts_keepalive(last_good_at, startup_keepalive_until=startup_keepalive_until, url=url):
        return None
    base_packets = int(_live_ts_settings(url).get('null_packets', 20) or 20)
    # sobe leve durante reconnects repetidos, sem virar mangueira de lixo
    packets = min(96, max(6, base_packets + (min(reconnects, 4) * 4)))
    return _ts_null_chunk(packets, url=url)


def _ts_keepalive_sleep(reconnects=0, url=''):
    base = float(_live_ts_settings(url).get('fill_sleep', 0.05) or 0.05)
    return min(0.18, base + (min(reconnects, 4) * 0.01))


def _stream_cache(client_ip, url):
    if not url:
        return None
    cache_key = _get_cache_key(client_ip, url) if any(ext in url.lower() for ext in ['.mp4', '.m3u8']) else client_ip
    if '.mp4' in url.lower():
        cache = IP_CACHE_MP4
    elif _is_hls_media_like(url):
        cache = IP_CACHE_TS
    else:
        cache = None
    if not cache:
        return None

    def generate_cached_chunks():
        if cache_key in cache:
            for chunk in cache.get(cache_key, [])[-5:]:
                yield chunk
        else:
            logging.debug('[HLS Proxy] Cache vazio para %s' % cache_key)
    return generate_cached_chunks()


def _clear_transient_runtime_state():
    for mapping in (IP_CACHE_TS, IP_CACHE_MP4, AGENT_OF_CHAOS, COUNT_CLEAR, _TS_STATUS_LOG, _ORIGIN_HEALTH, _LOG_BURSTS, _HLS_PLAYLIST_CACHE):
        try:
            mapping.clear()
        except Exception:
            pass
    # Um novo start merece uma única fotografia da configuração efetiva.
    _SETTINGS_CACHE.pop('diagnostic_signature', None)
    with _TARGET_LOCK:
        _TARGET_REGISTRY.clear()


def _write_client_chunk(handler, chunk, flush=False):
    handler.wfile.write(chunk)
    if flush:
        try:
            handler.wfile.flush()
        except Exception:
            pass


class _ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
    daemon_threads = True
    allow_reuse_address = True


class _ProxyHandler(BaseHTTPRequestHandler):
    protocol_version = 'HTTP/1.1'

    def handle(self):
        try:
            BaseHTTPRequestHandler.handle(self)
        except (ConnectionResetError, BrokenPipeError, OSError):
            pass

    def finish(self):
        try:
            BaseHTTPRequestHandler.finish(self)
        except (ConnectionResetError, BrokenPipeError, OSError):
            pass

    def log_message(self, format, *args):
        return

    def _send_json(self, payload, status=200):
        data = payload.encode('utf-8') if isinstance(payload, str) else payload
        self.send_response(status)
        self.send_header('Content-Type', 'application/json; charset=utf-8')
        self.send_header('Content-Length', str(len(data)))
        self.send_header('Connection', 'close')
        self.end_headers()
        if self.command != 'HEAD':
            self.wfile.write(data)

    def _send_text(self, payload, status=200, content_type='text/plain; charset=utf-8'):
        data = payload.encode('utf-8') if isinstance(payload, str) else payload
        self.send_response(status)
        self.send_header('Content-Type', content_type)
        self.send_header('Content-Length', str(len(data)))
        self.send_header('Connection', 'close')
        self.end_headers()
        if self.command != 'HEAD':
            self.wfile.write(data)

    def _send_cached_hls_playlist(self, url, header_blob='', kind='live'):
        item = _get_hls_playlist_cache(url, kind=kind)
        if not item:
            return False
        try:
            host = self.headers.get('Host', '127.0.0.1:%d' % PORT)
            base_url = item.get('base_url') or url.rsplit('/', 1)[0]
            rewritten = _rewrite_m3u8_urls(item.get('raw') or '', base_url, host, header_blob=header_blob, kind=kind)
            self._send_text(rewritten, content_type='application/vnd.apple.mpegurl')
            _throttled_log('hls-playlist-cache:%s' % _origin_key(url), 20, logging.INFO, '[HLS Proxy] Playlist anterior reutilizada para atravessar instabilidade curta.')
            return True
        except Exception:
            return False

    def do_GET(self):
        self._dispatch()

    def do_HEAD(self):
        self._dispatch(head_only=True)

    def _dispatch(self, head_only=False):
        parsed = urllib.parse.urlparse(self.path)
        path = parsed.path
        if path == PROXY_HEALTH_PATH:
            self._send_json(json.dumps({'service': PROXY_SERVICE_ID, 'port': int(PORT), 'version': 2}, separators=(',', ':')))
            return
        if path == '/':
            # VOD no modelo XC Pro: URL direta na raiz do proxy.
            # A origem chega por token opaco; ``url=`` permanece só como fallback legado.
            # Mantém /hlsretry e /tsdownloader intactos para live/HLS.
            root_params = urllib.parse.parse_qs(parsed.query)
            if parsed.query and (root_params.get('url') or root_params.get('token')):
                self._handle_vod_root_proxy(parsed, head_only=head_only)
            else:
                self._send_json('{"message": "OnePlay Proxy nativo ativo"}')
            return
        if path == '/hlsretry':
            self._handle_hlsretry(parsed, head_only=head_only)
            return
        if path == '/tsdownloader':
            self._handle_tsdownloader(parsed, head_only=head_only)
            return
        self._send_text('Not Found', status=404)

    def _filtered_request_headers(self):
        upstream = {}
        for k, v in self.headers.items():
            lk = k.lower()
            if lk in ('host', 'content-length', 'connection'):
                continue
            upstream[k] = v
        return upstream

    def _handle_hlsretry(self, parsed, head_only=False):
        _dns_override()
        params = urllib.parse.parse_qs(parsed.query)
        url, header_blob, target_kind = _target_from_query_params(params)
        kind = (target_kind or 'live').strip().lower()
        kind = 'vod' if kind == 'vod' else 'live'
        is_vod_hls = kind == 'vod'
        client_ip = _get_client_ip(self.headers, self.client_address)
        cache_key = _get_cache_key(client_ip, url) if url and any(x in url.lower() for x in ['.mp4', '.m3u8']) else client_ip
        if not url:
            self._send_json('{"detail": "No URL provided"}', status=400)
            return
        if not _is_valid_upstream_url(url):
            self._send_json('{"detail": "Invalid upstream URL"}', status=400)
            return

        session = requests.Session()
        original_headers = self._filtered_request_headers()
        original_headers.update(_decode_header_blob(header_blob))

        max_retries = _hls_retry_settings(url, kind=kind).get('max_retries', 7)
        attempts = 0
        tried_without_range = False
        media_type = 'video/mp4' if '.mp4' in url.lower() else 'video/mp2t' if '.ts' in url.lower() or '/hl' in url.lower() else 'application/octet-stream'
        response_headers = {}
        status = 200
        last_status = None
        # Conta somente 404 consecutivos de um segmento Live. Playlist, conexão,
        # 406 e demais respostas continuam usando a política geral já aprovada.
        live_segment_404_failures = 0

        try:
            while attempts < max_retries:
                response = None
                headers = _build_upstream_headers(original_headers, reconnects=attempts, response_code=last_status, url=url)
                try:
                    _apply_origin_cooldown(url)
                    range_header = headers.get('Range')
                    if '.mp4' in url.lower() and range_header and tried_without_range:
                        headers.pop('Range', None)
    
                    if AGENT_OF_CHAOS.get(cache_key) and '.ts' not in url.lower() and '/hl' not in url.lower():
                        headers['User-Agent'] = AGENT_OF_CHAOS.get(cache_key, DEFAULT_USER_AGENT)
                    elif '.ts' in url.lower() or '/hl' in url.lower():
                        headers.setdefault('User-Agent', DEFAULT_USER_AGENT)
    
                    host = _host_from_url(url)
                    if not _host_is_available(host):
                        _host_backoff(host)
                    headers['User-Agent'] = _pick_user_agent(host, attempts, prefer_stream=('.ts' in url.lower() or '/hl' in url.lower()))
                    response = session.get(url, headers=headers, allow_redirects=True, stream=True, timeout=_timeout_tuple(url, prefer_stream=(_is_probably_live_segment(url) or '.mp4' in url.lower())))
    
                    ok_response, content_type = _validate_upstream_response(response, expect_playlist=('.m3u8' in url.lower()), expect_stream=('.ts' in url.lower() or '/hl' in url.lower() or '.mp4' in url.lower()), url=url)
                    if ok_response:
                        _mark_origin_ok(response.url or url)
                        _mark_host_ok(_host_from_url(response.url or url), final_url=(response.url or url), user_agent=headers.get('User-Agent'), content_type=content_type)
                        _remember_source_profile(response.url or url, _classify_source(response.url or url))
                        if '.mp4' in url.lower() or '.m3u8' in url.lower():
                            url = response.url
                        if client_ip in COUNT_CLEAR:
                            if COUNT_CLEAR.get(client_ip, 0) > 4:
                                try:
                                    AGENT_OF_CHAOS.pop(cache_key, None)
                                    IP_CACHE_MP4.pop(cache_key, None)
                                    IP_CACHE_TS.pop(cache_key, None)
                                except Exception:
                                    pass
                        if client_ip not in COUNT_CLEAR:
                            COUNT_CLEAR[client_ip] = 0
                        elif int(COUNT_CLEAR.get(client_ip, 0) > 4):
                            COUNT_CLEAR[client_ip] = 0
                        else:
                            COUNT_CLEAR[client_ip] = COUNT_CLEAR.get(client_ip, 0) + 1
                        _trim_mapping(COUNT_CLEAR, max_items=128)
    
                        content_type = content_type or response.headers.get('content-type', '').lower()
                        if ('mpegurl' in content_type or 'x-directory/normal' in content_type or '.m3u8' in url.lower()):
                            base_url = url.rsplit('/', 1)[0]
                            playlist_content = response.content.decode('utf-8', errors='ignore')
                            _store_hls_playlist(url, playlist_content, base_url, content_type=content_type or 'application/vnd.apple.mpegurl', kind=kind)
                            host = self.headers.get('Host', '127.0.0.1:%d' % PORT)
                            rewritten = _rewrite_m3u8_urls(playlist_content, base_url, host, header_blob=header_blob, kind=kind)
                            try:
                                response.close()
                            except Exception:
                                pass
                            self._send_text(rewritten, content_type='application/vnd.apple.mpegurl')
                            return
    
                        if '/hl' in url.lower() and '_' in url.lower() and '.ts' in url.lower():
                            try:
                                seg_ = re.findall(r'_(.*?)\.ts', url)[0]
                                seg_before = '_%s.ts' % seg_
                                seg_after = '_%s.ts' % str(int(seg_) + 1)
                                url = url.replace(seg_before, seg_after)
                            except Exception:
                                pass
    
                        media_type = 'video/mp4' if '.mp4' in url.lower() else 'video/mp2t' if '.ts' in url.lower() or '/hl' in url.lower() else response.headers.get('content-type', 'application/octet-stream')
                        is_hls_media = _is_hls_media_response(url, media_type or content_type)
                        response_headers = {k: v for k, v in response.headers.items() if k.lower() in ['content-type', 'accept-ranges', 'content-range']}
                        status = 206 if response.status_code == 206 else 200
    
                        if head_only:
                            self.send_response(status)
                            self.send_header('Content-Type', media_type)
                            for k, v in response_headers.items():
                                if k.lower() == 'content-type':
                                    continue
                                self.send_header(k, v)
                            self.send_header('Cache-Control', 'no-store')
                            self.send_header('Connection', 'close')
                            self.end_headers()
                            try:
                                response.close()
                            except Exception:
                                pass
                            try:
                                session.close()
                            except Exception:
                                pass
                            return

                        if is_vod_hls and is_hls_media:
                            # VOD HLS: entrega um segmento somente quando ele está
                            # completo. Microqueda => novo download do mesmo
                            # segmento; nunca reutiliza cauda parcial/dados duplicados.
                            segment_retries = int(_vod_recovery_settings(url).get('segment_max_retries', 3) or 0)
                            buffered = _buffer_vod_hls_segment(
                                session,
                                response,
                                url,
                                original_headers,
                                max(0, min(5, segment_retries))
                            )
                            response = None
                            if not buffered:
                                break
                            buffered_headers = buffered.get('headers') or {}
                            status = 206 if int(buffered.get('status') or 200) == 206 else 200
                            self.send_response(status)
                            self.send_header('Content-Type', media_type)
                            self.send_header('Content-Length', str(int(buffered.get('size') or 0)))
                            for k, v in buffered_headers.items():
                                lk = str(k).lower()
                                if lk in ('content-type', 'content-length', 'transfer-encoding', 'connection'):
                                    continue
                                if lk in ('accept-ranges', 'content-range') and status != 206:
                                    continue
                                if lk in ('accept-ranges', 'content-range'):
                                    self.send_header(k, v)
                            self.send_header('Cache-Control', 'no-store')
                            self.send_header('Connection', 'close')
                            self.end_headers()
                            spool = buffered.get('spool')
                            try:
                                while True:
                                    chunk = spool.read(_hls_chunk_size(url))
                                    if not chunk:
                                        break
                                    _write_client_chunk(self, chunk, flush=True)
                            except Exception:
                                return
                            finally:
                                try:
                                    spool.close()
                                except Exception:
                                    pass
                            return

                        if (not is_vod_hls) and is_hls_media:
                            # Live HLS segue em baixa latência, mas a continuidade
                            # agora é por Range validado; a cauda em cache não é
                            # reutilizada porque poderia duplicar pacotes TS.
                            _relay_live_hls_segment(
                                self,
                                session,
                                response,
                                url,
                                original_headers,
                                media_type,
                                response_headers,
                                head_only=False
                            )
                            response = None
                            return

                        stream_iter = _stream_response(response, client_ip, url, session)
                        first_chunk = b''
                        try:
                            first_chunk = next(stream_iter)
                        except StopIteration:
                            first_chunk = b''

                        if not first_chunk and is_hls_media:
                            try:
                                response.close()
                            except Exception:
                                pass
                            last_status = 'empty_segment'
                            attempts += 1
                            _throttled_log('hls-empty-segment:%s' % _origin_key(url), 15, logging.INFO, '[HLS Proxy] Segmento vazio/curto demais; retry conservador antes de expor falha ao player.')
                            time.sleep(min(0.35, 0.10 * attempts))
                            continue

                        self.send_response(status)
                        self.send_header('Content-Type', media_type)
                        for k, v in response_headers.items():
                            lk = k.lower()
                            if lk in ('content-type', 'accept-ranges', 'content-range') and status != 206 and lk != 'content-type':
                                continue
                            if lk == 'content-type':
                                continue
                            self.send_header(k, v)
                        self.send_header('Cache-Control', 'no-store')
                        self.send_header('Connection', 'close')
                        self.end_headers()

                        startup_flush_until = time.time() + 1.2
                        bytes_out = 0
                        if first_chunk:
                            try:
                                bytes_out += len(first_chunk)
                                _write_client_chunk(self, first_chunk, flush=True)
                            except Exception:
                                return
                        for chunk in stream_iter:
                            try:
                                bytes_out += len(chunk)
                                flush_now = (time.time() <= startup_flush_until) or (bytes_out <= 131072)
                                _write_client_chunk(self, chunk, flush=flush_now)
                            except Exception:
                                return
                        return
    
                    elif response.status_code == 416 and range_header and not tried_without_range:
                        tried_without_range = True
                        last_status = 416
                        try:
                            response.close()
                        except Exception:
                            pass
                        continue
                    else:
                        last_status = response.status_code
                        _mark_origin_fail(response.url or url, code=response.status_code)
                        AGENT_OF_CHAOS[cache_key] = binascii.b2a_hex(os.urandom(20))[:32].decode('ascii')
                        _trim_mapping(AGENT_OF_CHAOS, max_items=_RUNTIME_CACHE_LIMITS.get('agent_chaos', 128))
                        attempts += 1

                        # HTTP 404 de um segmento HLS Live é um caso próprio:
                        # poucas tentativas de publicação antes de devolver 404
                        # ao player, que então atualiza a playlist. Não consumir
                        # o orçamento longo de abertura da playlist nesse caso.
                        is_live_hls_segment = (not is_vod_hls) and _is_hls_media_like(url)
                        if response.status_code == 404 and is_live_hls_segment:
                            live_segment_404_failures += 1
                            segment_404 = _live_hls_segment_404_settings(url)
                            retry_budget = int(segment_404.get('retries', 0) or 0)
                            if live_segment_404_failures <= retry_budget:
                                delay = _live_hls_segment_404_delay(live_segment_404_failures)
                                _throttled_log(
                                    'live-hls-segment-404-retry:%s' % _origin_key(url),
                                    8,
                                    logging.INFO,
                                    '[Live HLS] Segmento 404; aguardando publicação (%d/%d), perfil=%s.' % (
                                        live_segment_404_failures,
                                        retry_budget,
                                        {'fast': 'Rápido', 'balance': 'Balanceado', 'stable': 'Estável'}.get(_get_mode(), 'Estável')
                                    )
                                )
                                time.sleep(delay)
                                continue
                            _throttled_log(
                                'live-hls-segment-404-expired:%s' % _origin_key(url),
                                8,
                                logging.INFO,
                                '[Live HLS] Segmento 404 persistente após %d tentativa(s); liberando atualização da playlist.' % (
                                    int(segment_404.get('total_attempts', live_segment_404_failures) or live_segment_404_failures),
                                )
                            )
                            break

                        # O contador 404 é consecutivo: uma resposta diferente
                        # volta ao tratamento geral e não herda a contagem anterior.
                        live_segment_404_failures = 0
                        if response.status_code == 406:
                            _burst_log('hls-406:%s' % _origin_key(url), 60, logging.INFO, '[HLS Proxy] Origem respondeu 406; fallback discreto acionado (%sx).')
                            time.sleep(min(0.50, 0.15 * attempts))
                        else:
                            level = logging.INFO if _source_profile(url) == 'xtream_bad' else logging.WARNING
                            _throttled_log('hls-status:%s:%s' % (_origin_key(url), response.status_code), 20, level, '[HLS Proxy] Resposta HTTP %s da origem.' % response.status_code)
                            time.sleep(min(0.65, 0.18 * attempts))
                        if _is_hls_media_like(url):
                            # Segmento live: tenta recuperar algumas vezes antes de apelar para cauda em cache.
                            if attempts < max_retries:
                                continue
                            if not _stream_cache_has(client_ip, url):
                                break
                            self.send_response(status)
                            self.send_header('Content-Type', media_type)
                            for k, v in response_headers.items():
                                if k.lower() == 'content-type':
                                    continue
                                self.send_header(k, v)
                            self.send_header('Connection', 'close')
                            self.end_headers()
                            if not head_only:
                                for chunk in (_stream_cache(client_ip, url) or []):
                                    try:
                                        _write_client_chunk(self, chunk, flush=True)
                                    except Exception:
                                        return
                            return
                except RequestException as e:
                    last_status = e.__class__.__name__
                    _mark_origin_fail(url, exc_name=e.__class__.__name__)
                    _mark_host_fail(_host_from_url(url), err=e.__class__.__name__, soft=(e.__class__.__name__ in ('ReadTimeout','ConnectionError','ChunkedEncodingError')))
                    AGENT_OF_CHAOS[cache_key] = binascii.b2a_hex(os.urandom(20))[:32].decode('ascii')
                    _trim_mapping(AGENT_OF_CHAOS, max_items=_RUNTIME_CACHE_LIMITS.get('agent_chaos', 128))
                    attempts += 1
                    lvl = logging.INFO if _source_profile(url) == 'xtream_bad' else logging.WARNING
                    _throttled_log('hls-exc:%s:%s' % (_origin_key(url), e.__class__.__name__), 15, lvl, '[HLS Proxy] Origem instável; retry inteligente em curso (%s).' % _safe_text(e.__class__.__name__))
                    time.sleep(_smart_retry_delay('timeout' if e.__class__.__name__ in ('ReadTimeout', 'ChunkedEncodingError') else 'connection', attempts))
                    if _is_hls_media_like(url):
                        # Segmento live: tenta recuperar algumas vezes antes de apelar para cauda em cache.
                        if attempts < max_retries:
                            continue
                        if not _stream_cache_has(client_ip, url):
                            break
                        self.send_response(status)
                        self.send_header('Content-Type', media_type)
                        for k, v in response_headers.items():
                            if k.lower() == 'content-type':
                                continue
                            self.send_header(k, v)
                        self.send_header('Connection', 'close')
                        self.end_headers()
                        if not head_only:
                            for chunk in (_stream_cache(client_ip, url) or []):
                                try:
                                    _write_client_chunk(self, chunk, flush=True)
                                except Exception:
                                    return
                        return
                finally:
                    if response is not None and response.raw is not None and (head_only or attempts > 0):
                        try:
                            response.close()
                        except Exception:
                            pass

            if _is_hls_playlist_url(url) and self._send_cached_hls_playlist(url, header_blob=header_blob, kind=kind):
                return
            if _is_hls_playlist_url(url) or _is_hls_media_like(url):
                # HLS lida melhor com sinalização HTTP simples do que com payload JSON.
                # 404 é fallback conservador para playlist/segmento temporariamente indisponível.
                self._send_text('Not Found', status=404, content_type='text/plain; charset=utf-8')
            else:
                self._send_json('{"detail": "Falha ao conectar após múltiplas tentativas"}', status=502)
        finally:
            try:
                session.close()
            except Exception:
                pass

    def _handle_vod_root_proxy(self, parsed, head_only=False):
        """Proxy de arquivo VOD com retomada transparente por Range.

        O player recebe uma única resposta local. Quando a origem cai no meio do
        corpo, o proxy reabre apenas os bytes faltantes e só os concatena se o
        servidor confirmar exatamente o offset solicitado em ``Content-Range``.
        Uma origem que ignora Range nunca recebe bytes duplicados no player.
        """
        _dns_override()
        params = urllib.parse.parse_qs(parsed.query)
        url, header_blob, _target_kind = _target_from_query_params(params)
        if not url:
            self._send_text('URL VOD ausente', status=400)
            return
        if not _is_valid_upstream_url(url):
            self._send_text('URL VOD inválida', status=400)
            return

        def _safe_header_value(value):
            return str(value or '').replace('\r', '').replace('\n', '').strip()

        def _origin_from_url(target):
            try:
                parsed_target = urllib.parse.urlparse(target or '')
                if parsed_target.scheme and parsed_target.netloc:
                    return '%s://%s/' % (parsed_target.scheme, parsed_target.netloc)
            except Exception:
                pass
            return ''

        def _guess_content_type(target, upstream_value=''):
            raw = (upstream_value or '').split(';', 1)[0].strip().lower()
            if raw and raw not in ('application/json', 'text/html', 'text/plain', 'application/xml', 'text/xml'):
                return upstream_value
            try:
                path = (urllib.parse.urlparse(target).path or '').lower()
            except Exception:
                path = (target or '').lower()
            if path.endswith('.mkv'):
                return 'video/x-matroska'
            if path.endswith('.avi'):
                return 'video/x-msvideo'
            if path.endswith('.mov'):
                return 'video/quicktime'
            if path.endswith('.webm'):
                return 'video/webm'
            if path.endswith('.ts'):
                return 'video/mp2t'
            if path.endswith('.m4v'):
                return 'video/x-m4v'
            return 'video/mp4'

        decoded_headers = _decode_header_blob(header_blob)
        requested_range = _safe_header_value(self.headers.get('Range') or self.headers.get('range') or '')
        requested_start, requested_end = _parse_requested_byte_range(requested_range)
        recovery = _vod_recovery_settings(url)

        def _build_vod_headers(range_value=None, profile='direct', target_url=None):
            # Permite apenas headers com valor semântico para a origem. Host,
            # Connection, Length e Range continuam sob controle do proxy.
            allowed = ('User-Agent', 'Referer', 'Origin', 'Cookie', 'Authorization', 'Accept', 'Accept-Language')
            extras = {}
            for key, value in (decoded_headers or {}).items():
                canonical = str(key or '').strip()
                if canonical.lower() in {name.lower() for name in allowed}:
                    clean_value = _safe_header_value(value)
                    if clean_value:
                        extras[canonical] = clean_value

            def _get_extra(name):
                for key, value in extras.items():
                    if key.lower() == name.lower():
                        return value
                return ''

            headers = {
                'User-Agent': _get_extra('User-Agent') or DEFAULT_USER_AGENT,
                'Accept': _get_extra('Accept') or 'video/mp4,video/*;q=0.9,*/*;q=0.8',
                'Accept-Language': _get_extra('Accept-Language') or 'pt-BR,pt;q=0.9',
                'Accept-Encoding': 'identity',
                'Connection': 'close',
            }
            for name in ('Cookie', 'Authorization', 'Referer', 'Origin'):
                value = _get_extra(name)
                if value:
                    headers[name] = value

            origin = _origin_from_url(target_url or url)
            if profile == 'origin' and origin:
                headers.setdefault('Referer', origin)
                headers.setdefault('Origin', origin.rstrip('/'))
            elif profile == 'minimal':
                headers.pop('Referer', None)
                headers.pop('Origin', None)
                headers.pop('Accept-Language', None)
                headers['Accept'] = '*/*'
                headers['User-Agent'] = DEFAULT_USER_AGENT

            if range_value:
                headers['Range'] = range_value
            return headers

        def _status(response):
            try:
                return int(response.getcode() or 0)
            except Exception:
                return int(getattr(response, 'code', 0) or 0)

        def _open_one(method, target_url, range_value, profile):
            req = UrlRequest(target_url, headers=_build_vod_headers(range_value=range_value, profile=profile, target_url=target_url))
            try:
                req.get_method = lambda m=method: m
            except Exception:
                pass
            try:
                return urlopen_url(req, timeout=recovery.get('timeout', 15.0))
            except UrlHTTPError as exc:
                return exc
            except Exception:
                return None

        def _open_upstream(method, target_url, range_value=None):
            # A primeira abertura mantém os perfis conservadores já aprovados.
            # Na retomada, eles ajudam fontes que oscilam por Referer/Origin sem
            # jamais eliminar o Range calculado pelo proxy.
            profiles = ('direct', 'origin', 'minimal')
            last = None
            for profile in profiles:
                response = _open_one(method, target_url, range_value, profile)
                if response is None:
                    continue
                code = _status(response)
                if 200 <= code < 300:
                    return response
                last = response
                if code not in (403, 405, 406, 408, 409, 429, 500, 502, 503, 504, 520, 521, 522, 523, 524):
                    break
                try:
                    response.close()
                except Exception:
                    pass
                last = None
                time.sleep(0.12)
            return last

        # HEAD é consultado pelo Kodi para MIME. Muitos Xtream respondem 405;
        # então obtém headers com GET 0-0 sem alterar o contrato externo.
        upstream = _open_upstream('HEAD' if head_only else 'GET', url, requested_range)
        if head_only and (upstream is None or _status(upstream) >= 400):
            try:
                if upstream is not None:
                    upstream.close()
            except Exception:
                pass
            upstream = _open_upstream('GET', url, requested_range or 'bytes=0-0')

        if upstream is None:
            _throttled_log('vod-connect:%s' % _origin_key(url), 15, logging.WARNING, '[VOD Proxy] Não foi possível abrir a origem VOD.')
            self._send_text('Falha ao conectar VOD', status=502)
            return

        client_headers_sent = False
        try:
            status = _status(upstream)
            if status < 200 or status >= 300:
                _mark_origin_fail(url, code=status)
                _mark_host_fail(_host_from_url(url), err=status)
                self._send_text('Origem VOD respondeu HTTP %s' % status, status=502)
                return

            try:
                final_url = upstream.geturl() or url
            except Exception:
                final_url = url
            content_type = _guess_content_type(final_url, _response_header(upstream, 'Content-Type', ''))
            content_length = _positive_int(_response_header(upstream, 'Content-Length', ''))
            content_range = _response_header(upstream, 'Content-Range', '')
            range_start, range_end, range_total = _parse_content_range(content_range)
            accept_ranges = _response_header(upstream, 'Accept-Ranges', 'bytes') or 'bytes'

            # Content-Length mede o corpo que o Kodi receberá nesta requisição.
            # Em 206 ele já representa somente a faixa solicitada.
            expected_body_bytes = content_length
            if expected_body_bytes is None and range_start is not None and range_end is not None:
                expected_body_bytes = (range_end - range_start) + 1
            stream_start = range_start if range_start is not None else (requested_start if requested_start is not None else 0)
            stream_end = None
            if expected_body_bytes is not None:
                stream_end = stream_start + expected_body_bytes - 1
            if requested_end is not None and stream_end is not None:
                stream_end = min(stream_end, requested_end)
                expected_body_bytes = max(0, stream_end - stream_start + 1)

            _mark_origin_ok(final_url)
            _mark_host_ok(_host_from_url(final_url), final_url=final_url, user_agent=_build_vod_headers(target_url=final_url).get('User-Agent'), content_type=content_type)
            _remember_source_profile(final_url, _classify_source(final_url))

            self.send_response(status)
            self.send_header('Content-Type', content_type)
            self.send_header('Accept-Ranges', accept_ranges)
            if content_length is not None:
                # Mantém exatamente o tamanho apresentado pelo upstream inicial.
                # A retomada é interna e não altera cabeçalhos já vistos pelo Kodi.
                self.send_header('Content-Length', str(expected_body_bytes if expected_body_bytes is not None else content_length))
            if content_range:
                self.send_header('Content-Range', str(content_range))
            self.send_header('Access-Control-Allow-Origin', '*')
            self.send_header('Connection', 'close')
            self.end_headers()
            client_headers_sent = True

            if head_only:
                return

            delivered = 0
            retries = 0
            active_url = final_url
            while True:
                stream_failure = None
                try:
                    chunk = upstream.read(65536)
                    if chunk:
                        if expected_body_bytes is not None:
                            remaining = expected_body_bytes - delivered
                            if remaining <= 0:
                                return
                            if len(chunk) > remaining:
                                chunk = chunk[:remaining]
                        try:
                            self.wfile.write(chunk)
                        except (BrokenPipeError, ConnectionResetError, OSError):
                            return
                        delivered += len(chunk)
                        if expected_body_bytes is not None and delivered >= expected_body_bytes:
                            return
                        continue
                    # EOF só é conclusivo quando o tamanho anunciado foi todo
                    # entregue. Sem tamanho, não há como distinguir EOF legítimo
                    # de queda: não arriscamos duplicar dados.
                    if expected_body_bytes is None or delivered >= expected_body_bytes:
                        return
                    stream_failure = IOError('upstream_eof_antes_do_content_length')
                except Exception as exc:
                    # Leitura do upstream pode lançar ConnectionResetError,
                    # timeout ou IncompleteRead. Isso é microqueda da origem;
                    # a conexão do Kodi só é testada no write acima.
                    stream_failure = exc

                if stream_failure is None:
                    return
                _mark_origin_fail(active_url, exc_name=stream_failure.__class__.__name__)
                _mark_host_fail(_host_from_url(active_url), err=stream_failure.__class__.__name__, soft=True)

                if expected_body_bytes is None:
                    _throttled_log(
                        'vod-no-length:%s' % _origin_key(active_url),
                        20,
                        logging.WARNING,
                        '[VOD Proxy] Origem caiu sem Content-Length; retomada segura por byte não é possível.'
                    )
                    return
                if retries >= int(recovery.get('max_retries', 0) or 0):
                    _throttled_log(
                        'vod-retry-exhausted:%s' % _origin_key(active_url),
                        15,
                        logging.WARNING,
                        '[VOD Proxy] Limite de retomadas atingido; conexão encerrada sem dados duplicados.'
                    )
                    return

                retries += 1
                resume_at = stream_start + delivered
                resume_to = stream_end
                resume_range = 'bytes=%d-' % resume_at if resume_to is None else 'bytes=%d-%d' % (resume_at, resume_to)
                _throttled_log(
                    'vod-range-resume:%s' % _origin_key(active_url),
                    15,
                    logging.INFO,
                    '[VOD Proxy] Microqueda detectada; retomando VOD por Range (%d/%d).' % (retries, int(recovery.get('max_retries', 0) or 0))
                )
                try:
                    upstream.close()
                except Exception:
                    pass
                time.sleep(_vod_retry_delay(retries))
                _apply_origin_cooldown(active_url)
                resumed = _open_upstream('GET', active_url, resume_range)
                if resumed is None:
                    continue
                resumed_status = _status(resumed)
                resumed_range = _response_header(resumed, 'Content-Range', '')
                resumed_start, resumed_end, _resumed_total = _parse_content_range(resumed_range)
                resumed_length = _positive_int(_response_header(resumed, 'Content-Length', ''))
                needed = expected_body_bytes - delivered
                range_ok = (
                    resumed_status == 206 and
                    resumed_start == resume_at and
                    resumed_end is not None and
                    resumed_end >= (resume_at + max(0, needed - 1)) and
                    (resumed_length is None or resumed_length >= needed)
                )
                if not range_ok:
                    try:
                        resumed.close()
                    except Exception:
                        pass
                    _mark_origin_fail(active_url, code=resumed_status)
                    # Se a origem respondeu 200/Range incompatível, anexar o
                    # corpo reiniciado corromperia a mídia. Para aqui, limpo.
                    _throttled_log(
                        'vod-range-rejected:%s' % _origin_key(active_url),
                        15,
                        logging.WARNING,
                        '[VOD Proxy] Origem não confirmou Range exato; evitando bytes duplicados.'
                    )
                    return
                try:
                    next_url = resumed.geturl() or active_url
                except Exception:
                    next_url = active_url
                upstream = resumed
                active_url = next_url
                _mark_origin_ok(active_url)
        except Exception as exc:
            _mark_origin_fail(url, exc_name=exc.__class__.__name__)
            _mark_host_fail(_host_from_url(url), err=exc.__class__.__name__, soft=True)
            if not client_headers_sent:
                self._send_text('Falha ao transmitir VOD', status=502)
            _throttled_log('vod-stream:%s:%s' % (_origin_key(url), exc.__class__.__name__), 15, logging.WARNING, '[VOD Proxy] Erro ao transmitir VOD (%s).' % _safe_text(exc.__class__.__name__))
        finally:
            try:
                upstream.close()
            except Exception:
                pass

    def _handle_tsdownloader(self, parsed, head_only=False):
        _dns_override()
        params = urllib.parse.parse_qs(parsed.query)
        url, header_blob, _target_kind = _target_from_query_params(params)
        if not url:
            self._send_json("{\"error\": \"Missing url parameter\"}", status=400)
            return
        if not _is_valid_upstream_url(url):
            self._send_json("{\"error\": \"Invalid upstream URL\"}", status=400)
            return

        original_headers = self._filtered_request_headers()
        original_headers.update(_decode_header_blob(header_blob))
        stop_ts = [False]
        session = requests.Session()
        resolved_url = ['']
        last_status = [None]

        def _client_closed():
            try:
                sock = getattr(self, 'connection', None)
                if sock is None:
                    return False
                readable, _, _ = select.select([sock], [], [], 0)
                if not readable:
                    return False
                try:
                    data = sock.recv(1, socket.MSG_PEEK)
                    return data == b''
                except (BlockingIOError, InterruptedError):
                    return False
                except (ConnectionResetError, BrokenPipeError, OSError):
                    return True
            except Exception:
                return False

        def _should_stop_stream():
            if stop_ts[0]:
                return True
            if _client_closed():
                stop_ts[0] = True
                return True
            return False

        def _sleep_ts(seconds):
            try:
                seconds = float(seconds or 0)
            except Exception:
                seconds = 0.0
            end_at = time.time() + max(0.0, seconds)
            while time.time() < end_at:
                if _should_stop_stream():
                    return False
                time.sleep(min(0.10, max(0.0, end_at - time.time())))
            return not _should_stop_stream()

        def _resolve_stream_url(force=False):
            if resolved_url[0] and not force:
                return resolved_url[0]
            host = _host_from_url(url)
            if host and not force:
                cached = _SESSION_CACHE['last_working_url'].get(host)
                if cached:
                    resolved_url[0] = cached
                    return resolved_url[0]
            probe_headers = _build_upstream_headers(original_headers, reconnects=0, response_code=last_status[0], prefer_stream=True, url=url)
            probe = session.get(url, headers=probe_headers, allow_redirects=True, stream=True, timeout=_timeout_tuple(url, prefer_stream=True))
            try:
                resolved_url[0] = probe.url or url
            finally:
                try:
                    probe.close()
                except Exception:
                    pass
            return resolved_url[0]

        def generate_ts():
            reconnects = 0
            last_good_at = 0
            last_chunk_at = 0
            keepalive_hits = 0
            startup_keepalive_until = time.time() + float(_live_ts_settings(url).get('startup_keepalive_window', 0.0) or 0.0)
            while not _should_stop_stream():
                response = None
                try:
                    if _should_stop_stream():
                        return
                    target_url = _resolve_stream_url(force=(reconnects > 0 and reconnects % int(_live_ts_settings(resolved_url[0] or url).get('reprobe_every', 3) or 3) == 0))
                    _apply_origin_cooldown(target_url)
                    request_headers = _build_upstream_headers(original_headers, reconnects=reconnects, response_code=last_status[0], prefer_stream=True, url=target_url)
                    host = _host_from_url(target_url)
                    if not _host_is_available(host):
                        _host_backoff(host)
                    request_headers['User-Agent'] = _pick_user_agent(host, reconnects, prefer_stream=True)
                    timeout_tuple = _timeout_tuple(target_url, prefer_stream=True)
                    if not last_good_at:
                        timeout_tuple = (timeout_tuple[0], min(float(timeout_tuple[1]), float(_live_ts_settings(target_url).get('startup_read_timeout', 5.0) or 5.0)))
                    response = session.get(target_url, headers=request_headers, stream=True, timeout=timeout_tuple, allow_redirects=True)
                    status_code = response.status_code
                    last_status[0] = status_code

                    ok_response, content_type = _validate_upstream_response(response, expect_stream=True, url=target_url)
                    if ok_response:
                        _mark_origin_ok(response.url or target_url)
                        _mark_host_ok(_host_from_url(response.url or target_url), final_url=(response.url or target_url), user_agent=request_headers.get('User-Agent'), content_type=content_type)
                        _remember_source_profile(response.url or target_url, _classify_source(response.url or target_url))
                        reconnects = 0
                        keepalive_hits = 0
                        for chunk in response.iter_content(chunk_size=_ts_stream_chunk_size()):
                            if _should_stop_stream():
                                _throttled_log('ts-stop-client', 60, logging.DEBUG, '[TS Downloader] Stream interrompido pelo cliente.')
                                return
                            if not chunk:
                                continue
                            now = time.time()
                            last_good_at = now
                            last_chunk_at = now
                            startup_keepalive_until = 0
                            yield chunk
                    elif status_code == 406:
                        reconnects += 1
                        _mark_origin_fail(response.url or target_url, code=406)
                        _mark_host_fail(_host_from_url(response.url or target_url), err='406', soft=True)
                        if reconnects <= 2:
                            _throttled_log('ts-406-soft:%s' % _origin_key(target_url), 60, logging.INFO, '[TS Downloader] 406 tratado como ruído transitório; retry silencioso.')
                        else:
                            _log_ts_status_once(status_code)
                        filler = _ts_keepalive_once(last_good_at, reconnects, startup_keepalive_until=startup_keepalive_until, url=resolved_url[0] or url)
                        if filler is not None:
                            keepalive_hits += 1
                            if keepalive_hits in (1, 5, 15):
                                msg = '[TS Downloader] Mantendo stream vivo durante microqueda da origem.' if last_good_at else '[TS Downloader] Segurando conexão local enquanto a origem acorda.'
                                _throttled_log('ts-keepalive:%s' % _origin_key(target_url), 20, logging.INFO, msg)
                            last_chunk_at = time.time()
                            yield filler
                            if not _sleep_ts(_ts_keepalive_sleep(reconnects, url=resolved_url[0] or url)):
                                return
                        if reconnects >= int(_live_ts_settings(target_url).get('reprobe_every', 3) or 3):
                            resolved_url[0] = ''
                        if not _sleep_ts(_smart_retry_delay('406', reconnects)):
                            return
                    elif status_code == 404:
                        reconnects += 1
                        _mark_origin_fail(response.url or target_url, code=404)
                        _mark_host_fail(_host_from_url(response.url or target_url), err='404')
                        _throttled_log('status:404:%s' % _origin_key(target_url), 20, logging.INFO if _source_profile(target_url) == 'xtream_bad' else logging.WARNING, '[TS Downloader] Origem respondeu 404; nova tentativa conservadora.')
                        filler = _ts_keepalive_once(last_good_at, reconnects, startup_keepalive_until=startup_keepalive_until, url=resolved_url[0] or url)
                        if filler is not None:
                            last_chunk_at = time.time()
                            yield filler
                            if not _sleep_ts(_ts_keepalive_sleep(reconnects, url=resolved_url[0] or url)):
                                return
                        resolved_url[0] = ''
                        if not _sleep_ts(_smart_retry_delay('connection', reconnects)):
                            return
                    else:
                        reconnects += 1
                        _mark_origin_fail(response.url or target_url, code=status_code)
                        _mark_host_fail(_host_from_url(response.url or target_url), err=status_code)
                        _throttled_log('status:%s:%s' % (_origin_key(target_url), status_code), 20, logging.INFO if _source_profile(target_url) == 'xtream_bad' else logging.WARNING, '[TS Downloader] Resposta HTTP %s da origem.' % status_code)
                        filler = _ts_keepalive_once(last_good_at, reconnects, startup_keepalive_until=startup_keepalive_until, url=resolved_url[0] or url)
                        if filler is not None:
                            last_chunk_at = time.time()
                            yield filler
                            if not _sleep_ts(_ts_keepalive_sleep(reconnects, url=resolved_url[0] or url)):
                                return
                        if reconnects >= int(_live_ts_settings(target_url).get('reprobe_every', 3) or 3):
                            resolved_url[0] = ''
                        if not _sleep_ts(_smart_retry_delay('connection', reconnects)):
                            return
                except (ReadTimeout, ChunkedEncodingError, ConnectionError) as e:
                    if _should_stop_stream():
                        return
                    reconnects += 1
                    last_status[0] = e.__class__.__name__
                    _mark_origin_fail(resolved_url[0] or url, exc_name=e.__class__.__name__)
                    _mark_host_fail(_host_from_url(resolved_url[0] or url), err=e.__class__.__name__, soft=True)
                    resolved_url[0] = ''
                    if reconnects <= 2:
                        _throttled_log('ts-reconnect-net-soft:%s' % _origin_key(url), 12, logging.INFO, '[TS Downloader] Origem instável; reconectando sem drama (%s).' % _safe_text(e.__class__.__name__))
                    else:
                        _throttled_log('ts-reconnect-net-hard:%s' % _origin_key(url), 20, logging.INFO if _source_profile(url) == 'xtream_bad' else logging.WARNING, '[TS Downloader] Instabilidade persistente da origem; retry progressivo (%s).' % _safe_text(e.__class__.__name__))
                    filler = _ts_keepalive_once(last_good_at, reconnects, startup_keepalive_until=startup_keepalive_until, url=resolved_url[0] or url)
                    if filler is not None:
                        last_chunk_at = time.time()
                        yield filler
                        if not _sleep_ts(_ts_keepalive_sleep(reconnects, url=resolved_url[0] or url)):
                            return
                    if not _sleep_ts(_smart_retry_delay('timeout', reconnects)):
                        return
                except GeneratorExit:
                    stop_ts[0] = True
                    return
                except Exception as e:
                    if _should_stop_stream():
                        return
                    reconnects += 1
                    last_status[0] = e.__class__.__name__
                    msg = str(e).lower()
                    if 'broken pipe' in msg or 'connection reset' in msg:
                        stop_ts[0] = True
                        _throttled_log('ts-client-disconnect-exc', 60, logging.DEBUG, '[TS Downloader] Cliente desconectou durante o stream.')
                        return
                    _mark_origin_fail(resolved_url[0] or url, exc_name=e.__class__.__name__)
                    _mark_host_fail(_host_from_url(resolved_url[0] or url), err=e.__class__.__name__, soft=True)
                    resolved_url[0] = ''
                    _throttled_log('ts-stream-error:%s' % _origin_key(url), 15, logging.INFO if _source_profile(url) == 'xtream_bad' else logging.WARNING, '[TS Downloader] Erro no stream; reconectando com mínimo de interferência: %s' % _safe_text(e))
                    filler = _ts_keepalive_once(last_good_at, reconnects, startup_keepalive_until=startup_keepalive_until, url=resolved_url[0] or url)
                    if filler is not None:
                        last_chunk_at = time.time()
                        yield filler
                        if not _sleep_ts(_ts_keepalive_sleep(reconnects, url=resolved_url[0] or url)):
                            return
                    if not _sleep_ts(_smart_retry_delay('connection', reconnects)):
                        return
                finally:
                    if response is not None:
                        try:
                            response.close()
                        except Exception:
                            pass
                    # se a origem der uma respirada longa demais, para de injetar filler e deixa o retry seguir limpo
                    if last_chunk_at and (time.time() - last_chunk_at) > max(2.0, float(_live_ts_settings(resolved_url[0] or url).get('keepalive_window', 8.0) or 8.0) + 1.0):
                        last_good_at = 0
                        keepalive_hits = 0
            _throttled_log('ts-stream-closed', 60, logging.DEBUG, '[TS Downloader] Stream encerrado pelo cliente.')

        self.send_response(200)
        self.send_header('Content-Type', 'video/mp2t')
        self.send_header('Connection', 'close')
        self.end_headers()

        stream_started_at = time.time()
        first_client_write_at = [0.0]
        bytes_sent_to_client = [0]
        disconnect_note = {'kind': None}

        def _is_startup_handoff():
            if head_only:
                return False
            window = float(_live_ts_settings(url).get('startup_handoff_window', 5.0) or 5.0)
            byte_cap = int(_live_ts_settings(url).get('startup_handoff_bytes', 393216) or 393216)
            if (time.time() - stream_started_at) > window:
                return False
            return bytes_sent_to_client[0] <= byte_cap

        def _note_client_disconnect(where='close'):
            if disconnect_note['kind'] is not None:
                return
            if _is_startup_handoff():
                disconnect_note['kind'] = 'startup-handoff'
                _throttled_log('ts-startup-handoff', 30, logging.INFO, '[TS Downloader] Handoff local de startup detectado; o Kodi trocou a conexão inicial pela definitiva (normal).')
            else:
                disconnect_note['kind'] = 'client-close'
                level = logging.DEBUG if where == 'close' else logging.INFO
                _throttled_log('ts-client-close', 60, level, '[TS Downloader] Cliente local encerrou a conexão do TS.')

        if head_only:
            stop_ts[0] = True
            try:
                session.close()
            except Exception:
                pass
            return

        try:
            startup_flush_until = time.time() + 2.0
            for chunk in generate_ts():
                try:
                    if not first_client_write_at[0]:
                        first_client_write_at[0] = time.time()
                    bytes_sent_to_client[0] += len(chunk)
                    flush_now = (time.time() <= startup_flush_until) or (bytes_sent_to_client[0] <= 262144)
                    _write_client_chunk(self, chunk, flush=flush_now)
                except Exception:
                    stop_ts[0] = True
                    _note_client_disconnect(where='write')
                    return
        finally:
            stop_ts[0] = True
            try:
                session.close()
            except Exception:
                pass
            _note_client_disconnect(where='close')


def is_proxy_running(port=None):
    """Retorna True somente quando a porta pertence ao proxy do OnePlayVIP."""
    if port is not None:
        return _probe_proxy_identity(port)
    return bool(_discover_running_proxy())


def _server_run(server):
    try:
        server.serve_forever(poll_interval=0.2)
    except Exception as exc:
        _log('Falha no loop do proxy: %s' % exc, level=xbmc.LOGERROR)


def start_proxy():
    """Inicia um único proxy próprio ou encontra o proxy do service.

    O health marker é obrigatório. Quando não houver listener válido, o bind
    usa porta 0: o sistema operacional escolhe uma porta dinâmica livre, alta
    e imprevisível para esta sessão, sem encostar nas portas fixas de addons.
    """
    global _SERVER_THREAD, _SERVER_STARTED, _SERVER, PORT
    with _SERVER_LOCK:
        running_port = _discover_running_proxy()
        if running_port:
            return True

        # Esta invocação ainda pode possuir um servidor que acabou de iniciar.
        if _SERVER is not None and _SERVER_THREAD is not None and _SERVER_THREAD.is_alive():
            if _probe_proxy_identity(PORT):
                _SERVER_STARTED = True
                return True

        # Remove referências locais quebradas antes de tentar novo bind.
        try:
            if _SERVER is not None:
                _SERVER.server_close()
        except Exception:
            pass
        _SERVER = None
        _SERVER_THREAD = None
        _SERVER_STARTED = False

        server = None
        selected_port = None
        # Bind em 0 é atômico: não há janela de corrida entre procurar uma
        # porta livre e abrir o listener. O SO devolve a porta efetivamente
        # reservada em server_address; ela muda somente em novo ciclo do Kodi.
        for _attempt in range(4):
            try:
                candidate_server = _ThreadedHTTPServer(('127.0.0.1', AUTO_BIND_PORT), _ProxyHandler)
                candidate_port = int(candidate_server.server_address[1] or 0)
                if not (MIN_DYNAMIC_PORT <= candidate_port <= MAX_DYNAMIC_PORT):
                    try:
                        candidate_server.server_close()
                    except Exception:
                        pass
                    continue
                server = candidate_server
                selected_port = candidate_port
                break
            except OSError as exc:
                _log('Falha ao solicitar porta dinâmica local: %s' % _safe_text(exc), level=xbmc.LOGDEBUG)
                continue
            except Exception as exc:
                _log('Falha ao preparar porta dinâmica local: %s' % _safe_text(exc), level=xbmc.LOGDEBUG)
                continue

        if server is None or selected_port is None:
            _log('Não foi possível reservar uma porta dinâmica exclusiva para o proxy OnePlayVIP.', level=xbmc.LOGERROR)
            return False

        PORT = int(selected_port)
        _SERVER = server
        _SERVER_STARTED = True
        _SERVER_THREAD = threading.Thread(target=_server_run, args=(server,), name='OnePlayVIPProxyServer', daemon=True)
        _SERVER_THREAD.start()
        _write_runtime_state(PORT)
        _cleanup_proxy_targets(force=True)

    for _ in range(20):
        if _probe_proxy_identity(PORT):
            _log('Proxy nativo OnePlayVIP ativo na porta %d' % PORT)
            return True
        time.sleep(0.10)

    _log('Proxy nativo OnePlayVIP não respondeu na porta %d' % PORT, level=xbmc.LOGERROR)
    stop_proxy()
    return False


def stop_proxy():
    global _SERVER, _SERVER_THREAD, _SERVER_STARTED, PORT
    with _SERVER_LOCK:
        server = _SERVER
        thread = _SERVER_THREAD
        own_port = PORT
        owns_listener = server is not None
        _SERVER = None
        _SERVER_THREAD = None
        _SERVER_STARTED = False
        # Não mantenha uma porta antiga em memória depois de encerrar o
        # listener próprio. Isso impede URL localhost apontando para serviço
        # já inexistente durante trocas rápidas de configuração/reinício.
        if owns_listener:
            PORT = None
        try:
            if server is not None:
                server.shutdown()
                server.server_close()
        except Exception:
            pass
    try:
        if thread and thread.is_alive():
            thread.join(timeout=1.0)
    except Exception:
        pass
    # Uma invocação curta de main.py pode apenas ter descoberto o proxy do
    # service; ela nunca deve apagar o registro do processo que realmente o
    # mantém vivo.
    if owns_listener:
        _clear_runtime_state(port=own_port)
    _clear_transient_runtime_state()
    return True

def kodiproxy():
    return start_proxy()
