rate this script, 1 to 10

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)
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
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
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
requirements.txt
List of required packages:
aiohttp
colorlog
netifaces
PyYAML
beautifulsoup4
cerberus