#!/usr/bin/env python3 """ ParentZone Snapshots Web Server A simple web server that serves HTML snapshot files and their assets. Provides a directory listing and serves static files from the snapshots folder. """ import os import asyncio import argparse import logging from pathlib import Path from urllib.parse import unquote from datetime import datetime import aiohttp from aiohttp import web, hdrs from aiohttp.web_response import Response class SnapshotsWebServer: def __init__( self, snapshots_dir: str = "./snapshots", port: int = 8080, host: str = "0.0.0.0", ): self.snapshots_dir = Path(snapshots_dir).resolve() self.port = port self.host = host # Setup logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", ) self.logger = logging.getLogger(__name__) # Ensure snapshots directory exists self.snapshots_dir.mkdir(parents=True, exist_ok=True) self.logger.info(f"Serving snapshots from: {self.snapshots_dir}") async def index_handler(self, request): """Serve the main directory listing page.""" try: html_files = [] # Find all HTML files in the snapshots directory for file_path in self.snapshots_dir.glob("*.html"): stat = file_path.stat() html_files.append( { "name": file_path.name, "size": stat.st_size, "modified": datetime.fromtimestamp(stat.st_mtime), "path": file_path.name, } ) # Sort by modification time (newest first) html_files.sort(key=lambda x: x["modified"], reverse=True) # Generate HTML page html_content = self._generate_index_html(html_files) return web.Response(text=html_content, content_type="text/html") except Exception as e: self.logger.error(f"Error generating index: {e}") return web.Response( text=f"

Error

Could not generate directory listing: {e}

", status=500, content_type="text/html", ) def _generate_index_html(self, html_files): """Generate the HTML directory listing page.""" files_list = "" if not html_files: files_list = "

No snapshot files found.

" else: for file_info in html_files: size_mb = file_info["size"] / (1024 * 1024) files_list += f"""

{file_info["name"]}

{size_mb:.2f} MB {file_info["modified"].strftime("%Y-%m-%d %H:%M:%S")}

""" return f""" ParentZone Snapshots

📁 Available Snapshot Files ({len(html_files)} files)

{files_list}

""" async def file_handler(self, request): """Serve individual HTML files and their assets.""" try: # Get the requested file path file_path = unquote(request.match_info["filename"]) requested_file = self.snapshots_dir / file_path # Security check: ensure the file is within the snapshots directory try: requested_file.resolve().relative_to(self.snapshots_dir.resolve()) except ValueError: self.logger.warning(f"Attempted path traversal: {file_path}") return web.Response( text="

403 Forbidden

Access denied.

", status=403, content_type="text/html", ) # Check if file exists if not requested_file.exists(): return web.Response( text="

404 Not Found

The requested file was not found.

", status=404, content_type="text/html", ) # Determine content type content_type = self._get_content_type(requested_file) # Read and serve the file with open(requested_file, "rb") as f: content = f.read() return web.Response( body=content, content_type=content_type, headers={ "Cache-Control": "public, max-age=3600", "Last-Modified": datetime.fromtimestamp( requested_file.stat().st_mtime ).strftime("%a, %d %b %Y %H:%M:%S GMT"), }, ) except Exception as e: self.logger.error( f"Error serving file {request.match_info.get('filename', 'unknown')}: {e}" ) return web.Response( text=f"

500 Internal Server Error

Could not serve file: {e}

", status=500, content_type="text/html", ) async def assets_handler(self, request): """Serve asset files (images, CSS, JS, etc.) from assets subdirectories.""" try: # Get the requested asset path asset_path = unquote(request.match_info["path"]) requested_file = self.snapshots_dir / "assets" / asset_path # Security check try: requested_file.resolve().relative_to(self.snapshots_dir.resolve()) except ValueError: self.logger.warning(f"Attempted path traversal in assets: {asset_path}") return web.Response(text="403 Forbidden", status=403) # Check if file exists if not requested_file.exists(): return web.Response(text="404 Not Found", status=404) # Determine content type content_type = self._get_content_type(requested_file) # Read and serve the file with open(requested_file, "rb") as f: content = f.read() return web.Response( body=content, content_type=content_type, headers={ "Cache-Control": "public, max-age=86400", # Cache assets for 24 hours "Last-Modified": datetime.fromtimestamp( requested_file.stat().st_mtime ).strftime("%a, %d %b %Y %H:%M:%S GMT"), }, ) except Exception as e: self.logger.error( f"Error serving asset {request.match_info.get('path', 'unknown')}: {e}" ) return web.Response(text="500 Internal Server Error", status=500) def _get_content_type(self, file_path: Path) -> str: """Determine the content type based on file extension.""" suffix = file_path.suffix.lower() content_types = { ".html": "text/html; charset=utf-8", ".css": "text/css; charset=utf-8", ".js": "application/javascript; charset=utf-8", ".json": "application/json; charset=utf-8", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".gif": "image/gif", ".webp": "image/webp", ".svg": "image/svg+xml", ".ico": "image/x-icon", ".pdf": "application/pdf", ".txt": "text/plain; charset=utf-8", ".log": "text/plain; charset=utf-8", } return content_types.get(suffix, "application/octet-stream") def setup_routes(self, app): """Configure the web application routes.""" # Main index page app.router.add_get("/", self.index_handler) # Serve HTML files directly app.router.add_get("/{filename:.+\.html}", self.file_handler) # Serve assets (images, CSS, JS, etc.) app.router.add_get("/assets/{path:.+}", self.assets_handler) # Serve other static files (logs, etc.) app.router.add_get( "/{filename:.+\.(css|js|json|txt|log|ico)}", self.file_handler ) async def create_app(self): """Create and configure the web application.""" app = web.Application() # Setup routes self.setup_routes(app) # Add middleware for logging async def logging_middleware(request, handler): start_time = datetime.now() try: response = await handler(request) # Log the request duration = (datetime.now() - start_time).total_seconds() self.logger.info( f"{request.remote} - {request.method} {request.path} - " f"{response.status} - {duration:.3f}s" ) return response except Exception as e: duration = (datetime.now() - start_time).total_seconds() self.logger.error( f"{request.remote} - {request.method} {request.path} - " f"ERROR: {e} - {duration:.3f}s" ) raise app.middlewares.append(logging_middleware) return app async def start_server(self): """Start the web server.""" app = await self.create_app() runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, self.host, self.port) await site.start() self.logger.info(f"🚀 ParentZone Snapshots Web Server started!") self.logger.info(f"📂 Serving files from: {self.snapshots_dir}") self.logger.info(f"🌐 Server running at: http://{self.host}:{self.port}") self.logger.info(f"🔗 Open in browser: http://localhost:{self.port}") self.logger.info("Press Ctrl+C to stop the server") return runner def main(): parser = argparse.ArgumentParser( description="ParentZone Snapshots Web Server", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Start server with default settings python webserver.py # Start server on custom port python webserver.py --port 3000 # Serve from custom directory python webserver.py --snapshots-dir /path/to/snapshots # Start server on all interfaces python webserver.py --host 0.0.0.0 --port 8080 """, ) parser.add_argument( "--snapshots-dir", default="./snapshots", help="Directory containing snapshot files (default: ./snapshots)", ) parser.add_argument( "--port", type=int, default=8080, help="Port to run the server on (default: 8080)", ) parser.add_argument( "--host", default="0.0.0.0", help="Host to bind the server to (default: 0.0.0.0)", ) args = parser.parse_args() # Create and start the server server = SnapshotsWebServer( snapshots_dir=args.snapshots_dir, port=args.port, host=args.host ) async def run_server(): runner = None try: runner = await server.start_server() # Keep the server running while True: await asyncio.sleep(1) except KeyboardInterrupt: print("\n👋 Shutting down server...") except Exception as e: print(f"❌ Server error: {e}") finally: if runner: await runner.cleanup() try: asyncio.run(run_server()) except KeyboardInterrupt: print("\n✅ Server stopped") if __name__ == "__main__": main()

Error

{file_info["name"]}

📸 ParentZone Snapshots

403 Forbidden

404 Not Found

500 Internal Server Error