ccq-admin@sm-ccqsv001:~$ python final17.py
2024-11-11 00:36:16 [INFO] Iniciando servidor en el puerto 8080
2024-11-11 00:36:16 [INFO] Accesible en las siguientes URLs:
2024-11-11 00:36:16 [INFO] http://10.0.3.3:8080/data
2024-11-11 00:36:16 [INFO] El servidor está ahora en funcionamiento
2024-11-11 00:36:22 [INFO] Petición recibida en /data a las 2024-11-11 00:36:22.564489
2024-11-11 00:36:22 [INFO] ¡Inicio de sesión exitoso!
2024-11-11 00:36:22 [INFO] Procesando lote 1
2024-11-11 00:36:26 [INFO] Datos obtenidos exitosamente para PON 1, Grupo ONU 0
2024-11-11 00:36:27 [INFO] Datos obtenidos exitosamente para PON 1, Grupo ONU 1
2024-11-11 00:36:30 [INFO] Datos obtenidos exitosamente para PON 2, Grupo ONU 0
2024-11-11 00:36:30 [INFO] Datos obtenidos exitosamente para PON 2, Grupo ONU 1
2024-11-11 00:36:32 [INFO] Datos obtenidos exitosamente para PON 3, Grupo ONU 1
2024-11-11 00:36:35 [INFO] Datos obtenidos exitosamente para PON 3, Grupo ONU 0
2024-11-11 00:36:39 [INFO] Datos obtenidos exitosamente para PON 4, Grupo ONU 0
2024-11-11 00:36:42 [INFO] Datos obtenidos exitosamente para PON 4, Grupo ONU 1
/home/ccq-admin/final17.py:279: DeprecationWarning: drain method is deprecated, use await resp.write()
  await resp.drain()
2024-11-11 00:36:42 [INFO] Procesando lote 2
2024-11-11 00:36:46 [INFO] Datos obtenidos exitosamente para PON 6, Grupo ONU 0
2024-11-11 00:36:48 [INFO] Datos obtenidos exitosamente para PON 5, Grupo ONU 1

just fix that warning, no more

import argparse
import asyncio
import json
import logging
import signal
import ssl
import sys
from datetime import datetime
from functools import lru_cache, wraps
from typing import Dict, List, Optional, Tuple, Callable

import aiohttp
from aiohttp import web
from aiohttp.client_exceptions import ClientError
import colorlog
import netifaces
from bs4 import BeautifulSoup

# Configuración del script
CONFIG = {
    'BASE_URL': 'https://10.0.0.2',
    'USERNAME': 'misael_campos',
    'PASSWORD': 'd337O8AetR3p4',
    'POST_ENDPOINT': '/action/pononuopticalinfo.html',
    'WHO': 100,
    'ONUID': 0,
    'PON_RANGE': range(1, 17),
    'ONU_GROUP_RANGE': range(0, 2),
    'REQUEST_TIMEOUT': aiohttp.ClientTimeout(
        total=240,
        connect=120,
        sock_read=120
    ),
    'MAX_CONCURRENT_REQUESTS': 8,
    'MAX_RETRIES': 3,
    'RETRY_DELAY': 1,
    'BATCH_SIZE': 8,
    'HOST': '0.0.0.0',
    'PORT': 8080,
    'ENDPOINT': '/data',
    'IGNORE_ROW_LENGTH_WARNING': False
}

# Mapeo de nombres de campos para el JSON final
FIELD_MAP = {
    "ONU ID": "onu_id",
    "Desc": "name",
    "RX Power(ONU)": "rx_power_onu",
    "TX Power(ONU)": "tx_power_onu",
    "RX Power(OLT)": "rx_power_olt",
}

def setup_logger() -> logging.Logger:
    """Configura y retorna una instancia de logger con color."""
    logger = colorlog.getLogger('AsyncScraperLogger')
    if not logger.handlers:
        handler = colorlog.StreamHandler()
        handler.setFormatter(colorlog.ColoredFormatter(
            '%(log_color)s%(asctime)s [%(levelname)s] %(message)s',
            datefmt='%Y-%m-%d %H:%M:%S',
            log_colors={
                'DEBUG': 'light_cyan',
                'INFO': 'light_green',
                'WARNING': 'light_yellow',
                'ERROR': 'light_red',
                'CRITICAL': 'light_red,bg_white',
            }
        ))
        logger.addHandler(handler)
        logger.setLevel(logging.INFO)
        logger.propagate = False
    return logger

logger = setup_logger()

def retry(max_retries: int = CONFIG['MAX_RETRIES'], delay: int = CONFIG['RETRY_DELAY'], logger: Optional[logging.Logger] = None):
    """
    Decorador para manejar reintentos de una función asíncrona.

    Args:
        max_retries (int): Número máximo de intentos.
        delay (int): Retardo entre reintentos en segundos.
        logger (Optional[logging.Logger]): Logger para registrar eventos.

    Returns:
        Callable: Función decorada con lógica de reintentos.
    """
    def decorator(func: Callable):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(1, max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except (aiohttp.ClientResponseError, aiohttp.ClientConnectorError, asyncio.TimeoutError, ClientError) as e:
                    if logger:
                        logger.warning(f"Intento {attempt} de {max_retries} fallido: {e}")
                    if attempt == max_retries:
                        if logger:
                            logger.error(f"Todos los {max_retries} intentos fallaron para {func.__name__}")
                        raise
                    await asyncio.sleep(delay)
        return wrapper
    return decorator

# Factory para la configuración
class ConfigFactory:
    """Factory para obtener la configuración."""

    @staticmethod
    def get_config(env_config: Optional[Dict] = None) -> Dict:
        default_config = CONFIG.copy()
        if env_config:
            default_config.update(env_config)
        return default_config

# Factory para AsyncScraper
class ScraperFactory:
    """Factory para crear instancias de AsyncScraper."""

    @staticmethod
    def create_scraper() -> 'AsyncScraper':
        return AsyncScraper(
            base_url=CONFIG['BASE_URL'],
            config=CONFIG,
            logger=logger
        )

@lru_cache(maxsize=None)
def get_all_ip_addresses(logger: logging.Logger) -> List[str]:
    """Obtiene todas las direcciones IPv4 no locales de las interfaces de red de la máquina."""
    try:
        return [
            addr.get('addr')
            for iface in netifaces.interfaces()
            for addr in netifaces.ifaddresses(iface).get(netifaces.AF_INET, [])
            if addr.get('addr') and addr.get('addr') != '127.0.0.1'
        ]
    except ValueError as ve:
        logger.error(f"Error al obtener direcciones IP: {ve}")
    except Exception as e:
        logger.error(f"Error inesperado al obtener direcciones IP: {e}")
    return []

def generate_all_combinations(config: Dict) -> List[Tuple[int, int]]:
    """Genera todas las combinaciones de PON y grupos ONU."""
    return [(pon, onu_group) for pon in config['PON_RANGE'] for onu_group in config['ONU_GROUP_RANGE']]

class AsyncScraper:
    """Scraper asíncrono para obtener información óptica de un router."""

    def __init__(self, base_url: str, config: Dict, logger: logging.Logger):
        self.base_url = base_url.rstrip('/')
        self.config = config
        self.logger = logger
        self.ssl_context = ssl.create_default_context()
        self.ssl_context.check_hostname = False
        self.ssl_context.verify_mode = ssl.CERT_NONE
        self.session: Optional[aiohttp.ClientSession] = None
        self.sem = asyncio.Semaphore(self.config['MAX_CONCURRENT_REQUESTS'])

    async def __aenter__(self):
        timeout = self.config['REQUEST_TIMEOUT']
        self.session = aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(
                ssl=self.ssl_context,
                limit=self.config['MAX_CONCURRENT_REQUESTS']
            ),
            timeout=timeout,
            headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'}
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()

    @retry(logger=logger)
    async def login(self, username: str, password: str) -> bool:
        login_url = f"{self.base_url}/action/main.html"
        login_data = {
            'user': username,
            'pass': password,
            'button': 'Login',
            'who': str(self.config['WHO']),
        }
        async with self.session.post(login_url, data=login_data) as response:
            response.raise_for_status()
            text = await response.text()
            if "Sorry, you do not have access" in text:
                self.logger.error("Error de inicio de sesión: Acceso denegado")
                return False
            elif "Error" in text or "Incorrect password" in text:
                self.logger.error("Error de inicio de sesión: Credenciales incorrectas")
                return False
            self.logger.info("¡Inicio de sesión exitoso!")
            return True

    @retry(logger=logger)
    async def fetch_pon_onu_optical_info(self, pon: int, onu_group: int) -> Optional[str]:
        post_url = f"{self.base_url}{self.config['POST_ENDPOINT']}"
        payload = {
            'pon': pon,
            'onu_group': onu_group,
            'who': str(self.config['WHO']),
            'onuid': str(self.config['ONUID'])
        }
        async with self.sem, self.session.post(post_url, data=payload) as response:
            response.raise_for_status()
            content = await response.text()
            self.logger.info(
                f"Datos obtenidos exitosamente para PON {pon}, Grupo ONU {onu_group}"
            )
            return content

    @staticmethod
    def parse_html_table(html_content: str, logger: logging.Logger,
                         ignore_row_length_warning: bool) -> List[Dict[str, Optional[str]]]:
        """Analiza una tabla HTML y extrae datos en una lista de diccionarios."""
        if not html_content:
            return []
        soup = BeautifulSoup(html_content, 'lxml')
        table = soup.find('table', border="1", cellpadding="4", cellspacing="0")
        if not table:
            logger.warning("No se encontró tabla en el contenido HTML")
            return []
        headers = [
            th.get_text(strip=True) or f"Columna_{i + 1}"
            for i, th in enumerate(table.find('tr').find_all(['td', 'th']))
        ]
        mapped_headers = [FIELD_MAP.get(header, header.lower().replace(' ', '_')) for header in headers]
        data = []
        for row in table.find_all('tr')[1:]:
            cells = row.find_all('td')
            if len(cells) != len(mapped_headers):
                if not ignore_row_length_warning:
                    logger.warning("La fila no coincide con la longitud del encabezado. Omitiendo fila")
                continue
            row_data = {mapped: cell.get_text(strip=True) or None for mapped, cell in zip(mapped_headers, cells)}
            data.append(row_data)
        return data

async def fetch_and_parse(scraper: AsyncScraper, pon: int, onu_group: int) -> List[Dict[str, Optional[str]]]:
    """Obtiene y analiza información óptica para un PON y grupo ONU específico."""
    try:
        html_content = await scraper.fetch_pon_onu_optical_info(pon, onu_group)
        parsed_data = AsyncScraper.parse_html_table(
            html_content,
            scraper.logger,
            scraper.config['IGNORE_ROW_LENGTH_WARNING']
        ) if html_content else []
        if not parsed_data:
            scraper.logger.warning(f"No se analizaron datos para PON {pon}, Grupo ONU {onu_group}")
        return parsed_data
    except Exception as e:
        scraper.logger.error(f"Error en fetch_and_parse: {e}")
        return []

async def process_batch(scraper: AsyncScraper, batch: List[Tuple[int, int]]) -> List[Dict[str, Optional[str]]]:
    """Procesa un lote de combinaciones de PON y grupo ONU."""
    tasks = [fetch_and_parse(scraper, pon, onu_group) for pon, onu_group in batch]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    flat_results = []
    for result in results:
        if isinstance(result, Exception):
            scraper.logger.error(f"La tarea del lote falló con excepción: {result}")
            continue
        flat_results.extend(result)
    return flat_results

async def write_json_record(resp: web.StreamResponse, record: Dict[str, Optional[str]], first_record: bool):
    """Escribe un registro JSON en la respuesta HTTP."""
    if not first_record:
        await resp.write(b',\n')
    else:
        first_record = False
    record_json = json.dumps(record, indent=4, ensure_ascii=False)
    indented_record = '\n'.join(['    ' + line for line in record_json.split('\n')])
    await resp.write(indented_record.encode('utf-8'))
    await resp.drain()
    return first_record

async def handle_data(request: web.Request) -> web.StreamResponse:
    """Manejador HTTP para el endpoint configurado. Transmite datos JSON de forma incremental."""
    start_time = datetime.now()
    logger.info(f"Petición recibida en {CONFIG['ENDPOINT']} a las {start_time}")

    resp = web.StreamResponse(
        status=200,
        reason='OK',
        headers={
            'Content-Type': 'application/json',
            'Transfer-Encoding': 'chunked',
        }
    )
    await resp.prepare(request)

    try:
        scraper = ScraperFactory.create_scraper()
        async with scraper:
            if not await scraper.login(CONFIG['USERNAME'], CONFIG['PASSWORD']):
                error_data = {"error": "Error de inicio de sesión. Por favor, verifique las credenciales."}
                await resp.write(json.dumps(error_data, indent=2, ensure_ascii=False).encode('utf-8'))
                await resp.write_eof()
                return resp

            await resp.write(b'{\n  "data": [\n')
            first_record = True

            all_combinations = generate_all_combinations(scraper.config)
            for i in range(0, len(all_combinations), scraper.config['BATCH_SIZE']):
                batch = all_combinations[i:i + scraper.config['BATCH_SIZE']]
                logger.info(f"Procesando lote {i // scraper.config['BATCH_SIZE'] + 1}")
                results = await process_batch(scraper, batch)
                for record in results:
                    first_record = await write_json_record(resp, record, first_record)

            end_time = datetime.now()
            duration = end_time - start_time
            final_json = f'\n  ],\n  "processing_time": "{duration}"\n}}\n'
            await resp.write(final_json.encode('utf-8'))
            await resp.write_eof()

            logger.info(f"Recopilación de datos completa. Duración: {duration}")

    except ClientError as e:
        logger.error(f"Error de cliente en handle_data: {e}")
        error_event = {"error": f"Error de cliente: {e}"}
        await resp.write(json.dumps(error_event, indent=2, ensure_ascii=False).encode('utf-8'))
        await resp.write_eof()
    except asyncio.CancelledError:
        logger.warning("La petición fue cancelada.")
        await resp.write_eof()
    except Exception as e:
        logger.error(f"Error inesperado en handle_data: {e}")
        error_event = {"error": f"Error inesperado: {e}"}
        await resp.write(json.dumps(error_event, indent=2, ensure_ascii=False).encode('utf-8'))
        await resp.write_eof()

    return resp

async def init_app() -> web.Application:
    """Inicializa la aplicación web aiohttp."""
    app = web.Application()
    app.router.add_get(CONFIG['ENDPOINT'], handle_data)
    return app

def parse_args() -> argparse.Namespace:
    """Analiza los argumentos de línea de comandos."""
    parser = argparse.ArgumentParser(description="Servidor Scraper Asíncrono")
    parser.add_argument(
        '--ignore-row-length-warnings',
        action='store_true',
        help='Ignorar advertencias sobre desajustes en la longitud de filas en tablas HTML.'
    )
    return parser.parse_args()

async def main():
    """Punto de entrada principal para ejecutar el servidor web con apagado controlado."""
    args = parse_args()
    CONFIG['IGNORE_ROW_LENGTH_WARNING'] = args.ignore_row_length_warnings

    local_ips = get_all_ip_addresses(logger)
    if not local_ips:
        logger.warning("No se encontraron direcciones IPv4 no locales. Usando 127.0.0.1 como respaldo")
        local_ips = [CONFIG['HOST']]

    endpoint = CONFIG.get('ENDPOINT', '/data')
    logger.info(f"Iniciando servidor en el puerto {CONFIG['PORT']}")
    logger.info("Accesible en las siguientes URLs:")
    for ip in local_ips:
        logger.info(f"http://{ip}:{CONFIG['PORT']}{endpoint}")

    app = await init_app()
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, host=CONFIG['HOST'], port=CONFIG['PORT'])
    await site.start()

    logger.info("El servidor está ahora en funcionamiento")

    shutdown_event = asyncio.Event()

    def shutdown_signal_handler():
        logger.info("Señal de apagado recibida")
        shutdown_event.set()

    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        try:
            loop.add_signal_handler(sig, shutdown_signal_handler)
        except NotImplementedError:
            logger.warning(f"Manejadores de señales no implementados para {sig}")

    await shutdown_event.wait()
    logger.info("Apagando servidor...")
    await runner.shutdown()
    await runner.cleanup()
    logger.info("Servidor apagado exitosamente.")

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        logger.critical(f"Excepción no manejada: {e}")
        sys.exit(1)