api/m3u_parser.py

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
import requests
import re
import os
from typing import List, Dict, Optional
from datetime import datetime, timedelta
from auth import get_current_user

router = APIRouter()

# Cache en memoria
m3u_cache = {
    "data": None,
    "timestamp": None,
    "ttl": 3600  # 1 hora
}

class Channel(BaseModel):
    id: str
    name: str
    url: str
    logo: Optional[str] = None
    category: str
    group: Optional[str] = None

class ChannelCategory(BaseModel):
    name: str
    count: int

class ChannelResponse(BaseModel):
    channels: List[Channel]
    categories: List[ChannelCategory]

def parse_m3u_content(content: str) -> List[Channel]:
    """Parsea el contenido M3U y extrae los canales"""
    channels = []
    lines = content.strip().split('\n')
    
    i = 0
    while i < len(lines):
        line = lines[i].strip()
        
        if line.startswith('#EXTINF:'):
            # Extraer información del canal
            info_line = line
            
            # Buscar la URL en la siguiente línea
            url_line = ""
            if i + 1 < len(lines):
                url_line = lines[i + 1].strip()
            
            if url_line and not url_line.startswith('#'):
                channel = parse_extinf_line(info_line, url_line)
                if channel:
                    channels.append(channel)
                i += 2
            else:
                i += 1
        else:
            i += 1
    
    return channels

def parse_extinf_line(extinf_line: str, url: str) -> Optional[Channel]:
    """Parsea una línea EXTINF y extrae la información del canal"""
    try:
        # Extraer nombre del canal (después de la última coma)
        name_match = extinf_line.split(',')[-1].strip()
        if not name_match:
            return None
        
        name = name_match
        
        # Extraer logo
        logo_match = re.search(r'tvg-logo="([^"]*)"', extinf_line)
        logo = logo_match.group(1) if logo_match else None
        
        # Extraer categoría/grupo
        group_match = re.search(r'group-title="([^"]*)"', extinf_line)
        category = group_match.group(1) if group_match else "General"
        
        # Generar ID único
        channel_id = f"{hash(name + url) % 1000000:06d}"
        
        return Channel(
            id=channel_id,
            name=name,
            url=url,
            logo=logo,
            category=category,
            group=category
        )
    except Exception as e:
        print(f"Error parsing channel: {e}")
        return None

def get_cached_channels() -> Optional[List[Channel]]:
    """Obtiene los canales del caché si están vigentes"""
    if not m3u_cache["data"] or not m3u_cache["timestamp"]:
        return None
    
    # Verificar TTL
    if datetime.now() - m3u_cache["timestamp"] > timedelta(seconds=m3u_cache["ttl"]):
        return None
    
    return m3u_cache["data"]

def cache_channels(channels: List[Channel]):
    """Guarda los canales en caché"""
    m3u_cache["data"] = channels
    m3u_cache["timestamp"] = datetime.now()

async def fetch_m3u_playlist() -> List[Channel]:
    """Descarga y parsea la playlist M3U"""
    m3u_url = os.getenv("MAIN_M3U_URL")
    if not m3u_url:
        raise HTTPException(status_code=500, detail="URL de playlist no configurada")
    
    try:
        response = requests.get(m3u_url, timeout=30)
        response.raise_for_status()
        
        channels = parse_m3u_content(response.text)
        cache_channels(channels)
        
        return channels
    except requests.RequestException as e:
        raise HTTPException(status_code=500, detail=f"Error descargando playlist: {str(e)}")

@router.get("/channels", response_model=ChannelResponse)
async def get_channels(current_user: str = Depends(get_current_user)):
    """Obtiene la lista de canales disponibles"""
    
    # Intentar obtener del caché primero
    cached_channels = get_cached_channels()
    if cached_channels:
        channels = cached_channels
    else:
        # Descargar y parsear
        channels = await fetch_m3u_playlist()
    
    # Agrupar por categorías
    categories_dict = {}
    for channel in channels:
        category = channel.category
        if category not in categories_dict:
            categories_dict[category] = 0
        categories_dict[category] += 1
    
    categories = [
        ChannelCategory(name=name, count=count)
        for name, count in sorted(categories_dict.items())
    ]
    
    return ChannelResponse(channels=channels, categories=categories)

@router.get("/playlist.m3u")
async def get_playlist_m3u(current_user: str = Depends(get_current_user)):
    """Devuelve la playlist M3U con URLs reescritas para usar el proxy"""
    
    channels = get_cached_channels()
    if not channels:
        channels = await fetch_m3u_playlist()
    
    # Generar contenido M3U con URLs proxy
    base_url = os.getenv("BASE_URL", "http://localhost:8000")
    m3u_content = "#EXTM3U\n"
    
    for channel in channels:
        proxy_url = f"{base_url}/api/proxy?url={requests.utils.quote(channel.url, safe='')}"
        
        extinf_line = f'#EXTINF:-1'
        if channel.logo:
            extinf_line += f' tvg-logo="{channel.logo}"'
        if channel.category:
            extinf_line += f' group-title="{channel.category}"'
        extinf_line += f',{channel.name}\n'
        
        m3u_content += extinf_line
        m3u_content += f"{proxy_url}\n"
    
    return PlainTextResponse(
        content=m3u_content,
        media_type="application/vnd.apple.mpegurl",
        headers={"Content-Disposition": "attachment; filename=playlist.m3u"}
    )