#!/usr/bin/env python3 """Tiny dashboard server that serves static files and a /api/status endpoint.""" import http.server import socketserver import json import os import subprocess import threading import base64 PORT = 8787 DIR = os.path.dirname(os.path.abspath(__file__)) WORKSPACE = os.path.dirname(DIR) # serve from workspace root STATUS_FILE = os.path.join(DIR, 'status.json') KUHIO_SYNC = os.path.join(WORKSPACE, 'kuhio', 'sync_sheet.py') BINGEN_SYNC = os.path.join(WORKSPACE, 'bingen', 'sync_sheet.py') # HTTP Basic Auth credentials AUTH_USERNAME = 'farro' AUTH_PASSWORD = 'farro2026' # Change this to whatever you prefer class DashboardHandler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=WORKSPACE, **kwargs) def check_auth(self): """Check HTTP Basic Auth. Returns True if authorized.""" auth_header = self.headers.get('Authorization') if not auth_header: return False try: # Parse "Basic base64encodedcreds" auth_type, auth_string = auth_header.split(' ', 1) if auth_type.lower() != 'basic': return False # Decode credentials decoded = base64.b64decode(auth_string).decode('utf-8') username, password = decoded.split(':', 1) return username == AUTH_USERNAME and password == AUTH_PASSWORD except Exception: return False def send_auth_required(self): """Send 401 Unauthorized with auth challenge.""" self.send_response(401) self.send_header('WWW-Authenticate', 'Basic realm="Sprout Dashboard"') self.send_header('Content-Type', 'text/html') self.end_headers() self.wfile.write(b'

401 Unauthorized

Authentication required.

') def do_GET(self): # NO AUTH - completely removed for troubleshooting if self.path == '/api/status': self.handle_status() elif self.path == '/api/sync-kuhio': self.handle_sync_kuhio() elif self.path == '/api/sync-bingen': self.handle_sync_bingen() elif self.path == '/api/kuhio-data': self.handle_kuhio_data() elif self.path == '/api/bingen-data': self.handle_bingen_data() else: super().do_GET() def handle_status(self): try: if os.path.exists(STATUS_FILE): with open(STATUS_FILE, 'r') as f: data = json.load(f) self.send_response(200) else: data = {'ok': False, 'error': 'No status data yet. Waiting for Sprout to update...'} self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(data).encode()) except Exception as e: self.send_response(500) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False, 'error': str(e)}).encode()) def handle_sync_kuhio(self): """Run kuhio sync_sheet.py and return result.""" try: result = subprocess.run( ['python3', KUHIO_SYNC], capture_output=True, text=True, timeout=60, cwd=os.path.join(WORKSPACE, 'kuhio') ) ok = result.returncode == 0 data = { 'ok': ok, 'message': 'Sync complete' if ok else 'Sync failed', 'output': result.stdout[-500:] if result.stdout else '', 'error': result.stderr[-500:] if result.stderr and not ok else '', } self.send_response(200 if ok else 500) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(data).encode()) except subprocess.TimeoutExpired: self.send_response(504) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False, 'error': 'Sync timed out'}).encode()) except Exception as e: self.send_response(500) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False, 'error': str(e)}).encode()) def handle_sync_bingen(self): """Run bingen sync_sheet.py and return result.""" try: result = subprocess.run( ['python3', BINGEN_SYNC], capture_output=True, text=True, timeout=60, cwd=os.path.join(WORKSPACE, 'bingen') ) ok = result.returncode == 0 data = { 'ok': ok, 'message': 'Sync complete' if ok else 'Sync failed', 'output': result.stdout[-500:] if result.stdout else '', 'error': result.stderr[-500:] if result.stderr and not ok else '', } self.send_response(200 if ok else 500) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(data).encode()) except subprocess.TimeoutExpired: self.send_response(504) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False, 'error': 'Sync timed out'}).encode()) except Exception as e: self.send_response(500) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False, 'error': str(e)}).encode()) def handle_kuhio_data(self): """Serve kuhio/data.json with proper authentication.""" try: kuhio_data_file = os.path.join(WORKSPACE, 'kuhio', 'data.json') if os.path.exists(kuhio_data_file): with open(kuhio_data_file, 'r') as f: data = f.read() self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Cache-Control', 'no-cache, no-store, must-revalidate') self.end_headers() self.wfile.write(data.encode()) else: self.send_response(404) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False, 'error': 'Kuhio data not found'}).encode()) except Exception as e: self.send_response(500) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False, 'error': str(e)}).encode()) def handle_bingen_data(self): """Serve bingen/data.json with proper authentication.""" try: bingen_data_file = os.path.join(WORKSPACE, 'bingen', 'data.json') if os.path.exists(bingen_data_file): with open(bingen_data_file, 'r') as f: data = f.read() self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(data.encode()) else: self.send_response(404) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False, 'error': 'Bingen data not found'}).encode()) except Exception as e: self.send_response(500) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'ok': False, 'error': str(e)}).encode()) class ThreadedHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): """Handle requests in separate threads for parallel loading.""" daemon_threads = True if __name__ == '__main__': with ThreadedHTTPServer(('0.0.0.0', PORT), DashboardHandler) as httpd: print(f'Dashboard running on http://localhost:{PORT} (threaded)') httpd.serve_forever()