#!/usr/bin/env python3 """ VSCode Agent Integration Helper for DDNS Monitor Provides structured data access to all ddclient configs and logs for integration with VSCode agents and automation workflows. """ import json import os import subprocess import yaml from datetime import datetime from pathlib import Path from typing import Dict, List, Optional class DDNSAgentHelper: """Helper class for VSCode agents to interact with DDNS system""" def __init__(self): # Your actual ddclient paths self.paths = { 'ddclient_config': '/opt/homebrew/etc/ddclient/ddclient.conf', 'ddclient_binary': '/opt/homebrew/bin/ddclient', 'launch_daemon': '/Library/LaunchDaemons/org.ddclient.plist', 'main_log': '/var/log/ddclient.log', 'error_log': '/var/log/ddclient.err', 'monitor_config': '/usr/local/ddns-monitor/config.yaml', 'monitor_log': '/var/log/ddns_monitor.log', 'backup_config': '/Volumes/SamMgmt/Backups/ddclient.conf.bak', 'backup_plist': '/Volumes/SamMgmt/Backups/ddclient.plist.bak' } def get_system_status(self) -> Dict: """Get comprehensive system status for agents""" status = { 'timestamp': datetime.now().isoformat(), 'ddclient': self._get_ddclient_status(), 'monitor': self._get_monitor_status(), 'files': self._check_file_status(), 'network': self._get_network_status(), 'backups': self._check_backup_status() } return status def _get_ddclient_status(self) -> Dict: """Get ddclient daemon status""" try: # Check if ddclient process is running result = subprocess.run(['pgrep', '-f', 'ddclient'], capture_output=True) running = result.returncode == 0 # Check LaunchDaemon status daemon_result = subprocess.run( ['launchctl', 'list', '|', 'grep', 'ddclient'], shell=True, capture_output=True, text=True ) daemon_loaded = 'ddclient' in daemon_result.stdout # Get last run info from logs last_run = self._parse_ddclient_logs() return { 'process_running': running, 'daemon_loaded': daemon_loaded, 'last_run': last_run, 'config_valid': os.path.exists(self.paths['ddclient_config']) } except Exception as e: return {'error': str(e)} def _get_monitor_status(self) -> Dict: """Get DDNS monitor status""" try: # Check if monitor process is running result = subprocess.run(['pgrep', '-f', 'ddns-monitor'], capture_output=True) running = result.returncode == 0 # Check LaunchDaemon status daemon_result = subprocess.run( ['launchctl', 'list', '|', 'grep', 'ca.samcloud.ddns-monitor'], shell=True, capture_output=True, text=True ) daemon_loaded = 'ddns-monitor' in daemon_result.stdout return { 'process_running': running, 'daemon_loaded': daemon_loaded, 'config_exists': os.path.exists(self.paths['monitor_config']), 'dashboard_port': 54821 } except Exception as e: return {'error': str(e)} def _check_file_status(self) -> Dict: """Check status of all config and log files""" file_status = {} for name, path in self.paths.items(): try: if os.path.exists(path): stat = os.stat(path) file_status[name] = { 'exists': True, 'size': stat.st_size, 'modified': datetime.fromtimestamp(stat.st_mtime).isoformat(), 'readable': os.access(path, os.R_OK), 'writable': os.access(path, os.W_OK) } else: file_status[name] = {'exists': False} except Exception as e: file_status[name] = {'error': str(e)} return file_status def _get_network_status(self) -> Dict: """Get network connectivity status""" try: import requests # Test public IP services ip_services = [ 'https://ipv4.icanhazip.com', 'https://api.ipify.org', 'https://checkip.amazonaws.com' ] public_ip = None working_services = 0 for service in ip_services: try: response = requests.get(service, timeout=5) if response.status_code == 200: working_services += 1 if not public_ip: public_ip = response.text.strip() except: pass return { 'public_ip': public_ip, 'ip_services_working': working_services, 'total_ip_services': len(ip_services), 'connectivity_ok': working_services > 0 } except Exception as e: return {'error': str(e)} def _check_backup_status(self) -> Dict: """Check backup file status""" backup_status = {} backup_files = { 'config_backup': self.paths['backup_config'], 'plist_backup': self.paths['backup_plist'] } for name, path in backup_files.items(): if os.path.exists(path): stat = os.stat(path) backup_status[name] = { 'exists': True, 'size': stat.st_size, 'age_hours': (datetime.now().timestamp() - stat.st_mtime) / 3600 } else: backup_status[name] = {'exists': False} return backup_status def _parse_ddclient_logs(self) -> Dict: """Parse ddclient logs for recent activity""" try: log_file = self.paths['main_log'] error_file = self.paths['error_log'] last_run = {'success': False, 'message': 'No recent activity'} # Check main log if os.path.exists(log_file): with open(log_file, 'r') as f: lines = f.readlines() # Look for recent success/failure for line in reversed(lines[-50:]): if 'success' in line.lower(): last_run = { 'success': True, 'message': line.strip(), 'source': 'main_log' } break elif 'error' in line.lower() or 'fail' in line.lower(): last_run = { 'success': False, 'message': line.strip(), 'source': 'main_log' } break # Check error log for recent issues if os.path.exists(error_file): with open(error_file, 'r') as f: error_lines = f.readlines() if error_lines: last_run['recent_errors'] = len(error_lines[-10:]) return last_run except Exception as e: return {'error': str(e)} def get_domain_list(self) -> List[str]: """Extract domain list from ddclient config""" try: domains = [] config_file = self.paths['ddclient_config'] if os.path.exists(config_file): with open(config_file, 'r') as f: content = f.read() # Simple parsing for domains (this could be enhanced) lines = content.split('\n') for line in lines: line = line.strip() # Look for lines that might be domain names if ('.' in line and not line.startswith('#') and not '=' in line and 'samcloud.ca' in line): domains.append(line) return list(set(domains)) # Remove duplicates except Exception as e: return [f"Error reading config: {e}"] def force_update(self, domain: Optional[str] = None) -> Dict: """Force ddclient update for agent automation""" try: cmd = [self.paths['ddclient_binary'], '-force', '-verbose'] if domain: cmd.extend(['-host', domain]) result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) return { 'success': result.returncode == 0, 'return_code': result.returncode, 'stdout': result.stdout, 'stderr': result.stderr, 'command': ' '.join(cmd) } except Exception as e: return {'success': False, 'error': str(e)} def get_dashboard_url(self) -> str: """Get dashboard URL for agent use""" return f"http://localhost:54821" def export_for_agent(self, output_file: str = "/tmp/ddns_status.json"): """Export all status data for agent consumption""" status = self.get_system_status() status['domains'] = self.get_domain_list() status['dashboard_url'] = self.get_dashboard_url() with open(output_file, 'w') as f: json.dump(status, f, indent=2) return output_file def main(): """CLI interface for agent integration""" import argparse parser = argparse.ArgumentParser(description='DDNS Agent Helper') parser.add_argument('--status', action='store_true', help='Show system status') parser.add_argument('--domains', action='store_true', help='List configured domains') parser.add_argument('--force', help='Force update for specific domain') parser.add_argument('--export', help='Export status to JSON file') parser.add_argument('--dashboard', action='store_true', help='Show dashboard URL') args = parser.parse_args() helper = DDNSAgentHelper() if args.status: status = helper.get_system_status() print(json.dumps(status, indent=2)) elif args.domains: domains = helper.get_domain_list() for domain in domains: print(domain) elif args.force: result = helper.force_update(args.force) print(json.dumps(result, indent=2)) elif args.export: output_file = helper.export_for_agent(args.export) print(f"Status exported to: {output_file}") elif args.dashboard: print(helper.get_dashboard_url()) else: # Default: show summary status = helper.get_system_status() print("🌐 DDNS System Status Summary") print(f"ddclient running: {status['ddclient'].get('process_running', False)}") print(f"Monitor running: {status['monitor'].get('process_running', False)}") print(f"Public IP: {status['network'].get('public_ip', 'Unknown')}") print(f"Dashboard: {helper.get_dashboard_url()}") if __name__ == '__main__': main()