"""status_module.py — pagina de Status do Troco Play (/status, /api/status).

Vive separado do app.py por tamanho; os nomes do aplicativo (models, config,
helpers) sao REENCAMINADOS pelo app.py logo apos o import, no bloco marcado
"─── injecao de nomes ───". Assim o modulo continua testavel isolado e nao
precisa de um app factory.
"""
import json
import os
import sys
import time as _time
from datetime import datetime, timedelta, timezone

from flask import jsonify, render_template, request
from flask_login import current_user, login_required
from sqlalchemy import func, text as sa_text

# Padroes: substituidos pelos valores reais pelo app.py no momento do import.
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
BRASILIA_TZ = timezone(timedelta(hours=-3))
APP = None
APP_START_TS = _time.time()
db = None
Canal = ChannelHealth = EPGProgram = ChannelEPGMapping = M3USourceLog = None
WatchLog = ActiveSession = User = Payment = Notification = SupportTicket = None
load_settings = detect_platform = apply_adult_block = user_blocks_adult = None
_naive_as_utc = None



# ════════════════════════════════════════════════════════════════════════
# STATUS DO SISTEMA — /status, /api/status, /cron/status-sample
# ════════════════════════════════════════════════════════════════════════
# O historico mora em ARQUIVO, nao no banco: se o banco cair, a pagina de
# status ainda consegue mostrar as amostras anteriores (e e exatamente ai
# que ela passa a ser util).
STATUS_HISTORY_PATH = os.path.join(BASE_DIR, 'instance', 'status_history.jsonl')
CRON_STATE_PATH = os.path.join(BASE_DIR, 'instance', 'cron_state.json')
STATUS_RETENTION_DAYS = 14
STATUS_SAMPLE_SPAN = int(os.environ.get('STATUS_SAMPLE_SPAN', '55'))
STATUS_REPORT_TTL = 20        # cache do relatorio (muitas abas abertas)
STATUS_MAX_LINES = 40000
_status_report_cache = {}
_status_history_cache = {'ts': 0.0, 'data': None}

# (chave, rotulo, intervalo esperado em minutos ou nome da chave de settings)
CRON_JOBS = [
    ('update-m3u', 'Importacao do catalogo (M3U)', 'update_interval_minutes'),
    ('update-epg', 'Atualizacao do guia (EPG)', 'epg_update_interval_minutes'),
    ('health', 'Verificacao de saude dos canais', 30),
    ('check-payments', 'Conferencia de pagamentos', 10),
    ('xtream-enrich', 'Enriquecimento Xtream', 360),
    ('cleanup', 'Limpeza de sessoes e tokens', 360),
    ('subscription-notifications', 'Lembretes de assinatura', 1440),
    ('backup', 'Backup do banco', 1440),
    ('status-sample', 'Amostragem de status', 5),
]


def _status_now_br():
    return datetime.now(BRASILIA_TZ)


def _status_iso(dt):
    try:
        if dt is None:
            return None
        return _naive_as_utc(dt).astimezone(BRASILIA_TZ).isoformat(timespec='seconds')
    except Exception:
        return None


def _status_age_min(dt, now_ts):
    """Minutos desde um datetime do banco (naive = UTC) ate agora."""
    try:
        if dt is None:
            return None
        return max(0.0, (now_ts - _naive_as_utc(dt).timestamp()) / 60.0)
    except Exception:
        return None


def _status_write_json(path, payload):
    tmp = path + '.tmp'
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(tmp, 'w') as f:
        json.dump(payload, f, ensure_ascii=False)
    os.replace(tmp, path)


def _status_read_json(path, default):
    try:
        with open(path, 'r') as f:
            return json.load(f)
    except Exception:
        return default


def _record_cron_run(name):
    """Marca a execucao de um job para a pagina de Status.

    Falhar aqui nao pode derrubar o cron, entao tudo e engolido.
    """
    try:
        state = _status_read_json(CRON_STATE_PATH, {})
        state[name] = datetime.now(timezone.utc).isoformat(timespec='seconds')
        _status_write_json(CRON_STATE_PATH, state)
    except Exception:
        pass


def _cron_state():
    return _status_read_json(CRON_STATE_PATH, {})


def _status_history_load():
    """Le as amostras guardadas (mais novas por ultimo), dentro da retencao."""
    global _status_history_cache
    now = _time.time()
    if (_status_history_cache['data'] is not None
            and now - _status_history_cache['ts'] < STATUS_SAMPLE_SPAN):
        return _status_history_cache['data']
    rows = []
    try:
        if os.path.exists(STATUS_HISTORY_PATH):
            with open(STATUS_HISTORY_PATH, 'r') as f:
                lines = f.readlines()[-STATUS_MAX_LINES:]
            cutoff = now - STATUS_RETENTION_DAYS * 86400
            for ln in lines:
                ln = ln.strip()
                if not ln:
                    continue
                try:
                    d = json.loads(ln)
                except Exception:
                    continue
                if d.get('ts', 0) >= cutoff:
                    rows.append(d)
    except Exception:
        rows = []
    _status_history_cache = {'ts': now, 'data': rows}
    return rows


def _status_history_line_count():
    """Quantas amostras ha no arquivo (para o admin julgar a amostragem)."""
    try:
        if not os.path.exists(STATUS_HISTORY_PATH):
            return 0
        with open(STATUS_HISTORY_PATH, 'r') as f:
            return sum(1 for ln in f if ln.strip())
    except Exception:
        return 0


def _status_append_sample(sample):
    global _status_history_cache
    try:
        os.makedirs(os.path.dirname(STATUS_HISTORY_PATH), exist_ok=True)
        rewrite = False
        if os.path.exists(STATUS_HISTORY_PATH):
            cutoff = _time.time() - STATUS_RETENTION_DAYS * 86400
            with open(STATUS_HISTORY_PATH, 'r') as f:
                head = f.readline()
            try:
                rewrite = bool(head) and json.loads(head).get('ts', cutoff) < cutoff
            except Exception:
                rewrite = False
        with open(STATUS_HISTORY_PATH, 'a') as f:
            f.write(json.dumps(sample, ensure_ascii=False, separators=(',', ':')) + '\n')
        if rewrite:
            keep = []
            cutoff = _time.time() - STATUS_RETENTION_DAYS * 86400
            with open(STATUS_HISTORY_PATH, 'r') as f:
                for ln in f:
                    try:
                        if json.loads(ln).get('ts', 0) >= cutoff:
                            keep.append(ln)
                    except Exception:
                        continue
            _status_write_json_path_text(STATUS_HISTORY_PATH, ''.join(keep))
        _status_history_cache = {'ts': 0.0, 'data': None}
    except Exception:
        pass


def _status_write_json_path_text(path, text):
    tmp = path + '.tmp'
    with open(tmp, 'w') as f:
        f.write(text)
    os.replace(tmp, path)


def _status_minutes_label(mins):
    if mins is None:
        return 'sem registro'
    mins = int(mins)
    if mins < 1:
        return 'agora'
    if mins < 60:
        return f'{mins} min atras'
    if mins < 1440:
        return f'{mins // 60}h {mins % 60}min atras'
    d = mins // 1440
    h = (mins % 1440) // 60
    return f'{d}d {h}h atras'


def _status_uptime_label(seconds):
    if not seconds or seconds < 0:
        return '-'
    seconds = int(seconds)
    d, rem = divmod(seconds, 86400)
    h, rem = divmod(rem, 3600)
    m, s = divmod(rem, 60)
    if d:
        return f'{d}d {h}h {m}min'
    if h:
        return f'{h}h {m}min'
    if m:
        return f'{m}min {s}s'
    return f'{s}s'


def _status_svc(key, label, status, detail='', latency_ms=None, age_min=None):
    return {'key': key, 'label': label, 'status': status, 'detail': detail,
            'latency_ms': latency_ms,
            'updated_at': None if age_min is None else _status_now_br().timestamp() - age_min * 60,
            'updated_label': _status_minutes_label(age_min)}


def _current_platform():
    try:
        return detect_platform(request.headers.get('User-Agent', ''))
    except Exception:
        return 'site'


def _status_cache_key():
    """O relatorio tem dados do usuario (o que ele enxerga, se e admin), entao
    a cache nao pode ser compartilhada entre perfis diferentes."""
    try:
        uid = getattr(current_user, 'id', None) or 'anon'
        adm = '1' if (getattr(current_user, 'is_authenticated', False)
                      and current_user.is_admin) else '0'
        ab = '1' if user_blocks_adult() else '0'
        return f'{uid}:{adm}:{ab}'
    except Exception:
        return 'anon:0:0'


def _status_report():
    """Monta o relatorio completo. Barato o bastante para rodar a cada minuto."""
    global _status_report_cache
    now_ts = _time.time()
    ckey = _status_cache_key()
    if len(_status_report_cache) > 50:
        _status_report_cache.clear()
    cached = _status_report_cache.get(ckey)
    if cached is not None and (now_ts - cached['ts']) < STATUS_REPORT_TTL:
        return cached['data']

    t_start = now_ts
    settings = load_settings()
    platform = _current_platform()
    services = []
    metrics = {}
    errors = []

    # ── banco de dados ──
    db_ms = None
    db_ok = False
    try:
        t0 = _time.time()
        db.session.execute(sa_text('SELECT 1'))
        db_ms = round((_time.time() - t0) * 1000, 1)
        db_ok = True
    except Exception as exc:
        errors.append({'service': 'database', 'message': f'Banco sem resposta: {str(exc)[:160]}'})
    services.append(_status_svc(
        'database', 'Banco de dados', 'ok' if db_ok else 'down',
        'conexao estabelecida' if db_ok else 'nao foi possivel consultar o banco',
        db_ms))

    # ── contagem de catalogos (depende do banco) ──
    counts = {'live': 0, 'movie': 0, 'series': 0, 'episodes': 0, 'epg': 0, 'mappings': 0}
    last_added = {}
    section_counts = {}
    online = offline = 0
    health_checked_min = None
    epg_latest_min = None
    epg_mapping_min = None
    hidden = 0
    if db_ok:
        try:
            from sqlalchemy import case
            for tipo in ('live', 'movie', 'series'):
                row = db.session.query(
                    func.count(Canal.id),
                    func.sum(case((Canal.is_hidden == False, 1), else_=0)),
                    func.max(Canal.data_added)
                ).filter(Canal.tipo == tipo).first()
                total = int((row[0] if row else 0) or 0)
                visiveis = int((row[1] if row else 0) or 0)
                section_counts[tipo] = {'total': total, 'visible': visiveis}
                counts[tipo] = visiveis
                if row and row[2]:
                    last_added[tipo] = _status_age_min(row[2], now_ts)
            counts['episodes'] = section_counts.get('series', {}).get('visible', 0)
            hidden = sum(v['total'] - v['visible'] for v in section_counts.values())
            counts['series'] = db.session.query(Canal.series_name).filter(
                Canal.tipo == 'series', Canal.is_hidden == False,
                Canal.series_name != '', Canal.series_name.isnot(None)
            ).distinct().count()
            counts['epg'] = EPGProgram.query.count()
            counts['mappings'] = ChannelEPGMapping.query.count()
            if counts['epg']:
                epg_latest_min = _status_age_min(db.session.query(
                    func.max(EPGProgram.start_time)).scalar(), now_ts)
            if counts['mappings']:
                epg_mapping_min = _status_age_min(db.session.query(
                    func.max(Canal.data_added)).filter(
                    Canal.id.in_(db.session.query(ChannelEPGMapping.canal_id))).scalar(), now_ts)
            online = db.session.query(func.count(ChannelHealth.id)).filter(
                ChannelHealth.is_online == True).scalar() or 0
            offline = db.session.query(func.count(ChannelHealth.id)).filter(
                ChannelHealth.is_online == False).scalar() or 0
            if (online + offline) > 0:
                health_checked_min = _status_age_min(db.session.query(
                    func.max(ChannelHealth.last_checked)).scalar(), now_ts)
        except Exception as exc:
            errors.append({'service': 'catalog', 'message': f'Falha ao ler catalogos: {str(exc)[:160]}'})
            db_ok = False

    metrics['canais'] = counts['live']
    metrics['mappings'] = counts['mappings']
    metrics['filmes'] = counts['movie']
    metrics['series'] = counts['series']
    metrics['episodios'] = counts['episodes']
    metrics['epg_programas'] = counts['epg']
    metrics['canais_ocultos'] = hidden
    metrics['canais_online'] = online
    metrics['canais_offline'] = offline

    # ── o que este usuario especificamente enxerga ──
    user_counts = {}
    if db_ok:
        try:
            user_counts['canais'] = apply_adult_block(
                Canal.query.filter_by(tipo='live', is_hidden=False)).count()
            user_counts['filmes'] = apply_adult_block(
                Canal.query.filter_by(tipo='movie', is_hidden=False)).count()
        except Exception:
            user_counts = {}

    # ── eventos recentes das fontes M3U / proxy / saude (historico real) ──
    recent_events = []
    proxy_fails_24h = 0
    sync_error_24h = 0
    last_sync_min = None
    last_sync_error = None
    if db_ok:
        try:
            since = datetime.now(timezone.utc) - timedelta(hours=24)
            rows = (M3USourceLog.query
                    .filter(M3USourceLog.created_at >= since)
                    .order_by(M3USourceLog.created_at.desc()).limit(80).all())
            metrics['eventos_24h'] = len(rows)
            for r in rows:
                age = _status_age_min(r.created_at, now_ts)
                act = (r.action or 'm3u')
                if act == 'sync' and r.status == 'ok' and last_sync_min is None:
                    last_sync_min = age
                if act == 'sync' and r.status == 'error':
                    sync_error_24h += 1
                    if last_sync_error is None or (age is not None and age < last_sync_error[0]):
                        last_sync_error = (age, (r.message or '')[:220])
                if act.startswith('proxy'):
                    if r.status in ('error', 'warn'):
                        proxy_fails_24h += 1
                    continue
                if r.status in ('error', 'warn'):
                    recent_events.append({
                        'at': _status_iso(r.created_at),
                        'ts': (_naive_as_utc(r.created_at).timestamp()
                               if r.created_at else 0),
                        'level': r.status,
                        'service': act,
                        'message': (r.message or '')[:220]
                    })
        except Exception:
            pass
    services.append(_status_svc(
        'proxy', 'Proxy de streaming',
        'degraded' if proxy_fails_24h >= 20 else 'ok',
        f'{proxy_fails_24h} falhas de proxy nas ultimas 24h' if proxy_fails_24h
        else 'sem falhas registradas'))

    # ── usuarios / visualizadores ──
    if db_ok:
        try:
            cutoff = datetime.now(timezone.utc) - timedelta(minutes=5)
            metrics['assistindo_agora'] = db.session.query(func.count(WatchLog.id)).filter(
                WatchLog.is_active == True, WatchLog.last_seen >= cutoff).scalar() or 0
            metrics['sessoes_ativas'] = db.session.query(func.count(ActiveSession.id)).filter(
                ActiveSession.started_at >= datetime.now(timezone.utc) - timedelta(hours=24)).scalar() or 0
            metrics['usuarios'] = User.query.count()
        except Exception:
            metrics['assistindo_agora'] = 0

    # ── servico web (o proprio request) ──
    site_ms = round((_time.time() - t_start) * 1000, 1)
    services.insert(0, _status_svc(
        'site', 'Servidor web', 'ok', 'Flask respondendo', site_ms))
    services.append(_status_svc(
        'response', 'Tempo de resposta', 'ok' if site_ms < 1500 else 'degraded',
        f'{site_ms} ms para montar o relatorio', site_ms))

    # ── contas / login ──
    auth_ms = None
    auth_ok = False
    try:
        t0 = _time.time()
        uid = getattr(current_user, 'id', None)
        if uid is not None:
            User.query.get(uid)
        else:
            User.query.first()
        auth_ms = round((_time.time() - t0) * 1000, 1)
        auth_ok = True
    except Exception as exc:
        errors.append({'service': 'auth', 'message': f'Contas indisponiveis: {str(exc)[:140]}'})
    services.insert(1, _status_svc(
        'auth', 'Contas e login', 'ok' if auth_ok else 'down',
        'sessao e autenticacao funcionando', auth_ms))

    # ── secoes / abas ──
    maint = status_maintenance_state(settings, platform)

    def section_state(section, count, empty_is_bad=True):
        if maint['enabled'] and (maint['scope'] == 'global' or
                                 section in maint['sections'] or
                                 any(section in str(p) for p in maint['pages'])):
            return 'maintenance'
        if not db_ok:
            return 'down'
        if empty_is_bad and not count:
            return 'degraded'
        return 'ok'

    can_watch = True
    try:
        can_watch = bool(current_user.can_watch())
    except Exception:
        can_watch = True

    def with_access(st):
        return st if st != 'ok' else ('ok' if can_watch else 'access')

    def section_row(key, label, path, tipo, section, detail):
        sc = section_counts.get(tipo, {})
        return {
            'key': key, 'label': label, 'path': path,
            'count': sc.get('visible', 0), 'total': sc.get('total', 0),
            'hidden': max(0, sc.get('total', 0) - sc.get('visible', 0)),
            'status': with_access(section_state(section, sc.get('visible', 0))),
            'detail': detail,
            'updated_label': _status_minutes_label(last_added.get(tipo)),
            'updated_min': last_added.get(tipo),
        }

    sections_out = [
        section_row('canais', 'Canais ao vivo', '/canais', 'live', 'canais',
                    f'{user_counts.get("canais", counts["live"])} canais liberados para voce'),
        section_row('filmes', 'Filmes', '/filmes', 'movie', 'filmes',
                    f'{user_counts.get("filmes", counts["movie"])} filmes liberados para voce'),
        section_row('series', 'Series', '/series', 'series', 'series',
                    f'{counts["series"]} series · {counts["episodes"]} episodios'),
        {'key': 'tv', 'label': 'Modo TV', 'path': '/tv/canais',
         'count': counts['live'], 'total': counts['live'], 'hidden': 0,
         'status': section_state('tv_canais', counts['live']),
         'detail': 'espelho da television (/, /canais, /filmes, /series)',
         'updated_label': '', 'updated_min': None},
        {'key': 'app', 'label': 'App Android', 'path': '/download',
         'count': 0, 'total': 0, 'hidden': 0,
         'status': section_state('app_mobile', 1, empty_is_bad=False),
         'detail': f"APK v{settings.get('app_version', '-')} · {settings.get('apk_size', '')}",
         'updated_label': '', 'updated_min': None},
    ]
    if not can_watch:
        sections_out[0]['user_note'] = ('Sua assinatura nao esta ativa no momento — '
                                        'renove em Planos para assistir.')

    catalog_detail = (f"{counts['live']} canais, {counts['movie']} filmes, "
                      f"{counts['series']} series")
    if sync_error_24h:
        cat_status = 'degraded'
        catalog_detail += f' · {sync_error_24h} erro(s) de sincronizacao em 24h'
    else:
        cat_status = 'ok' if (db_ok and counts['live']) else ('down' if not db_ok else 'degraded')
    services.append(_status_svc(
        'catalog', 'Catalogo de conteudo', cat_status, catalog_detail,
        None, last_sync_min if last_sync_min is not None else last_added.get('live')))
    services.append(_status_svc(
        'epg', 'Guia de programacao (EPG)',
        'ok' if counts['epg'] else 'degraded',
        f"{counts['epg']} programas, {counts['mappings']} canais com guia",
        None, epg_latest_min))

    # saude dos canais ao vivo
    ch_status = 'ok'
    ch_detail = 'sem verificacao de saude registrada'
    if (online + offline) > 0:
        pct_off = (offline / float(online + offline)) * 100.0
        ch_detail = f'{online} online / {offline} offline ({pct_off:.1f}%)'
        if pct_off >= 25:
            ch_status = 'down'
        elif pct_off >= 8:
            ch_status = 'degraded'
    services.append(_status_svc('channels', 'Saude dos canais', ch_status, ch_detail,
                                None, health_checked_min))


    # ── pagamentos e notificacoes ──
    pending_old = 0
    if db_ok:
        try:
            pend_since = datetime.now(timezone.utc) - timedelta(days=2)
            pending_old = Payment.query.filter(
                Payment.status == 'pending',
                Payment.created_at >= pend_since).count()
            metrics['notificacoes_24h'] = Notification.query.filter(
                Notification.created_at >= pend_since).count()
        except Exception:
            pass
    services.append(_status_svc(
        'payments', 'Pagamentos',
        'degraded' if pending_old > 20 else 'ok',
        f'{pending_old} pagamentos pendentes (2 dias)' if pending_old else 'nada pendente'))
    services.append(_status_svc(
        'notifications', 'Notificacoes', 'ok',
        f"{metrics.get('notificacoes_24h', 0)} enviadas nas ultimas 48h"))

    for e in errors:
        services.append(_status_svc(
            e['service'] + '_error', e['service'], 'down', e['message']))

    # ── tarefas agendadas ──
    cron_state = _cron_state()
    jobs = []
    for key, label, interval in CRON_JOBS:
        mins = interval
        if isinstance(interval, str):
            try:
                mins = int(settings.get(interval, 360) or 360)
            except Exception:
                mins = 360
        last_iso = cron_state.get(key)
        last_ts = None
        if last_iso:
            try:
                last_ts = datetime.fromisoformat(last_iso).timestamp()
            except Exception:
                last_ts = None
        age_min = None if last_ts is None else (now_ts - last_ts) / 60.0
        if age_min is None:
            jstatus = 'unknown'
        elif age_min <= max(mins * 2, 15):
            jstatus = 'ok'
        elif age_min <= max(mins * 4, 60):
            jstatus = 'degraded'
        else:
            jstatus = 'down'
        jobs.append({'key': key, 'label': label, 'interval_min': mins,
                     'last_run': last_iso, 'last_run_label': _status_minutes_label(age_min),
                     'age_min': None if age_min is None else round(age_min, 1),
                     'status': jstatus})
    job_status = 'ok'
    known = [j for j in jobs if j['status'] != 'unknown']
    if known and any(j['status'] == 'down' for j in known):
        job_status = 'degraded'
    services.append(_status_svc('jobs', 'Tarefas agendadas (cron)', job_status,
                                      f'{len(known)} de {len(jobs)} com execucao registrada'))

    # ── agregacao de estado ──
    statuses = [s['status'] for s in services]
    if 'down' in statuses:
        overall = 'down'
    elif 'degraded' in statuses or maint['enabled']:
        overall = 'degraded'
    else:
        overall = 'ok'

    history = _status_history_load()
    last_sample_ts = history[-1]['ts'] if history else 0
    sample = {
        'ts': int(now_ts),
        's': overall,
        'lat': site_ms,
        'db': db_ms,
        'c': [counts['live'], counts['movie'], counts['series']],
        'off': offline,
        'inc': (len(errors) + sync_error_24h + (1 if proxy_fails_24h >= 20 else 0)
                + (1 if maint['enabled'] else 0)),
        'svc': [s['key'] for s in services if s['status'] in ('down', 'degraded')][:8],
    }
    if (now_ts - last_sample_ts) >= STATUS_SAMPLE_SPAN:
        _status_append_sample(sample)
        history.append(sample)

    extra_incidents = []
    if last_sync_error:
        extra_incidents.append({
            'ts': now_ts - last_sync_error[0] * 60,
            'at': datetime.fromtimestamp(now_ts - last_sync_error[0] * 60,
                                         BRASILIA_TZ).isoformat(timespec='seconds'),
            'level': 'error', 'service': 'sync',
            'message': last_sync_error[1] or 'Sincronizacao do catalogo falhou'})

    report = {
        'overall': overall,
        'ts': int(now_ts),
        'checked_at': _status_now_br().isoformat(timespec='seconds'),
        'uptime_s': int(now_ts - APP_START_TS),
        'uptime_label': _status_uptime_label(now_ts - APP_START_TS),
        'response_ms': site_ms,
        'site': {'name': 'Troco Play', 'url': request.url_root if request else ''},
        'app_version': settings.get('app_version', ''),
        'apk_size': settings.get('apk_size', ''),
        'services': services,
        'sections': sections_out,
        'jobs': jobs,
        'metrics': metrics,
        'maintenance': maint,
        'history': _status_history_aggregate(history),
        'incidents': _status_incidents(history, recent_events + extra_incidents),
        'events': recent_events[:30],
    }
    if current_user and current_user.is_authenticated and current_user.is_admin:
        report['admin'] = _status_admin_view(settings, db_ok)
    _status_report_cache[ckey] = {'ts': _time.time(), 'data': report}
    return report


def _status_history_aggregate(history):
    """Samples brutos (últimas ~60 min) + blocos por hora e por dia."""
    now = _time.time()
    recent = [h for h in history if h['ts'] >= now - 3600]
    hourly = {}
    daily = {}
    for h in history:
        dt = datetime.fromtimestamp(h['ts'], BRASILIA_TZ)
        hk = dt.strftime('%Y-%m-%dT%H')
        dk = dt.strftime('%Y-%m-%d')
        for bucket, key in ((hourly, hk), (daily, dk)):
            b = bucket.setdefault(key, {'n': 0, 'up': 0, 'warn': 0, 'lat': 0.0, 'db': 0.0, 'inc': 0})
            b['n'] += 1
            state = h.get('s')
            if state != 'down':
                b['up'] += 1
            if state == 'degraded':
                b['warn'] += 1
            b['lat'] += float(h.get('lat') or 0)
            b['db'] += float(h.get('db') or 0)
            b['inc'] += int(h.get('inc') or 0)

    def finish(bucket, fmt_key):
        out = []
        for key in sorted(bucket.keys()):
            b = bucket[key]
            out.append({
                fmt_key: key,
                'samples': b['n'],
                'uptime': round((b['up'] / float(b['n'])) * 100.0, 2) if b['n'] else 0.0,
                'avg_ms': round(b['lat'] / b['n'], 1) if b['n'] else 0.0,
                'avg_db_ms': round(b['db'] / b['n'], 1) if b['n'] else 0.0,
                'attention': b['warn'],
                'incidents': b['inc'],
            })
        return out[-48:]

    return {'samples': recent[-70:],
            'hourly': finish(hourly, 'hour'),
            'daily': finish(daily, 'day')}


def _status_incident_run(run):
    svc = ', '.join(run['svc']) or 'sistema'
    down = run['s'] == 'down'
    message = 'Servico indisponivel' if down else 'Desempenho abaixo do ideal'
    if run['n'] > 1:
        message += ' por ' + _status_minutes_label((run['end'] - run['start']) / 60.0)
    return {'ts': run['end'],
            'at': datetime.fromtimestamp(run['end'], BRASILIA_TZ).isoformat(timespec='seconds'),
            'since': datetime.fromtimestamp(run['start'], BRASILIA_TZ).isoformat(timespec='seconds'),
            'level': 'error' if down else 'warn',
            'service': svc, 'message': message, 'samples': run['n']}


def _status_incidents(history, recent_events):
    """Lista navegavel do que saiu do normal nos ultimos dias.

    Amostras seguidas com o mesmo problema viram UM incidente (com duracao),
    senao cada minuto de degradado poluiria a lista com a mesma linha.
    """
    now = _time.time()
    window = sorted((h for h in history if h['ts'] >= now - 7 * 86400),
                    key=lambda h: h['ts'])
    out = []
    run = None
    for h in window:
        bad = h.get('s') != 'ok' or h.get('inc')
        key = (h.get('s'), tuple(sorted(h.get('svc') or [])))
        if bad and run and run['key'] == key:
            run['n'] += 1
            run['end'] = h['ts']
            continue
        if run:
            out.append(_status_incident_run(run))
            run = None
        if bad:
            run = {'key': key, 's': h.get('s'), 'start': h['ts'], 'end': h['ts'], 'n': 1,
                   'svc': sorted(h.get('svc') or [])}
    if run:
        out.append(_status_incident_run(run))
    for e in recent_events:
        if e['ts'] and e['ts'] >= now - 7 * 86400:
            out.append({'ts': e['ts'], 'at': e['at'], 'since': e['at'], 'samples': 1,
                        'level': e['level'], 'service': e['service'], 'message': e['message']})
    out.sort(key=lambda x: x['ts'], reverse=True)
    return out[:60]


def _status_source_host(url):
    """Mostra só o host da fonte: a URL completa carrega usuario/senha."""
    try:
        from urllib.parse import urlparse
        parsed = urlparse(str(url or ''))
        host = parsed.netloc.split('@')[-1]
        return host or ('arquivo local' if url else '-')
    except Exception:
        return '-'


def _status_admin_view(settings, db_ok):
    """Detalhes extras visiveis apenas para administradores."""
    info = {'environment': os.environ.get('CPANEL_MODE', ''),
            'python': sys.version.split()[0],
            'db_file': None, 'db_size_mb': None, 'settings_keys': len(settings)}
    try:
        db_file = APP.config.get('SQLALCHEMY_DATABASE_URI', '')
        if db_file.startswith('sqlite'):
            path = db_file.split('///', 1)[-1]
            info['db_file'] = os.path.basename(path)
            if os.path.exists(path):
                info['db_size_mb'] = round(os.path.getsize(path) / 1048576.0, 1)
    except Exception:
        pass
    if db_ok:
        try:
            info['usuarios'] = User.query.count()
            info['ativos'] = User.query.filter(
                User.is_paid == True,
                (User.paid_until.is_(None)) |
                (User.paid_until >= datetime.now(timezone.utc))).count()
            info['tickets_abertos'] = SupportTicket.query.filter(
                SupportTicket.status.in_(['aberto', 'em_atendimento'])).count()
        except Exception:
            pass
    # As fontes reais (sem as credenciais embutidas na URL) e os arquivos de
    # estado: sem isso o admin nao consegue diferenciar "cron atrasado" de
    # "fonte que nao responde".
    try:
        fontes = [f for f in ((settings.get('m3u_url') or '').strip(),
                              *[(x or '').strip() for x in (settings.get('m3u_urls') or [])]) if f]
        info['m3u_hosts'] = [_status_source_host(f) for f in fontes]
        info['epg_hosts'] = [_status_source_host(x.strip())
                             for x in str(settings.get('epg_url') or '').splitlines() if x.strip()]
    except Exception:
        info['m3u_hosts'] = []
        info['epg_hosts'] = []
    info['arquivos'] = {
        'historico': STATUS_HISTORY_PATH,
        'amostras': _status_history_line_count(),
        'retencao_dias': STATUS_RETENTION_DAYS,
        'amostragem_segundos': STATUS_SAMPLE_SPAN,
        'cron_state': CRON_STATE_PATH,
        'progresso_m3u': os.path.join(BASE_DIR, 'progress_m3u.json'),
        'log': os.path.join(BASE_DIR, 'bg_m3u.log'),
    }
    info['catalogo'] = {'app_version': settings.get('app_version', ''),
                        'apk_size': settings.get('apk_size', ''),
                        'release': settings.get('app_release_date', ''),
                        'free_daily_hours': settings.get('free_daily_hours', '')}
    info['cron_state'] = _cron_state()
    info['last_error_log'] = _status_tail_log('erro.log', 6)
    info['last_bg_log'] = _status_tail_log('bg_m3u.log', 10)
    return info


def _status_tail_log(name, n):
    try:
        path = os.path.join(BASE_DIR, name)
        if not os.path.exists(path):
            return []
        with open(path, 'r', errors='replace') as f:
            lines = f.readlines()
        return [ln.strip()[:300] for ln in lines[-n:]]
    except Exception:
        return []


CLIENT_STATE_LABEL = {'ok': 'No ar', 'attention': 'Com instabilidade',
                      'down': 'Fora do ar', 'maintenance': 'Em manutencao',
                      'access': 'Precisa de assinatura', 'unknown': 'Verificando'}


def _client_state(status):
    return {'ok': 'ok', 'degraded': 'attention', 'down': 'down',
            'maintenance': 'maintenance', 'access': 'access',
            'unknown': 'unknown'}.get(status, 'unknown')


def _client_view(report):
    """Projecao do relatorio para o cliente: so o status geral e as tres abas.

    O resto (latencia, banco, cron, saude dos canais, mensagens tecnicas,
    incidentes) fica no painel tecnico /admin/status.
    """
    by_key = {s.get('key'): s for s in report.get('services') or []}
    sections = {s.get('key'): s for s in report.get('sections') or []}
    maint = report.get('maintenance') or {}

    ordem = {'ok': 0, 'unknown': 1, 'access': 2, 'attention': 3,
             'maintenance': 4, 'down': 5}

    def pior(chaves):
        estados = [_client_state((by_key.get(k) or {}).get('status')) for k in chaves
                   if by_key.get(k)]
        return sorted(estados + ['unknown'], key=lambda x: -ordem.get(x, 0))[0]

    def item(chave, label, caminho, state):
        return {'key': chave, 'label': label, 'href': caminho, 'state': state,
                'text': CLIENT_STATE_LABEL.get(state, '')}

    canal_state = _client_state((sections.get('canais') or {}).get('status'))
    filme_state = _client_state((sections.get('filmes') or {}).get('status'))
    serie_state = _client_state((sections.get('series') or {}).get('status'))
    itens = [item('canais', 'Canais ao vivo', '/canais', canal_state),
             item('filmes', 'Filmes', '/filmes', filme_state),
             item('series', 'Series', '/series', serie_state)]

    # O status geral e o pior entre as abas que o cliente usa e o acesso ao site
    # (banco/login/manutencao), porque "nao abre" também é uma queda percebida.
    # 'response' fica de fora de proposito: a latencia medida e a do proprio
    # relatorio, entao abrir a pagina de Status deixava tudo "instavel" sem nada
    # ter acontecido para o cliente.
    base = pior(['site', 'database', 'auth'])
    estados = [i['state'] for i in itens] + [base]
    if maint.get('active_for_you'):
        estado_geral = 'maintenance'
    elif 'down' in estados:
        estado_geral = 'down'
    elif 'attention' in estados:
        estado_geral = 'attention'
    else:
        estado_geral = 'ok'

    nomes = {'canais': 'os canais ao vivo', 'filmes': 'os filmes', 'series': 'as series'}
    afetados = [nomes[i['key']] for i in itens if i['state'] in ('down', 'attention')]
    headline = {'ok': 'Tudo funcionando',
                'attention': 'Estamos com pequenas oscilacoes',
                'down': 'Temos uma interrupcao agora',
                'maintenance': 'Site em manutencao'}[estado_geral]
    if estado_geral in ('down', 'attention') and afetados:
        if len(afetados) == 1:
            headline += ' — ' + afetados[0]
        else:
            headline += ' — ' + ', '.join(afetados[:-1]) + ' e ' + afetados[-1]

    if estado_geral == 'maintenance':
        detalhe = maint.get('message') or 'Estamos melhorando algumas areas agora, ja ja volta.'
    elif estado_geral == 'ok':
        detalhe = 'Todos os conteudos estao no ar para voce.'
    else:
        detalhe = 'Nossa equipe ja foi avisada e esta resolvendo. Tente de novo em alguns minutos.'

    return {
        'state': estado_geral,
        'headline': headline,
        'detail': detalhe,
        'checked_at': report.get('checked_at'),
        'ts': report.get('ts'),
        'refresh_seconds': 60,
        'items': itens,
    }


@login_required
def status_page():
    """Aba Status (site PC, celular e app: mesmo HTML)."""
    return render_template('status.html')


@login_required
def api_status():
    try:
        data = _client_view(_status_report())
    except Exception:
        data = {'state': 'unknown',
                'headline': 'Nao foi possivel verificar agora',
                'detail': 'Tente atualizar daqui a pouco. Se um video nao abrir, recarregue a pagina.',
                'checked_at': _status_now_br().isoformat(timespec='seconds'),
                'ts': int(_time.time()),
                'refresh_seconds': 60, 'items': []}
    return jsonify(data), 200, {'Cache-Control': 'no-store'}


def admin_status_page():
    """Painel tecnico completo (so administrador).

    O relatorio ja vem embutido na primeira resposta (bloco <script id="adBoot">):
    assim o painel pinta de imediato e nao depende do fetch para mostrar algo.
    """
    try:
        boot = _status_report()
    except Exception as exc:
        APP.logger.exception('relatorio de status falhou na pagina do admin')
        boot = {'overall': 'down',
                'checked_at': _status_now_br().isoformat(timespec='seconds'),
                'ts': int(_time.time()), 'services': [], 'sections': [], 'jobs': [],
                'metrics': {}, 'history': {'samples': [], 'hourly': [], 'daily': []},
                'incidents': [{'ts': _time.time(), 'level': 'error', 'service': 'status',
                               'message': 'Geracao do relatorio falhou: ' + str(exc)[:200]}],
                'events': []}
    try:
        return render_template('admin_status.html', boot=boot)
    except Exception as exc:
        APP.logger.exception('falha ao renderizar admin_status.html')
        # Nunca entregar tela preta: o painel e JS-only, entao um template ausente
        # ou quebrado pareceria "nao apareceu nada" para o administrador.
        msg = ('Painel de Status indisponivel.\n\n'
               'Erro: ' + type(exc).__name__ + ': ' + str(exc) + '\n\n'
               'Causa mais provavel: templates/admin_status.html nao foi enviado '
               'para o servidor (ou o processo Python nao foi reiniciado depois do '
               'deploy).\n')
        return msg, 500, {'Content-Type': 'text/plain; charset=utf-8'}


def api_admin_status():
    try:
        data = _status_report()
    except Exception as exc:
        data = {'overall': 'down',
                'checked_at': _status_now_br().isoformat(timespec='seconds'),
                'ts': int(_time.time()),
                'services': [], 'sections': [], 'jobs': [], 'metrics': {},
                'history': {'samples': [], 'hourly': [], 'daily': []},
                'incidents': [{'ts': _time.time(), 'level': 'error', 'service': 'status',
                               'message': f'Geracao do relatorio falhou: {str(exc)[:200]}'}],
                'events': [], 'errors': [{'service': 'status',
                                          'message': f'Relatorio falhou: {str(exc)[:200]}'}],
                'maintenance': status_maintenance_state(load_settings(), 'site')}
    return jsonify(data), 200, {'Cache-Control': 'no-store'}


def status_sample_tick():
    """Amostra o status sem depender de visitantes (scheduler ou cron do cPanel)."""
    _status_report_cache.clear()
    with APP.app_context(), APP.test_request_context('/api/status'):
        report = _status_report()
    _record_cron_run('status-sample')
    return report['overall']


def cron_status_sample():
    """Gera uma amostra mesmo sem ninguem visitando (cron do cPanel)."""
    _cron_auth()
    try:
        return f"OK {status_sample_tick()}", 200
    except Exception as e:
        return f'ERR: {e}', 500