rate this code, 1 to 10
1. main.py

import argparse
import asyncio
import logging
import signal
import sys

import aiohttp
import colorlog
from aiohttp import web

from config import CONFIG, load_config
from handlers import handle_data
from utils import get_all_ip_addresses

# 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': 'bold_red',
    }
))
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 parse_args() -> argparse.Namespace:
    """
    Parses command-line arguments.
    """
    parser = argparse.ArgumentParser(description="Async Scraper Server")

    parser.add_argument(
        '--config-file',
        type=str,
        default='config.yaml',
        help='Path to the YAML configuration file.'
    )

    # You can add more CLI parameters here if needed.

    return parser.parse_args()


async def init_app() -> web.Application:
    """
    Initializes the aiohttp web application.
    """
    app = web.Application()
    app.router.add_get(CONFIG['endpoint'], handle_data)  # Use configurable endpoint
    return app


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

    # Load configuration from YAML file
    load_config(args.config_file)

    # 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:
        logger.exception("Unhandled exception occurred")
        sys.exit(1)
2. config.py

import ipaddress
import logging
import sys

import aiohttp
import yaml
from cerberus import Validator

from utils import parse_range_string

logger = logging.getLogger(__name__)

CONFIG = {}


def load_config(config_file: str = 'config.yaml'):
    """
    Load configuration settings from a YAML file with validation using Cerberus.
    """
    try:
        with open(config_file, 'r') as f:
            config = yaml.safe_load(f)

        # Define configuration schema for validation
        schema = {
            'base_url': {'type': 'string', 'required': True, 'regex': r'^https?://.+'},
            'username': {'type': 'string', 'required': True},
            'password': {'type': 'string', 'required': True},
            'who': {'type': 'string', 'required': True},
            'onuid': {'type': 'string', 'required': True},
            'endpoint': {'type': 'string', 'required': True},
            'post_endpoint': {'type': 'string', 'required': True},
            'host': {'type': 'string', 'required': True},
            'port': {'type': 'integer', 'required': True, 'min': 1, 'max': 65535},
            'pon_range': {'type': 'string', 'required': True},
            'onu_group_range': {'type': 'string', 'required': True},
            'max_concurrent_requests': {'type': 'integer', 'required': True, 'min': 1},
            'max_retries': {'type': 'integer', 'required': True, 'min': 1},
            'retry_delay': {'type': 'number', 'required': True, 'min': 0},
            'batch_size': {'type': 'integer', 'required': True, 'min': 1},
            'request_timeout': {
                'type': 'dict',
                'required': True,
                'schema': {
                    'total': {'type': 'number', 'required': True, 'min': 0},
                    'connect': {'type': 'number', 'required': True, 'min': 0},
                    'sock_read': {'type': 'number', 'required': True, 'min': 0},
                }
            },
            'ignore_row_length_warning': {'type': 'boolean', 'required': False, 'default': False},
        }

        validator = Validator(schema)
        if not validator.validate(config):
            errors = validator.errors
            for field, error in errors.items():
                logger.error(f"Configuration error in '{field}': {error}")
            sys.exit(1)

        # Validate 'host' as a valid IP address or hostname
        host = config.get('host')
        try:
            ipaddress.ip_address(host)
        except ValueError:
            # If not a valid IP, check if it is a valid hostname
            if not host.replace('.', '').replace('-', '').isalnum():
                logger.error(f"Invalid 'host': '{host}' is not a valid IP address or hostname")
                sys.exit(1)

        # Parse 'pon_range' and 'onu_group_range'
        try:
            config['pon_range'] = parse_range_string(config['pon_range'])
        except ValueError as e:
            logger.error(f"Invalid 'pon_range': {e}")
            sys.exit(1)

        try:
            config['onu_group_range'] = parse_range_string(config['onu_group_range'])
        except ValueError as e:
            logger.error(f"Invalid 'onu_group_range': {e}")
            sys.exit(1)

        if not config['pon_range']:
            logger.error("'pon_range' cannot be empty")
            sys.exit(1)
        if not config['onu_group_range']:
            logger.error("'onu_group_range' cannot be empty")
            sys.exit(1)

        # Set the request timeouts
        timeout_config = config.get('request_timeout')
        config['request_timeout'] = aiohttp.ClientTimeout(
            total=timeout_config.get('total'),
            connect=timeout_config.get('connect'),
            sock_read=timeout_config.get('sock_read')
        )

        # Set default value if 'ignore_row_length_warning' is missing
        config.setdefault('ignore_row_length_warning', False)

        # Update the global CONFIG dictionary
        global CONFIG
        CONFIG.update(config)

    except FileNotFoundError:
        logger.error(f"Configuration file '{config_file}' not found.")
        sys.exit(1)
    except yaml.YAMLError as e:
        logger.error(f"Error parsing configuration file: {e}")
        sys.exit(1)
    except Exception:
        logger.exception("Unexpected error loading configuration")
        sys.exit(1)
3. scraper.py

import asyncio
import logging
import ssl
from typing import Dict, List, Optional

import aiohttp
from aiohttp import ClientConnectorError, ClientResponseError

from config import CONFIG

logger = logging.getLogger(__name__)


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

    def __init__(self, base_url: str):
        """
        Initialize the AsyncScraper with the given base URL and configuration.
        """
        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.
        """
        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.
        """
        if self.session:
            await self.session.close()

    async def login(self, username: str, password: str) -> bool:
        """
        Perform login to the router.
        """
        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 ClientResponseError as e:
                logger.error(f"HTTP error during login attempt {attempt + 1}: {e}")
            except 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:
                logger.exception(f"Unexpected error during login attempt {attempt + 1}")

            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.
        """
        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 ClientResponseError as e:
                    logger.warning(
                        f"HTTP error {e.status} for PON {pon}, ONU group {onu_group} "
                        f"(attempt {attempt + 1}): {e}"
                    )
                except 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:
                    logger.exception(
                        f"Unexpected error fetching PON {pon}, ONU group {onu_group} "
                        f"(attempt {attempt + 1})"
                    )

                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
4. utils.py

import logging
from typing import List

import netifaces

logger = logging.getLogger(__name__)


def parse_range_string(range_str: str) -> List[int]:
    """
    Parses a range string into a list of integers.
    """
    numbers = set()
    parts = range_str.split(',')
    for part in parts:
        part = part.strip()
        if '-' in part:
            try:
                start_str, end_str = part.split('-', 1)
                start = int(start_str)
                end = int(end_str)
                if start > end:
                    raise ValueError(f"Range start ({start}) must not be greater than end ({end}) in '{part}'")
                numbers.update(range(start, end + 1))
            except ValueError as ve:
                raise ValueError(f"Invalid range '{part}': {ve}")
        else:
            try:
                number = int(part)
                numbers.add(number)
            except ValueError as ve:
                raise ValueError(f"Invalid number '{part}': {ve}")
    return sorted(numbers)


def get_all_ip_addresses() -> List[str]:
    """
    Retrieve all non-loopback IPv4 addresses from the machine's network interfaces.
    """
    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:
        logger.exception("Unexpected error retrieving IP addresses")
    return ips


def format_time(ms: int) -> str:
    """
    Convert milliseconds to a formatted string (minutes:seconds:milliseconds).
    """
    minutes, ms = divmod(ms, 60000)
    seconds, ms = divmod(ms, 1000)
    return f"{minutes}m:{seconds}s:{ms}ms"
5. handlers.py

import asyncio
import json
import logging
import time
from datetime import datetime

from aiohttp import ClientError, web

from config import CONFIG
from scraper import AsyncScraper
from utils import format_time
from worker import collect_data

logger = logging.getLogger(__name__)


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.
    """
    start_time = time.time()
    logger.info(f"Received {CONFIG['endpoint']} request at {datetime.now()}")

    # Initialize the chunked JSON response
    resp = web.StreamResponse(
        status=200,
        reason='OK',
        headers={
            'Content-Type': 'application/json',
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive',
            'Transfer-Encoding': 'chunked'
        }
    )
    await resp.prepare(request)

    # Start JSON object with "data" key
    await resp.write(b'{"data":[')
    first_item = True

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

            async for batch_data in collect_data(scraper):
                for entry in batch_data:
                    json_entry = json.dumps(entry)

                    if not first_item:
                        await resp.write(b',')
                    else:
                        first_item = False

                    await resp.write(json_entry.encode('utf-8'))

        # End JSON array
        await resp.write(b']')

        # Calculate total processing time in milliseconds
        end_time = time.time()
        total_time_ms = int((end_time - start_time) * 1000)

        # Format the processing time for logging
        formatted_time = format_time(total_time_ms)

        # Add "time" key to JSON response with milliseconds
        time_json = f', "time": {total_time_ms}'.encode('utf-8')
        await resp.write(time_json)

        # Close JSON object
        await resp.write(b'}')

        logger.info(f"Data collection complete. Total duration: {formatted_time}")

        await resp.write_eof()

    except ClientError as e:
        logger.error(f"Client error in handle_data: {e}")
        logger.exception(e)
        error_obj = {"error": f"Client error: {str(e)}"}
        if not first_item:
            await resp.write(b',')
        else:
            first_item = False
        await resp.write(json.dumps(error_obj).encode('utf-8'))
        await resp.write(b'],"time":0}')
        await resp.write_eof()
    except asyncio.CancelledError:
        logger.warning("Request was cancelled.")
        await resp.write(b'],"time":0}')
        await resp.write_eof()
    except Exception:
        logger.exception("Unexpected error in handle_data")
        error_obj = {"error": "Unexpected server error occurred."}
        if not first_item:
            await resp.write(b',')
        else:
            first_item = False
        await resp.write(json.dumps(error_obj).encode('utf-8'))
        await resp.write(b'],"time":0}')
        await resp.write_eof()

    return resp
6. worker.py

import asyncio
import logging
from typing import Dict, List, Optional, Tuple

from bs4 import BeautifulSoup

from config import CONFIG
from scraper import AsyncScraper

logger = logging.getLogger(__name__)


async def fetch_and_parse(scraper: AsyncScraper, pon: int, onu_group: int) -> List[Dict[str, Optional[str]]]:
    """
    Fetch and parse optical information for a specific PON and ONU group.
    """
    try:
        html_content = await scraper.fetch_pon_onu_optical_info(pon, onu_group)
        if html_content:
            parsed_data = parse_html_table(html_content)
            if parsed_data:
                return parsed_data
            else:
                logger.warning(f"No data parsed for PON {pon}, ONU group {onu_group}")
                return []
        else:
            logger.warning(f"No content fetched for PON {pon}, ONU group {onu_group}")
            return []
    except Exception:
        logger.exception(f"Error processing PON {pon}, ONU group {onu_group}")
        return []


def parse_html_table(html_content: str) -> List[Dict[str, Optional[str]]]:
    """
    Parse an HTML table and extract data into a list of dictionaries.
    """
    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}")

    # Define the field mapping
    field_mapping = {
        "ONU ID": "onu_id",
        "Desc": "name",
        "RX Power(ONU)": "rx_power_onu",
        "TX Power(ONU)": "tx_power_onu",
        "RX Power(OLT)": "rx_power_olt",
        "TX Power": "tx_power_onu",
        "RX Power": "rx_power_onu",
    }

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

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

        # Rename fields according to the mapping
        row_data = {
            field_mapping.get(key, key): value
            for key, value in raw_row_data.items()
        }

        data.append(row_data)

    return data


async def process_batch(scraper: AsyncScraper, batch: Tuple[int, int]) -> List[Dict[str, Optional[str]]]:
    """
    Process a single PON and ONU group combination.
    """
    pon, onu_group = batch
    result = await fetch_and_parse(scraper, pon, onu_group)
    return result


async def collect_data(scraper: AsyncScraper):
    """
    Collect data by fetching and parsing information for all PON and ONU group combinations.
    This is an asynchronous generator that yields each batch of 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
    semaphore = asyncio.Semaphore(CONFIG['max_concurrent_requests'])

    async def sem_task(batch):
        async with semaphore:
            return await process_batch(scraper, batch)

    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}")

        # Create tasks for the current batch
        tasks = [asyncio.create_task(sem_task(b)) for b in batch]
        # Wait for all tasks in the batch to complete
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Flatten the list and handle exceptions
        batch_data = []
        for result, (pon, onu_group) in zip(results, batch):
            if isinstance(result, Exception):
                logger.error(
                    f"Batch task failed for PON {pon}, ONU group {onu_group}"
                )
                logger.exception(result)
                batch_data.append({
                    "error": f"PON {pon}, ONU group {onu_group} processing failed: {str(result)}"
                })
            elif result:
                batch_data.extend(result)

        if batch_data:
            yield batch_data
7. requirements.txt
List of required packages:


aiohttp
colorlog
netifaces
PyYAML
beautifulsoup4
cerberus