#!/usr/bin/env python3 import json import os import socket import sys import time import urllib.parse from datetime import datetime from http.server import HTTPServer, BaseHTTPRequestHandler from socketserver import ThreadingMixIn class ThreadedHTTPServer(ThreadingMixIn, HTTPServer): """Handle requests in a separate thread.""" daemon_threads = True class ControlPanelHandler(BaseHTTPRequestHandler): def __init__(self, *args, **kwargs): self.data_dir = "/home/wdjones/.openclaw/workspace/projects/control-panel/data" super().__init__(*args, **kwargs) def do_GET(self): try: self.handle_get() except Exception as e: self.send_error(500, f"Internal error: {e}") def do_POST(self): try: self.handle_post() except Exception as e: self.send_error(500, f"Internal error: {e}") def handle_get(self): if self.path == '/': self.serve_dashboard() elif self.path == '/accounts': self.serve_accounts() elif self.path == '/api-keys': self.serve_api_keys() elif self.path == '/services': self.serve_services() elif self.path == '/budget': self.serve_budget() elif self.path == '/activity': self.serve_activity() else: self.send_error(404, "Not found") def handle_post(self): content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length).decode('utf-8') form_data = urllib.parse.parse_qs(post_data) if self.path == '/accounts': self.handle_accounts_post(form_data) elif self.path == '/api-keys': self.handle_api_keys_post(form_data) elif self.path == '/budget': self.handle_budget_post(form_data) else: self.send_error(404, "Not found") def load_data(self, filename): filepath = os.path.join(self.data_dir, filename) if os.path.exists(filepath): with open(filepath, 'r') as f: return json.load(f) return [] def save_data(self, filename, data): os.makedirs(self.data_dir, exist_ok=True) filepath = os.path.join(self.data_dir, filename) with open(filepath, 'w') as f: json.dump(data, f, indent=2) def log_activity(self, action, details=""): activity = self.load_data('activity.json') entry = { "timestamp": datetime.now().isoformat(), "action": action, "details": details } activity.insert(0, entry) # Latest first activity = activity[:100] # Keep last 100 entries self.save_data('activity.json', activity) def check_service_health(self, port): try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(2) result = sock.connect_ex(('localhost', port)) sock.close() return result == 0 except: return False def get_base_template(self, title, content): return f""" {title} - Case Control Panel
{content}
""" def serve_dashboard(self): accounts = self.load_data('accounts.json') api_keys = self.load_data('api-keys.json') budget = self.load_data('budget.json') # Calculate stats total_accounts = len(accounts) active_accounts = len([a for a in accounts if a.get('status') == 'active']) total_api_keys = len(api_keys) monthly_spend = sum([b.get('amount', 0) for b in budget if b.get('type') == 'spending' and b.get('timestamp', '').startswith(datetime.now().strftime('%Y-%m'))]) content = f"""
{total_accounts}
Total Accounts
{active_accounts}
Active Services
{total_api_keys}
API Keys
${monthly_spend:.2f}
Monthly Spend
Quick Actions
Manage Accounts Manage API Keys Add Budget Entry Check Services
""" html = self.get_base_template("Dashboard", content) self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write(html.encode()) def serve_accounts(self): accounts = self.load_data('accounts.json') accounts_table = "" for i, account in enumerate(accounts): status_class = "status-active" if account.get('status') == 'active' else "status-inactive" login_btn = f'Login' if account.get('url') else "" accounts_table += f""" {account.get('service', 'N/A')} {account.get('url', 'N/A')} {account.get('username', 'N/A')} {account.get('status', 'unknown')} {account.get('last_accessed', 'Never')} {account.get('notes', '')} {login_btn} """ content = f"""
Account Management
{accounts_table}
Service URL Username/Email Status Last Accessed Notes Actions
""" html = self.get_base_template("Accounts", content) self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write(html.encode()) def serve_api_keys(self): api_keys = self.load_data('api-keys.json') keys_table = "" for key in api_keys: masked_key = f'' + \ ('*' * len(key.get('key', ''))) + '' keys_table += f""" {key.get('service', 'N/A')} {key.get('name', 'N/A')} {masked_key} {key.get('created', 'N/A')} {key.get('expires', 'Never')} {key.get('usage', 0)} """ content = f"""
API Key Management
{keys_table}
Service Name Key (click to reveal) Created Expires Usage Count
""" html = self.get_base_template("API Keys", content) self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write(html.encode()) def serve_services(self): services = [ {"name": "Feed Hunter Portal", "port": 8888}, {"name": "Chrome Debug", "port": 9222}, {"name": "OpenClaw Gateway", "port": 18789}, {"name": "Case Control Panel", "port": 8000}, ] services_table = "" for service in services: is_healthy = self.check_service_health(service["port"]) status = "Running" if is_healthy else "Stopped" status_class = "status-active" if is_healthy else "status-inactive" services_table += f""" {service['name']} {service['port']} {status} N/A """ content = f"""
Running Services
{services_table}
Service Name Port Status Uptime
""" html = self.get_base_template("Services", content) self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write(html.encode()) def serve_budget(self): budget = self.load_data('budget.json') # Calculate totals total_balance = sum([b.get('amount', 0) for b in budget if b.get('type') == 'deposit']) - \ sum([b.get('amount', 0) for b in budget if b.get('type') in ['withdrawal', 'spending']]) current_month = datetime.now().strftime('%Y-%m') monthly_spending = sum([b.get('amount', 0) for b in budget if b.get('type') == 'spending' and b.get('timestamp', '').startswith(current_month)]) budget_table = "" for entry in sorted(budget, key=lambda x: x.get('timestamp', ''), reverse=True)[:50]: amount_str = f"${entry.get('amount', 0):.2f}" if entry.get('type') == 'deposit': amount_str = f"+{amount_str}" elif entry.get('type') in ['withdrawal', 'spending']: amount_str = f"-{amount_str}" budget_table += f""" {entry.get('timestamp', 'N/A')} {entry.get('type', 'N/A')} {entry.get('service', 'General')} {amount_str} {entry.get('description', '')} """ content = f"""
${total_balance:.2f}
Total Balance
${monthly_spending:.2f}
Monthly Spending
Budget Management
{budget_table}
Date Type Service Amount Description
""" html = self.get_base_template("Budget", content) self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write(html.encode()) def serve_activity(self): activity = self.load_data('activity.json') activity_list = "" for entry in activity: activity_list += f"""
{entry.get('timestamp', 'N/A')}
{entry.get('action', 'N/A')}
{entry.get('details', '')}
""" content = f"""
Activity Log
{activity_list if activity_list else '

No activity recorded yet.

'}
""" html = self.get_base_template("Activity", content) self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write(html.encode()) def handle_accounts_post(self, form_data): if form_data.get('action', [''])[0] == 'add': accounts = self.load_data('accounts.json') new_account = { "service": form_data.get('service', [''])[0], "url": form_data.get('url', [''])[0], "username": form_data.get('username', [''])[0], "status": form_data.get('status', ['active'])[0], "notes": form_data.get('notes', [''])[0], "created": datetime.now().isoformat(), "last_accessed": "Never" } accounts.append(new_account) self.save_data('accounts.json', accounts) self.log_activity("Account Added", f"Added {new_account['service']}") # Redirect back to accounts page self.send_response(302) self.send_header('Location', '/accounts') self.end_headers() def handle_api_keys_post(self, form_data): if form_data.get('action', [''])[0] == 'add': api_keys = self.load_data('api-keys.json') new_key = { "service": form_data.get('service', [''])[0], "name": form_data.get('name', [''])[0], "key": form_data.get('key', [''])[0], "created": datetime.now().strftime('%Y-%m-%d'), "expires": form_data.get('expires', ['Never'])[0] or "Never", "usage": 0 } api_keys.append(new_key) self.save_data('api-keys.json', api_keys) self.log_activity("API Key Added", f"Added key for {new_key['service']}") # Redirect back to api-keys page self.send_response(302) self.send_header('Location', '/api-keys') self.end_headers() def handle_budget_post(self, form_data): if form_data.get('action', [''])[0] == 'add': budget = self.load_data('budget.json') new_entry = { "type": form_data.get('type', [''])[0], "service": form_data.get('service', ['General'])[0] or "General", "amount": float(form_data.get('amount', ['0'])[0]), "description": form_data.get('description', [''])[0], "timestamp": datetime.now().isoformat() } budget.append(new_entry) self.save_data('budget.json', budget) self.log_activity("Budget Entry Added", f"{new_entry['type']} of ${new_entry['amount']:.2f}") # Redirect back to budget page self.send_response(302) self.send_header('Location', '/budget') self.end_headers() def log_message(self, format, *args): """Override to reduce logging noise""" pass def initialize_data(): """Pre-populate with known accounts and services""" data_dir = "/home/wdjones/.openclaw/workspace/projects/control-panel/data" os.makedirs(data_dir, exist_ok=True) # Pre-populate accounts accounts_file = os.path.join(data_dir, "accounts.json") if not os.path.exists(accounts_file): initial_accounts = [ { "service": "ProtonMail", "url": "https://mail.proton.me", "username": "case-lgn@protonmail.com", "status": "active", "notes": "Primary email account", "created": datetime.now().isoformat(), "last_accessed": "Never" }, { "service": "Polymarket", "url": "https://polymarket.com", "username": "", "status": "inactive", "notes": "Not yet registered", "created": datetime.now().isoformat(), "last_accessed": "Never" }, { "service": "Feed Hunter Portal", "url": "http://localhost:8888", "username": "", "status": "active", "notes": "Local service", "created": datetime.now().isoformat(), "last_accessed": "Never" }, { "service": "Chrome Debug", "url": "http://localhost:9222", "username": "", "status": "active", "notes": "Browser debugging interface", "created": datetime.now().isoformat(), "last_accessed": "Never" }, { "service": "OpenClaw Gateway", "url": "http://localhost:18789", "username": "", "status": "active", "notes": "OpenClaw main service", "created": datetime.now().isoformat(), "last_accessed": "Never" } ] with open(accounts_file, 'w') as f: json.dump(initial_accounts, f, indent=2) # Initialize empty files if they don't exist for filename in ["api-keys.json", "budget.json", "activity.json"]: filepath = os.path.join(data_dir, filename) if not os.path.exists(filepath): with open(filepath, 'w') as f: json.dump([], f) def main(): initialize_data() server_address = ('0.0.0.0', 8000) httpd = ThreadedHTTPServer(server_address, ControlPanelHandler) print(f"🖤 Case Control Panel starting on http://0.0.0.0:8000") print("Press Ctrl+C to stop") try: httpd.serve_forever() except KeyboardInterrupt: print("\nShutting down...") httpd.shutdown() sys.exit(0) if __name__ == '__main__': main()