rate this code
ccq-admin@sm-ccqsv001:~$ cat final11.py 
import asyncio
import aiohttp
from aiohttp import web
from aiohttp.client_exceptions import ClientError
from bs4 import BeautifulSoup
import json
from typing import List, Dict, Optional
import ssl
import logging
from datetime import datetime
import colorlog
import socket
import netifaces
import argparse
import sys
import signal


# Configuration settings
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,  # Total operation timeout
        connect=120,  # Connection timeout
        sock_read=120  # Socket read timeout
    ),
    'max_concurrent_requests': 8,  # Reduced to prevent server overload
    'max_retries': 3,  # Maximum number of retry attempts
    'retry_delay': 1,  # Delay between retries in seconds
    'batch_size': 8,  # Number of concurrent requests per batch
    'host': '0.0.0.0',  # Host address
    'port': 8080,  # Port number
    'endpoint': '/data',  # Added endpoint path
    'ignore_row_length_warning': False  # New configuration option
}

# Configure logging
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 = colorlog.getLogger(__name__)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Remove any existing handlers to avoid duplicate logs
for hdlr in logger.handlers[:-1]:
    logger.removeHandler(hdlr)


def get_all_ip_addresses() -> List[str]:
    """
    Retrieve all non-loopback IPv4 addresses from the machine's network interfaces.

    Returns:
        List[str]: A list of IPv4 addresses.
    """
    ips = []
    try:
        interfaces = netifaces.interfaces()
        for iface in interfaces:
            addrs = netifaces.ifaddresses(iface)
            ipv4 = addrs.get(netifaces.AF_INET)
            if ipv4:
                for addr in ipv4:
                    ip = addr.get('addr')
                    if ip and ip != '127.0.0.1':
                        ips.append(ip)
    except ValueError as ve:
        logger.error(f"Error retrieving IP addresses: {ve}")
    except Exception as e:
        logger.error(f"Unexpected error retrieving IP addresses: {e}")
    return ips


class AsyncScraper:
    """
    Asynchronous scraper for fetching optical information from a router.

    Attributes:
        base_url (str): The base URL of the router.
        ssl_context (ssl.SSLContext): SSL context for HTTPS requests.
        session (aiohttp.ClientSession): The HTTP session for making requests.
        sem (asyncio.Semaphore): Semaphore to limit concurrent requests.
    """

    def __init__(self, base_url: str):
        """
        Initialize the AsyncScraper with the given base URL.

        Args:
            base_url (str): The base URL of the router.
        """
        self.base_url = base_url.rstrip('/')
        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(CONFIG['max_concurrent_requests'])

    async def __aenter__(self):
        """
        Enter the asynchronous context manager, initializing the HTTP session.

        Returns:
            AsyncScraper: The instance itself with an active session.
        """
        timeout = CONFIG['request_timeout']
        self.session = aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(
                ssl=self.ssl_context,
                limit=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):
        """
        Exit the asynchronous context manager, closing the HTTP session.

        Args:
            exc_type: Exception type.
            exc_val: Exception value.
            exc_tb: Exception traceback.
        """
        if self.session:
            await self.session.close()

    async def login(self, username: str, password: str) -> bool:
        """
        Perform login to the router.

        Args:
            username (str): The username for authentication.
            password (str): The password for authentication.

        Returns:
            bool: True if login is successful, False otherwise.
        """
        login_url = f"{self.base_url}/action/main.html"
        login_data = {
            'user': username,
            'pass': password,
            'button': 'Login',
            'who': '100',
        }

        for attempt in range(CONFIG['max_retries']):
            try:
                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:
                        logger.error("Login failed: Access denied")
                        return False
                    elif "Error" in text or "Incorrect password" in text:
                        logger.error("Login failed: Incorrect credentials")
                        return False
                    logger.info("Login successful!")
                    return True
            except aiohttp.ClientResponseError as e:
                logger.error(f"HTTP error during login attempt {attempt + 1}: {e}")
            except aiohttp.ClientConnectorError as e:
                logger.error(f"Connection error during login attempt {attempt + 1}: {e}")
            except asyncio.TimeoutError:
                logger.error(f"Timeout during login attempt {attempt + 1}")
            except Exception as e:
                logger.error(f"Unexpected error during login attempt {attempt + 1}: {e}")

            if attempt < CONFIG['max_retries'] - 1:
                await asyncio.sleep(CONFIG['retry_delay'])
                logger.info(f"Retrying login (Attempt {attempt + 2})...")
            else:
                logger.error("All login attempts failed.")
        return False

    async def fetch_pon_onu_optical_info(self, pon: int, onu_group: int) -> Optional[str]:
        """
        Fetch PON ONU optical information from the router.

        Args:
            pon (int): PON identifier.
            onu_group (int): ONU group identifier.

        Returns:
            Optional[str]: The HTML content if successful, None otherwise.
        """
        async with self.sem:
            post_url = f"{self.base_url}{CONFIG['post_endpoint']}"
            payload = {
                'pon': pon,
                'onu_group': onu_group,
                'who': CONFIG['who'],
                'onuid': CONFIG['onuid']
            }

            for attempt in range(CONFIG['max_retries']):
                try:
                    async with self.session.post(post_url, data=payload) as response:
                        response.raise_for_status()
                        content = await response.text()
                        logger.info(
                            f"Successfully fetched data for PON {pon}, ONU Group {onu_group} "
                            f"(Attempt {attempt + 1})"
                        )
                        return content
                except aiohttp.ClientResponseError as e:
                    logger.warning(
                        f"HTTP error {e.status} for PON {pon}, ONU Group {onu_group} "
                        f"(Attempt {attempt + 1}): {e.message}"
                    )
                except aiohttp.ClientConnectorError as e:
                    logger.warning(
                        f"Connection error for PON {pon}, ONU Group {onu_group} "
                        f"(Attempt {attempt + 1}): {e}"
                    )
                except asyncio.TimeoutError:
                    logger.warning(
                        f"Timeout fetching PON {pon}, ONU Group {onu_group} "
                        f"(Attempt {attempt + 1})"
                    )
                except Exception as e:
                    logger.warning(
                        f"Unexpected error fetching PON {pon}, ONU Group {onu_group} "
                        f"(Attempt {attempt + 1}): {e}"
                    )

                if attempt < CONFIG['max_retries'] - 1:
                    await asyncio.sleep(CONFIG['retry_delay'])
                    logger.info(f"Retrying PON {pon}, ONU Group {onu_group}...")
                else:
                    logger.error(
                        f"All attempts failed for PON {pon}, ONU Group {onu_group}"
                    )

            return None

    @staticmethod
    def parse_html_table(html_content: str) -> List[Dict[str, Optional[str]]]:
        """
        Parse an HTML table and extract data into a list of dictionaries.

        Args:
            html_content (str): The HTML content containing the table.

        Returns:
            List[Dict[str, Optional[str]]]: Parsed table data.
        """
        if not html_content:
            return []

        soup = BeautifulSoup(html_content, 'html.parser')
        table = soup.find('table', border="1", cellpadding="4", cellspacing="0")

        if not table:
            logger.warning("No table found in the HTML content")
            return []

        headers = []
        header_row = table.find('tr')
        if not header_row:
            logger.warning("No header row found in the table")
            return []

        for th in header_row.find_all(['td', 'th']):
            header_text = th.get_text(strip=True)
            headers.append(header_text if header_text else f"Column_{len(headers) + 1}")

        data = []
        for row in table.find_all('tr')[1:]:
            cells = row.find_all('td')
            if len(cells) != len(headers):
                if not CONFIG['ignore_row_length_warning']:
                    logger.warning("Row does not match header length. Skipping row")
                continue

            row_data = {
                header: cell.get_text(strip=True) or None
                for header, cell in zip(headers, cells)
            }
            data.append(row_data)

        return data


async def fetch_and_parse(scraper: AsyncScraper, pon: int, onu_group: int) -> Dict[int, Dict[int, List[Dict[str, Optional[str]]]]]:
    """
    Fetch and parse optical information for a specific PON and ONU group.

    Args:
        scraper (AsyncScraper): The scraper instance.
        pon (int): PON identifier.
        onu_group (int): ONU group identifier.

    Returns:
        Dict[int, Dict[int, List[Dict[str, Optional[str]]]]]: Parsed data structured by PON and ONU group.
    """
    html_content = await scraper.fetch_pon_onu_optical_info(pon, onu_group)
    parsed_data = AsyncScraper.parse_html_table(html_content) if html_content else []
    if not parsed_data:
        logger.warning(f"No data parsed for PON {pon}, ONU Group {onu_group}")
    return {pon: {onu_group: parsed_data}}


async def process_batch(scraper: AsyncScraper, batch: List[tuple]) -> List[Dict]:
    """
    Process a batch of PON and ONU group combinations.

    Args:
        scraper (AsyncScraper): The scraper instance.
        batch (List[tuple]): A list of (pon, onu_group) tuples.

    Returns:
        List[Dict]: A list of parsed data dictionaries.
    """
    tasks = [fetch_and_parse(scraper, pon, onu_group) for pon, onu_group in batch]
    return await asyncio.gather(*tasks, return_exceptions=True)


async def collect_data(scraper: AsyncScraper) -> Dict[int, Dict[int, List[Dict[str, Optional[str]]]]]:
    """
    Collect data by fetching and parsing information for all PON and ONU group combinations.

    Args:
        scraper (AsyncScraper): The scraper instance.

    Returns:
        Dict[int, Dict[int, List[Dict[str, Optional[str]]]]]: Aggregated parsed data.
    """
    # Create all combinations of PON and ONU groups
    all_combinations = [
        (pon, onu_group)
        for pon in CONFIG['pon_range']
        for onu_group in CONFIG['onu_group_range']
    ]

    # Process in batches
    aggregated_data: Dict[int, Dict[int, List[Dict[str, Optional[str]]]]] = {}
    for i in range(0, len(all_combinations), CONFIG['batch_size']):
        batch = all_combinations[i:i + CONFIG['batch_size']]
        logger.info(f"Processing batch {i // CONFIG['batch_size'] + 1}")

        results = await process_batch(scraper, batch)

        for result in results:
            if isinstance(result, Exception):
                logger.error(f"Batch task failed with exception: {result}")
                continue

            for pon, onu_data in result.items():
                if pon not in aggregated_data:
                    aggregated_data[pon] = {}
                aggregated_data[pon].update(onu_data)

    return aggregated_data


async def handle_data(request: web.Request) -> web.StreamResponse:
    """
    HTTP handler for the configured endpoint.
    Triggers data collection and streams the aggregated JSON data incrementally using SSE.

    Args:
        request (web.Request): The incoming HTTP request.

    Returns:
        web.StreamResponse: The SSE response streaming the data.
    """
    start_time = datetime.now()
    logger.info(f"Received {CONFIG['endpoint']} request at {start_time}")

    # Initialize the SSE response
    resp = web.StreamResponse(
        status=200,
        reason='OK',
        headers={
            'Content-Type': 'text/event-stream',
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive',
        }
    )
    await resp.prepare(request)

    try:
        async with AsyncScraper(CONFIG['base_url']) as scraper:
            if not await scraper.login(CONFIG['username'], CONFIG['password']):
                error_data = json.dumps({"error": "Login failed. Please check credentials."})
                await resp.write(f"event: error\ndata: {error_data}\n\n".encode('utf-8'))
                await resp.write_eof()
                return resp

            aggregated_data = await collect_data(scraper)

            end_time = datetime.now()
            duration = end_time - start_time
            logger.info(f"Data collection complete. Duration: {duration}")

            # Send the aggregated data as an SSE event
            json_data = json.dumps(aggregated_data)
            sse_event = f"event: complete\ndata: {json_data}\n\n"
            await resp.write(sse_event.encode('utf-8'))
            await resp.write_eof()

    except ClientError as e:
        logger.error(f"Client error in handle_data: {e}")
        error_event = f"event: error\ndata: {{\"error\": \"Client error: {e}\"}}\n\n"
        await resp.write(error_event.encode('utf-8'))
        await resp.write_eof()
    except asyncio.CancelledError:
        logger.warning("Request was cancelled.")
        await resp.write_eof()
    except Exception as e:
        logger.error(f"Unexpected error in handle_data: {e}")
        error_event = f"event: error\ndata: {{\"error\": \"Unexpected error: {e}\"}}\n\n"
        await resp.write(error_event.encode('utf-8'))
        await resp.write_eof()

    return resp


async def init_app() -> web.Application:
    """
    Initializes the aiohttp web application.

    Returns:
        web.Application: The configured aiohttp web application.
    """
    app = web.Application()
    app.router.add_get(CONFIG['endpoint'], handle_data)  # Use configurable endpoint
    return app


def parse_args() -> argparse.Namespace:
    """
    Parses command-line arguments.

    Returns:
        argparse.Namespace: The parsed command-line arguments.
    """
    parser = argparse.ArgumentParser(description="Async Scraper Server")

    parser.add_argument(
        '--ignore-row-length-warnings',
        action='store_true',
        help='Ignore warnings about row length mismatches in HTML tables.'
    )

    # You can add more CLI parameters here if needed.

    return parser.parse_args()


async def main():
    """
    Main entry point to run the web server with graceful shutdown.
    """
    # Parse CLI arguments
    args = parse_args()

    # Update CONFIG based on CLI arguments
    CONFIG['ignore_row_length_warning'] = args.ignore_row_length_warnings

    # Retrieve all IP addresses
    local_ips = get_all_ip_addresses()
    if not local_ips:
        logger.warning("No non-loopback IPv4 addresses found. Falling back to 127.0.0.1")
        local_ips = [CONFIG['host']]

    endpoint = CONFIG.get('endpoint', '/data')

    logger.info(f"Starting server on port {CONFIG['port']}")
    logger.info("Accessible at the following 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("Server is now running")

    # Create an event to wait for shutdown
    shutdown_event = asyncio.Event()

    # Define signal handler
    def shutdown_signal_handler():
        logger.info("Shutdown signal received")
        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:
            # Signal handlers are not implemented on some platforms (e.g., Windows)
            logger.warning(f"Signal handlers not implemented for {sig}")

    # Wait until a shutdown signal is received
    await shutdown_event.wait()

    logger.info("Shutting down server...")

    # Perform shutdown steps
    await runner.shutdown()
    await runner.cleanup()

    logger.info("Server shut down successfully.")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        logger.critical(f"Unhandled exception: {e}")
        sys.exit(1)
ccq-admin@sm-ccqsv001:~$