Output
ccq-admin@sm-ccqsv001:~$ python final3.py
2024-11-10 18:47:24 [INFO] Starting server on http://127.0.1.1:8080/data
2024-11-10 18:47:24 [INFO] Server running on http://127.0.1.1:8080/data
It’s good that socks is imported to determine which IP the server could listen on. However, since I'm still using 0.0.0.0, I also need to know all the IPs on which the server can listen, not just one

Additionally, I want to be able to change the URL of my endpoint in the configuration. For example, from ip:8080/data to ip:/poninfo, so it should be possible to change that.

here is the code

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

# 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
}

# 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)

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):
            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 /data endpoint.
    Triggers data collection and streams the aggregated JSON data incrementally using SSE.
    """
    start_time = datetime.now()
    logger.info(f"Received /data 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']):
                await resp.write(b"event: error\ndata: {\"error\": \"Login failed. Please check credentials.\"}\n\n")
                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'))
                await resp.drain()  # Ensure the data is sent to the client

            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('/data', handle_data)
    return app

def main():
    """
    Main entry point to run the web server.
    """
    # Retrieve the machine's IP address
    hostname = socket.gethostname()
    try:
        local_ip = socket.gethostbyname(hostname)
    except socket.gaierror:
        local_ip = CONFIG['host']  # Fallback to configured host

    logger.info(f"Starting server on http://{local_ip}:{CONFIG['port']}/data")

    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(f"Server running on http://{local_ip}:{CONFIG['port']}/data")
    
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        logger.info("Shutting down server...")
    finally:
        loop.run_until_complete(runner.cleanup())

if __name__ == "__main__":
    main()