DDNS Monitor Dashboard - SamCloud Network

#!/usr/bin/env python3 """ Simple web dashboard for DDNS monitoring status Provides a basic web interface to view current status of all monitored domains. """ import json import os import yaml from datetime import datetime, timedelta from flask import Flask, render_template_string, jsonify from pathlib import Path app = Flask(__name__) # HTML template for the dashboard DASHBOARD_TEMPLATE = """ DDNS Monitor Dashboard - SamCloud Network

📊 Summary

Healthy Domains

Warnings

Errors

Total Domains

🏠 Domain Status

{% for domain in domains %}

🌍 Public IP: {{ domain.public_ip or 'Unknown' }}

🔍 DNS IP: {{ domain.dns_ip or 'Unknown' }}

{% if domain.error %}

❌ Error: {{ domain.error }}

{% endif %} {% if domain.last_update %}

⏰ Last Update: {{ domain.last_update }}

{% endif %}

Last checked: {{ domain.last_checked }}

{% endfor %}

Last updated: {{ last_update }}

Auto-refresh in 60 seconds

""" class DDNSDashboard: def __init__(self, config_path: str = "/usr/local/ddns-monitor/config.yaml"): self.config_path = config_path self.status_file = "/var/log/ddns_monitor_status.json" self.ensure_status_file() def ensure_status_file(self): """Ensure status file exists with current data""" if not os.path.exists(self.status_file): try: # Run a quick check to generate status import subprocess result = subprocess.run([ 'python3', '/usr/local/ddns-monitor/monitor.py', '--test' ], capture_output=True, text=True, timeout=30) if result.returncode == 0: # Extract JSON from output lines = result.stdout.strip().split('\n') json_start = -1 for i, line in enumerate(lines): if line.strip().startswith('['): json_start = i break if json_start >= 0: json_content = '\n'.join(lines[json_start:]) try: data = json.loads(json_content) with open(self.status_file, 'w') as f: json.dump({ 'timestamp': datetime.now().isoformat(), 'domains': {item['domain']: item for item in data} }, f, indent=2) except json.JSONDecodeError: pass except Exception: pass def load_config(self): """Load configuration""" try: with open(self.config_path, 'r') as f: return yaml.safe_load(f) except Exception: return {'domains': []} def load_status(self): """Load latest status from monitoring""" try: if os.path.exists(self.status_file): with open(self.status_file, 'r') as f: return json.load(f) except Exception: pass return {} def get_domain_status(self): """Get formatted status for all domains""" config = self.load_config() status = self.load_status() domains = [] healthy_count = 0 warning_count = 0 error_count = 0 for domain_config in config.get('domains', []): domain_name = domain_config['name'] domain_status = status.get('domains', {}).get(domain_name, {}) # Determine status error = domain_status.get('error') match = domain_status.get('match', True) public_ip = domain_status.get('public_ip') dns_ip = domain_status.get('dns_ip') last_checked = domain_status.get('timestamp', 'Never') if error: css_class = 'error' status_text = 'Error' error_count += 1 elif not match: css_class = 'warning' status_text = 'IP Mismatch' warning_count += 1 else: css_class = 'healthy' status_text = 'Healthy' healthy_count += 1 # Format last checked time try: if last_checked != 'Never': dt = datetime.fromisoformat(last_checked.replace('Z', '+00:00')) last_checked = dt.strftime('%Y-%m-%d %H:%M:%S') except: pass domains.append({ 'name': domain_name, 'css_class': css_class, 'status_text': status_text, 'public_ip': public_ip, 'dns_ip': dns_ip, 'error': error, 'last_checked': last_checked, 'last_update': domain_status.get('last_update') }) return { 'domains': domains, 'healthy_count': healthy_count, 'warning_count': warning_count, 'error_count': error_count, 'total_count': len(domains), 'last_update': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } dashboard = DDNSDashboard() @app.route('/') def index(): """Main dashboard page""" data = dashboard.get_domain_status() return render_template_string(DASHBOARD_TEMPLATE, **data) @app.route('/api/status') def api_status(): """API endpoint for status data""" return jsonify(dashboard.get_domain_status()) @app.route('/health') def health(): """Health check endpoint""" return {'status': 'ok', 'timestamp': datetime.now().isoformat()} if __name__ == '__main__': import argparse parser = argparse.ArgumentParser(description='DDNS Monitor Dashboard') parser.add_argument('--host', default='0.0.0.0', help='Host to bind to') parser.add_argument('--port', type=int, default=54821, help='Port to bind to') parser.add_argument('--config', help='Config file path') args = parser.parse_args() if args.config: dashboard.config_path = args.config print(f"🌐 Starting DDNS Dashboard on http://{args.host}:{args.port}") app.run(host=args.host, port=args.port, debug=False)

🌐 DDNS Monitor

📊 Summary

{{ healthy_count }}

{{ warning_count }}

{{ error_count }}

{{ total_count }}

🏠 Domain Status