Implement graceful shutdown in this script

import asyncio
import aiohttp
from aiohttp import web
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 # Newly added
import argparse # Newly added
import sys # Newly added

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)

Function to retrieve all IPv4 addresses
def get_all_ip_addresses() -> List[str]:
ips = []
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)
return ips

class AsyncScraper:
def init(self, base_url: str):
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 = None
self.sem = asyncio.Semaphore(CONFIG['max_concurrent_requests'])


async def __aenter__(self):
    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):
    if self.session:
        await self.session.close()

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': '100',
    }

    for attempt in range(CONFIG['max_retries']):
        try:
            async with self.session.post(login_url, data=login_data) as response:
                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 Exception as e:
            logger.error(f"Login attempt {attempt + 1} failed: {e}")
            if attempt < CONFIG['max_retries'] - 1:
                await asyncio.sleep(CONFIG['retry_delay'])
            else:
                return False

async def fetch_pon_onu_optical_info(self, pon: int, onu_group: int) -> Optional[str]:
    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:
                    if response.status == 200:
                        content = await response.text()
                        logger.info(f"Successfully fetched data for PON {pon}, ONU Group {onu_group} (Attempt {attempt + 1})")
                        return content
                    else:
                        logger.warning(f"Error {response.status} for PON {pon}, ONU Group {onu_group} (Attempt {attempt + 1})")
            except asyncio.TimeoutError:
                logger.warning(f"Timeout fetching PON {pon}, ONU Group {onu_group} (Attempt {attempt + 1})")
            except Exception as e:
                logger.warning(f"Error fetching PON {pon}, ONU Group {onu_group} (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
def parse_html_table(html_content: str) -> List[Dict[str, Optional[str]]]:
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]]]]]:
html_content = await scraper.fetch_pon_onu_optical_info(pon, onu_group)
parsed_data = 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]:
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]]]]]:
# 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):
"""
HTTP handler for the configured endpoint.
Triggers data collection and streams the aggregated JSON data incrementally using SSE.
"""
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

        # 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']
        ]

        aggregated_data: Dict[int, Dict[int, List[Dict[str, Optional[str]]]]] = {}

        # Process in batches
        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)

            batch_data = {}
            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)

                    if pon not in batch_data:
                        batch_data[pon] = {}
                    batch_data[pon].update(onu_data)

            # Prepare the batch data to send
            json_part = json.dumps(batch_data)

            # Send the batch data as an SSE event
            sse_event = f"event: batch\ndata: {json_part}\n\n"
            await resp.write(sse_event.encode('utf-8'))
            # Removed the deprecated drain call

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

        # Send a final completion event
        final_event = f"event: complete\ndata: {{\"status\": \"complete\", \"duration\": \"{duration}\"}}\n\n"
        await resp.write(final_event.encode('utf-8'))
        await resp.write_eof()

except Exception as e:
    logger.error(f"Error in handle_data: {e}")
    error_event = f"event: error\ndata: {{\"error\": \"{str(e)}\"}}\n\n"
    await resp.write(error_event.encode('utf-8'))
    await resp.write_eof()

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

def parse_args():
"""
Parses 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()
def main():
"""
Main entry point to run the web server.
"""
# 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}")

loop = asyncio.get_event_loop()
app = loop.run_until_complete(init_app())

runner = web.AppRunner(app)
loop.run_until_complete(runner.setup())

site = web.TCPSite(runner, host=CONFIG['host'], port=CONFIG['port'])
loop.run_until_complete(site.start())

logger.info("Server is now running")

try:
    loop.run_forever()
except KeyboardInterrupt:
    logger.info("Shutting down server...")
finally:
    loop.run_until_complete(runner.cleanup())
if name == "main":
main()