from __future__ import annotations from dataclasses import dataclass from typing import Any from urllib.parse import urlparse import httpx from ..core.config import settings from ..core.redaction import redact_message from .base import BaseConnector @dataclass class _WidefrogConfig: url: str api_key: str | None verify_tls: bool timeout_seconds: float def _as_float(value: Any, default: float) -> float: try: return float(value) except (TypeError, ValueError): return default def _safe_url() -> str: value = settings.apis.get('WIDEFROG_URL') if not value: return '' parsed = urlparse(str(value).rstrip('/')) if not parsed.scheme or not parsed.netloc: return str(value).rstrip('/') return f'{parsed.scheme}://{parsed.netloc}' def _port() -> int: parsed = urlparse(_safe_url()) return parsed.port or 1207 class WidefrogConnector(BaseConnector): name = 'Widefrog' def is_configured(self) -> bool: return bool(settings.apis.get('WIDEFROG_URL')) def _config(self) -> _WidefrogConfig | None: if not self.is_configured(): return None return _WidefrogConfig( url=str(settings.apis['WIDEFROG_URL']).rstrip('/'), api_key=str(settings.apis['WIDEFROG_API_KEY']) if settings.apis.get('WIDEFROG_API_KEY') else None, verify_tls=bool(settings.apis.get('WIDEFROG_VERIFY_TLS', False)), timeout_seconds=_as_float(settings.apis.get('WIDEFROG_TIMEOUT_SECONDS'), 10.0), ) def _request_json(self, method: str, path: str, **kwargs: Any) -> Any: config = self._config() if not config: raise RuntimeError('missing_config') headers = dict(kwargs.pop('headers', {}) or {}) if config.api_key: headers['X-Api-Key'] = config.api_key try: with httpx.Client(timeout=config.timeout_seconds, verify=config.verify_tls) as client: response = client.request(method=method, url=f'{config.url}{path}', headers=headers, **kwargs) except (httpx.TimeoutException, httpx.ConnectError, httpx.RequestError, ValueError) as exc: message = redact_message(str(exc)) self.set_error(message) raise RuntimeError(message) if response.status_code == 404: raise RuntimeError('unsupported_endpoint') if response.status_code in {401, 403}: raise RuntimeError('unauthorized') if response.status_code >= 500: raise RuntimeError(f'Widefrog server error: HTTP {response.status_code}') if response.status_code >= 400: raise RuntimeError(redact_message(f'Widefrog API error: HTTP {response.status_code} {(response.text or "")[:256]}')) if response.status_code == 204: return {} content_type = response.headers.get('content-type', '') if 'json' not in content_type and response.text: return {'message': response.text[:256], 'status': 'ok'} try: return response.json() if response.text else {} except ValueError as exc: raise RuntimeError(redact_message(str(exc))) def _first_supported(self, method: str, paths: list[str], **kwargs: Any) -> tuple[Any | None, str | None, str | None]: last_error: str | None = None for path in paths: try: return self._request_json(method, path, **kwargs), path, None except RuntimeError as exc: reason = str(exc) if reason == 'unsupported_endpoint': last_error = reason continue return None, path, redact_message(reason) return None, None, last_error or 'unsupported_endpoint' def test_connection(self) -> dict[str, Any]: if not self.is_configured(): return {'status': 'missing_config', 'reason': 'missing_config'} payload, _path, error = self._first_supported('GET', ['/api/health', '/health', '/']) if error: return {'status': 'unreachable' if error != 'unsupported_endpoint' else 'error', 'reason': redact_message(error)} version = payload.get('version') if isinstance(payload, dict) else None return {'status': 'connected', 'version': str(version or 'unknown')} def health(self) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config', 'url': '', 'port': 1207, 'version': 'unknown', 'checks': [], 'required_env_vars': ['WIDEFROG_URL']} payload, _path, error = self._first_supported('GET', ['/api/health', '/health', '/']) if error: return {'success': False, 'status': 'unreachable' if error != 'unsupported_endpoint' else 'error', 'message': redact_message(error), 'url': _safe_url(), 'port': _port(), 'version': 'unknown', 'checks': []} data = payload if isinstance(payload, dict) else {} return {'success': True, 'status': 'connected', 'message': 'Widefrog connector ready.', 'url': _safe_url(), 'port': _port(), 'version': str(data.get('version', 'unknown')), 'checks': data.get('checks', []) if isinstance(data.get('checks'), list) else []} def queue_status(self) -> dict[str, Any]: if not self.is_configured(): return self._empty_queue('missing_config') payload, _path, error = self._first_supported('GET', ['/api/queue', '/api/jobs']) if error: result = self._empty_queue('unavailable') result['message'] = redact_message(error) return result items = _extract_items(payload) return _queue_summary(items) def failed_jobs(self) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config', 'failed': 0, 'items': []} payload, _path, error = self._first_supported('GET', ['/api/failed', '/api/jobs/failed', '/api/queue/failed', '/api/jobs']) if error: return {'success': False, 'status': 'unavailable', 'message': redact_message(error), 'failed': 0, 'items': []} items = [item for item in _extract_items(payload) if str(item.get('status', '')).lower() == 'failed'] if not items: items = _extract_items(payload) return {'success': True, 'status': 'ok', 'failed': len(items), 'items': items} def activity(self) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config', 'recent': [], 'last_event': None} payload, _path, error = self._first_supported('GET', ['/api/activity', '/api/events', '/api/history']) if error: return {'success': False, 'status': 'unavailable', 'message': redact_message(error), 'recent': [], 'last_event': None} recent = _extract_items(payload, key='recent') return {'success': True, 'status': 'ok', 'recent': recent, 'last_event': recent[0] if recent else None} def download_status(self) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config', 'active': 0, 'speed': None, 'processing': 0, 'items': []} payload, _path, error = self._first_supported('GET', ['/api/download-status', '/api/downloads', '/api/queue/downloads']) if error: return {'success': False, 'status': 'unavailable', 'message': redact_message(error), 'active': 0, 'speed': None, 'processing': 0, 'items': []} data = payload if isinstance(payload, dict) else {} items = _extract_items(payload) return {'success': True, 'status': 'ok', 'active': int(data.get('active', len(items)) or 0), 'speed': data.get('speed'), 'processing': int(data.get('processing', 0) or 0), 'items': items} def retry_failed(self, payload: dict[str, Any] | None = None) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config'} result, _path, error = self._first_supported('POST', ['/api/retry-failed', '/api/jobs/retry-failed', '/api/failed/retry'], json=payload or {}) if error: return {'success': False, 'status': 'unsupported' if error == 'unsupported_endpoint' else 'error', 'message': redact_message(error)} return {'success': True, 'status': 'ok', 'message': 'retry_started', 'data': result if isinstance(result, dict) else {}} def scan_library(self, payload: dict[str, Any] | None = None) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config'} result, _path, error = self._first_supported('POST', ['/api/scan-library', '/api/library/scan', '/api/scan'], json=payload or {}) if error: return {'success': False, 'status': 'unsupported' if error == 'unsupported_endpoint' else 'error', 'message': redact_message(error)} return {'success': True, 'status': 'ok', 'message': 'scan_started', 'data': result if isinstance(result, dict) else {}} @staticmethod def _empty_queue(status: str) -> dict[str, Any]: return {'success': False, 'status': status, 'message': status, 'total': 0, 'active': 0, 'queued': 0, 'completed': 0, 'failed': 0, 'items': []} def _extract_items(payload: Any, key: str = 'items') -> list[dict[str, Any]]: if isinstance(payload, list): return [item for item in payload if isinstance(item, dict)] if isinstance(payload, dict): value = payload.get(key) or payload.get('jobs') or payload.get('queue') or [] if isinstance(value, list): return [item for item in value if isinstance(item, dict)] return [] def _queue_summary(items: list[dict[str, Any]]) -> dict[str, Any]: normalized = [{**item, 'status': str(item.get('status', 'unknown')).lower()} for item in items] return { 'success': True, 'status': 'ok', 'total': len(normalized), 'active': sum(1 for item in normalized if item['status'] in {'active', 'running', 'processing'}), 'queued': sum(1 for item in normalized if item['status'] in {'queued', 'pending'}), 'completed': sum(1 for item in normalized if item['status'] in {'completed', 'done', 'success'}), 'failed': sum(1 for item in normalized if item['status'] in {'failed', 'error'}), 'items': normalized, }