# -*- 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
from collections import deque

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

from urllib.parse import urljoin
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

PORT = 8088
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()

_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': {},
}
_SETTINGS_CACHE = {'ts': 0, 'addon': None, 'mode': 'fast', 'xtream_hardening': False, '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 = {
    'fast': {'hls_max_retries': 5, 'ts_max_retries': 999999, 'timeout_connect': 4, 'timeout_read': 10, 'cooldown_base': 0.10},
    'balance': {'hls_max_retries': 7, 'ts_max_retries': 999999, 'timeout_connect': 5, 'timeout_read': 15, 'cooldown_base': 0.15},
    'stable': {'hls_max_retries': 9, 'ts_max_retries': 999999, '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('hls_chunk_size', 32768 if _mode_name == 'fast' else 49152 if _mode_name == 'balance' else 65536)

# 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:
        return str(value)
    except Exception:
        try:
            return repr(value)
        except Exception:
            return 'erro_desconhecido'


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):
    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):
    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('ts', 0)) < 30:
        return _SETTINGS_CACHE.get('addon')
    try:
        _SETTINGS_CACHE['addon'] = xbmcaddon.Addon('plugin.video.oneplayvip')  # type: ignore[name-defined]
        _SETTINGS_CACHE['ts'] = now
    except Exception:
        _SETTINGS_CACHE['addon'] = None
    return _SETTINGS_CACHE.get('addon')


def _settings_snapshot(force=False):
    now = time.time()
    if not force and (now - _SETTINGS_CACHE.get('ts', 0)) < 10:
        return _SETTINGS_CACHE
    addon = _addon_handle()
    mode = 'fast'
    xtream_hardening = False
    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({'ts': now, 'mode': mode, 'xtream_hardening': xtream_hardening, 'host_memory': host_memory, 'debug': debug})
    return _SETTINGS_CACHE


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


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


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():
    try:
        from doh import DNSOverrideDoH
        DNSOverrideDoH()
        return
    except Exception:
        pass
    try:
        from resources.lib.dns import customdns
        customdns()
    except Exception:
        pass


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=''):
    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 profile == 'xtream_bad' and _settings_snapshot().get('xtream_hardening'):
        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')

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

    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 profile == 'xtream_bad' and _settings_snapshot().get('xtream_hardening'):
        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_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):
    try:
        parsed = urllib.parse.urlparse(url or '')
        return urllib.parse.urlunparse((parsed.scheme, parsed.netloc, parsed.path, '', parsed.query, ''))
    except Exception:
        return 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'):
    try:
        if not url or not raw_playlist or '#EXTM3U' not in raw_playlist[:256].upper():
            return
        key = _hls_playlist_cache_key(url)
        _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):
    try:
        item = _HLS_PLAYLIST_CACHE.get(_hls_playlist_cache_key(url))
        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 _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=''):
    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
        proxied_url = 'http://%s/hlsretry?url=%s' % (host, urllib.parse.quote(absolute_url, safe=''))
        if header_blob:
            proxied_url += '&headers=' + urllib.parse.quote(header_blob, safe='')
        return proxied_url
    except Exception:
        return uri


def _rewrite_m3u8_urls(playlist_content, base_url, host, header_blob=''):
    """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) + 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) + 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()

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


def _ts_null_chunk(packet_count=None):
    try:
        packets = int(packet_count or _policy().get('ts_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):
    # 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(_policy().get('ts_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):
    if not _should_ts_keepalive(last_good_at, startup_keepalive_until=startup_keepalive_until):
        return None
    base_packets = int(_policy().get('ts_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)


def _ts_keepalive_sleep(reconnects=0):
    base = float(_policy().get('ts_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


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=''):
        item = _get_hls_playlist_cache(url)
        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)
            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 == '/':
            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)
        header_blob = params.get('headers', [''])[0]
        url = params.get('url', [None])[0]
        if url:
            url = urllib.parse.unquote(url)
        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 = _policy().get('hls_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

        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')
                            host = self.headers.get('Host', '127.0.0.1:%d' % PORT)
                            rewritten = _rewrite_m3u8_urls(playlist_content, base_url, host, header_blob=header_blob)
                            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')
                        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

                        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_like(url):
                            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
                        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):
                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_tsdownloader(self, parsed, head_only=False):
        _dns_override()
        params = urllib.parse.parse_qs(parsed.query)
        header_blob = params.get('headers', [''])[0]
        url = params.get('url', [None])[0]
        if url:
            url = urllib.parse.unquote(url)
        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(_policy().get('ts_startup_keepalive_window', 5.0) or 5.0)
            while not _should_stop_stream():
                response = None
                try:
                    if _should_stop_stream():
                        return
                    target_url = _resolve_stream_url(force=(reconnects > 0 and reconnects % 4 == 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(_policy().get('ts_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)
                        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)):
                                return
                        if reconnects >= 4:
                            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)
                        if filler is not None:
                            last_chunk_at = time.time()
                            yield filler
                            if not _sleep_ts(_ts_keepalive_sleep(reconnects)):
                                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)
                        if filler is not None:
                            last_chunk_at = time.time()
                            yield filler
                            if not _sleep_ts(_ts_keepalive_sleep(reconnects)):
                                return
                        if reconnects >= 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)
                    if filler is not None:
                        last_chunk_at = time.time()
                        yield filler
                        if not _sleep_ts(_ts_keepalive_sleep(reconnects)):
                            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)
                    if filler is not None:
                        last_chunk_at = time.time()
                        yield filler
                        if not _sleep_ts(_ts_keepalive_sleep(reconnects)):
                            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(_policy().get('ts_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(_policy().get('ts_startup_handoff_window', 5.0) or 5.0)
            byte_cap = int(_policy().get('ts_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():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(0.35)
    try:
        return s.connect_ex(('127.0.0.1', PORT)) == 0
    finally:
        try:
            s.close()
        except Exception:
            pass


def _server_run():
    global _SERVER
    try:
        _SERVER = _ThreadedHTTPServer(('127.0.0.1', PORT), _ProxyHandler)
        _SERVER.serve_forever(poll_interval=0.2)
    except Exception as e:
        _log('Falha ao iniciar proxy: %s' % e, level=xbmc.LOGERROR)


def start_proxy():
    global _SERVER_THREAD, _SERVER_STARTED
    with _SERVER_LOCK:
        if is_proxy_running():
            _SERVER_STARTED = True
            return True
        if _SERVER_THREAD and _SERVER_THREAD.is_alive():
            return True
        _SERVER_THREAD = threading.Thread(target=_server_run, name='OnePlayProxyServer', daemon=True)
        _SERVER_THREAD.start()
    for _ in range(30):
        if is_proxy_running():
            _SERVER_STARTED = True
            _log('Proxy nativo OnePlayVIP ativo na porta %d' % PORT)
            return True
        time.sleep(0.2)
    _log('Proxy nativo OnePlayVIP não respondeu na porta %d' % PORT, level=xbmc.LOGERROR)
    return False


def stop_proxy():
    global _SERVER, _SERVER_THREAD, _SERVER_STARTED
    with _SERVER_LOCK:
        thread = _SERVER_THREAD
        try:
            if _SERVER is not None:
                _SERVER.shutdown()
                _SERVER.server_close()
        except Exception:
            pass
        finally:
            _SERVER = None
            _SERVER_THREAD = None
            _SERVER_STARTED = False
    try:
        if thread and thread.is_alive():
            thread.join(timeout=1.0)
    except Exception:
        pass
    _clear_transient_runtime_state()
    return True


def kodiproxy():
    return start_proxy()
