from __future__ import annotations from dataclasses import dataclass from typing import Any import httpx from ..core.config import settings from ..core.redaction import redact_message from .base import BaseConnector @dataclass class _TautulliConfig: url: str api_key: str verify_tls: bool timeout_seconds: float def _as_bool(value: Any, default: bool = False) -> bool: if isinstance(value, bool): return value if value is None: return default return str(value).strip().lower() in {'1', 'true', 'yes', 'on'} def _as_float(value: Any, default: float) -> float: try: return float(value) except (TypeError, ValueError): return default def _as_int(value: Any, default: int = 0) -> int: try: return int(float(value)) except (TypeError, ValueError): return default def _safe_url() -> str: value = settings.apis.get('TAUTULLI_URL') return str(value).rstrip('/') if value else '' def _mask_ip(value: Any) -> str: text = str(value or '') parts = text.split('.') if len(parts) == 4: return f'{parts[0]}.{parts[1]}.{parts[2]}.x' return '[masked]' if text else '' def _list_from_data(data: Any, *keys: str) -> list[dict[str, Any]]: if isinstance(data, list): return [item for item in data if isinstance(item, dict)] if isinstance(data, dict): for key in keys: value = data.get(key) if isinstance(value, list): return [item for item in value if isinstance(item, dict)] value = data.get('data') if isinstance(value, list): return [item for item in value if isinstance(item, dict)] return [] def _title(raw: dict[str, Any]) -> str: title = str(raw.get('full_title') or raw.get('title') or raw.get('media_title') or 'unknown') parent = raw.get('grandparent_title') or raw.get('grandparentTitle') return f'{parent} - {title}' if parent and str(parent) not in title else title class TautulliConnector(BaseConnector): name = 'Tautulli' def is_configured(self) -> bool: return bool(settings.apis.get('TAUTULLI_URL') and settings.apis.get('TAUTULLI_API_KEY')) def _config(self) -> _TautulliConfig | None: if not self.is_configured(): return None return _TautulliConfig( url=str(settings.apis['TAUTULLI_URL']).rstrip('/'), api_key=str(settings.apis['TAUTULLI_API_KEY']), verify_tls=_as_bool(settings.apis.get('TAUTULLI_VERIFY_TLS'), False), timeout_seconds=_as_float(settings.apis.get('TAUTULLI_TIMEOUT_SECONDS'), 10.0), ) def _command(self, cmd: str, params: dict[str, Any] | None = None) -> dict[str, Any]: config = self._config() if not config: raise RuntimeError('missing_config') query = {'apikey': config.api_key, 'cmd': cmd, **(params or {})} try: self.clear_error() with httpx.Client(timeout=config.timeout_seconds, verify=config.verify_tls) as client: response = client.get(f'{config.url}/api/v2', params=query, headers={'Accept': 'application/json'}) except httpx.RequestError as exc: message = redact_message(str(exc)) self.set_error(message) raise RuntimeError(message) if response.status_code in {401, 403}: self.set_error('auth_failed') raise RuntimeError('auth_failed') if response.status_code >= 400: message = redact_message(f'Tautulli API error: HTTP {response.status_code} {(response.text or "")[:256]}') self.set_error(message) raise RuntimeError(message) try: payload = response.json() except ValueError as exc: message = redact_message(str(exc)) self.set_error(message) raise RuntimeError(message) response_meta = payload.get('response') if isinstance(payload, dict) else {} if isinstance(response_meta, dict) and str(response_meta.get('result', '')).lower() == 'error': message = redact_message(str(response_meta.get('message') or 'tautulli_error')) if 'api' in message.lower() and 'key' in message.lower(): self.set_error('auth_failed') raise RuntimeError('auth_failed') self.set_error(message) raise RuntimeError(message) return payload @staticmethod def _data(payload: dict[str, Any]) -> Any: response = payload.get('response') if isinstance(payload, dict) else {} return response.get('data') if isinstance(response, dict) else None def test_connection(self) -> dict[str, Any]: if not self.is_configured(): return {'status': 'missing_config', 'reason': 'missing_config'} try: data = self._data(self._command('get_server_info')) except RuntimeError as exc: reason = redact_message(str(exc)) if reason.startswith('auth_failed'): return {'status': 'auth_failed', 'reason': reason} if reason == 'missing_config': return {'status': 'missing_config', 'reason': reason} return {'status': 'unreachable', 'reason': reason} return { 'status': 'connected', 'tautulliVersion': str((data or {}).get('tautulli_version') or 'unknown') if isinstance(data, dict) else 'unknown', 'plexServerName': str((data or {}).get('pms_name') or '') if isinstance(data, dict) else '', } def health(self) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config', 'url': '', 'required_env_vars': ['TAUTULLI_URL', 'TAUTULLI_API_KEY']} result = self.test_connection() status = str(result.get('status', 'error')) return { 'success': status == 'connected', 'status': status, 'message': 'Tautulli connector ready.' if status == 'connected' else redact_message(str(result.get('reason', 'connection_failed'))), 'url': _safe_url(), 'tautulliVersion': str(result.get('tautulliVersion', 'unknown')), 'plexServerName': str(result.get('plexServerName', '')), } def status(self) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config', 'data': {}} try: data = self._data(self._command('status')) except RuntimeError as exc: return self._error_payload(exc, data={}) return {'success': True, 'status': 'ok', 'message': 'status_ok', 'data': data or {}} def activity(self) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config', 'streamCount': 0, 'sessions': []} try: data = self._data(self._command('get_activity')) except RuntimeError as exc: return self._error_payload(exc, streamCount=0, sessions=[]) records = _list_from_data(data, 'sessions') sessions = [self._map_session(item) for item in records] return { 'success': True, 'status': 'ok', 'message': 'activity_ok', 'streamCount': _as_int((data or {}).get('stream_count'), len(sessions)) if isinstance(data, dict) else len(sessions), 'lanBandwidth': _as_int((data or {}).get('lan_bandwidth')) if isinstance(data, dict) else 0, 'wanBandwidth': _as_int((data or {}).get('wan_bandwidth')) if isinstance(data, dict) else 0, 'totalBandwidth': _as_int((data or {}).get('total_bandwidth')) if isinstance(data, dict) else sum(item['bandwidth'] for item in sessions), 'sessions': sessions, 'data': {'sessions': sessions}, } def history(self, limit: int = 25) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config', 'total': 0, 'items': []} try: data = self._data(self._command('get_history', {'length': limit})) except RuntimeError as exc: return self._error_payload(exc, total=0, items=[]) items = [self._map_history(item) for item in _list_from_data(data, 'data', 'history')] return {'success': True, 'status': 'ok', 'message': 'history_ok', 'total': len(items), 'items': items, 'data': {'items': items}} def users(self) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config', 'total': 0, 'items': []} try: data = self._data(self._command('get_users')) except RuntimeError as exc: return self._error_payload(exc, total=0, items=[]) items = [self._map_user(item) for item in _list_from_data(data, 'users')] return {'success': True, 'status': 'ok', 'message': 'users_ok', 'total': len(items), 'items': items, 'data': {'items': items}} def libraries(self) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config', 'total': 0, 'items': []} try: data = self._data(self._command('get_libraries')) except RuntimeError as exc: return self._error_payload(exc, total=0, items=[]) items = [self._map_library(item) for item in _list_from_data(data, 'libraries')] return {'success': True, 'status': 'ok', 'message': 'libraries_ok', 'total': len(items), 'items': items, 'libraries': items, 'data': {'items': items}} def recently_added(self) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config', 'total': 0, 'items': []} try: data = self._data(self._command('get_recently_added', {'count': 25})) except RuntimeError as exc: return self._error_payload(exc, total=0, items=[]) items = [self._map_recent(item) for item in _list_from_data(data, 'recently_added', 'items')] return {'success': True, 'status': 'ok', 'message': 'recently_added_ok', 'total': len(items), 'items': items, 'data': {'items': items}} def watch_stats(self) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config', 'totalPlays': 0, 'totalDuration': 0, 'topUsers': [], 'topLibraries': [], 'topPlatforms': [], 'playsByDay': []} try: by_day = [self._map_day(item) for item in _list_from_data(self._data(self._command('get_plays_by_date')), 'data')] top_users = [self._map_top_user(item) for item in _list_from_data(self._data(self._command('get_plays_by_top_10_users')), 'data')] top_platforms = [self._map_top_platform(item) for item in _list_from_data(self._data(self._command('get_plays_by_top_10_platforms')), 'data')] except RuntimeError as exc: return self._error_payload(exc, totalPlays=0, totalDuration=0, topUsers=[], topLibraries=[], topPlatforms=[], playsByDay=[]) total_plays = sum(_as_int(item.get('plays')) for item in by_day) or sum(_as_int(item.get('plays')) for item in top_users) total_duration = sum(_as_int(item.get('duration')) for item in by_day) or sum(_as_int(item.get('duration')) for item in top_users) return { 'success': True, 'status': 'ok', 'message': 'watch_stats_ok', 'totalPlays': total_plays, 'totalDuration': total_duration, 'topUsers': top_users, 'topLibraries': [], 'topPlatforms': top_platforms, 'playsByDay': by_day, 'data': {'topUsers': top_users, 'topPlatforms': top_platforms, 'playsByDay': by_day}, } def stream_summary(self) -> dict[str, Any]: activity = self.activity() sessions = activity.get('sessions', []) if isinstance(activity, dict) else [] return { 'success': bool(activity.get('success')), 'status': activity.get('status', 'unreachable'), 'message': activity.get('message', 'stream_summary'), 'activeStreams': len(sessions), 'directPlay': sum(1 for item in sessions if item.get('transcodeDecision') == 'directplay'), 'directStream': sum(1 for item in sessions if item.get('transcodeDecision') == 'directstream'), 'transcoding': sum(1 for item in sessions if item.get('transcodeDecision') == 'transcode'), 'lanStreams': sum(1 for item in sessions if item.get('location') == 'lan'), 'wanStreams': sum(1 for item in sessions if item.get('location') == 'wan'), 'bandwidth': sum(_as_int(item.get('bandwidth')) for item in sessions), 'data': {'sessions': sessions}, } def plex_info(self) -> dict[str, Any]: if not self.is_configured(): return {'success': False, 'status': 'missing_config', 'message': 'missing_config'} try: data = self._data(self._command('get_server_info')) except RuntimeError as exc: return self._error_payload(exc) data = data if isinstance(data, dict) else {} machine = str(data.get('pms_identifier') or data.get('machine_identifier') or '') return { 'success': True, 'status': 'ok', 'message': 'plex_info_ok', 'name': str(data.get('pms_name') or data.get('name') or ''), 'version': str(data.get('pms_version') or data.get('version') or 'unknown'), 'platform': str(data.get('pms_platform') or data.get('platform') or ''), 'machineIdentifier': f'{machine[:8]}...' if machine else '', 'plexPass': bool(data.get('pms_is_plexpass', data.get('plexpass', False))), 'data': {}, } @staticmethod def _map_session(raw: dict[str, Any]) -> dict[str, Any]: decision = str(raw.get('transcode_decision') or raw.get('transcodeDecision') or 'direct play').lower().replace(' ', '') if decision == 'copy': decision = 'directstream' return { 'sessionKey': str(raw.get('session_key') or raw.get('sessionKey') or ''), 'user': str(raw.get('friendly_name') or raw.get('username') or raw.get('user') or 'unknown'), 'title': _title(raw), 'mediaType': str(raw.get('media_type') or raw.get('type') or ''), 'player': str(raw.get('player') or ''), 'platform': str(raw.get('platform') or ''), 'state': str(raw.get('state') or ''), 'progress': _as_int(raw.get('progress_percent') or raw.get('progress')), 'qualityProfile': str(raw.get('quality_profile') or raw.get('qualityProfile') or ''), 'transcodeDecision': decision, 'videoDecision': str(raw.get('video_decision') or raw.get('videoDecision') or decision).lower().replace(' ', ''), 'audioDecision': str(raw.get('audio_decision') or raw.get('audioDecision') or '').lower().replace(' ', ''), 'bandwidth': _as_int(raw.get('bandwidth')), 'ipAddress': _mask_ip(raw.get('ip_address') or raw.get('ipAddress')), 'location': str(raw.get('location') or '').lower(), } @staticmethod def _map_history(raw: dict[str, Any]) -> dict[str, Any]: return { 'rowId': raw.get('row_id') or raw.get('rowId'), 'user': str(raw.get('friendly_name') or raw.get('username') or 'unknown'), 'title': _title(raw), 'mediaType': str(raw.get('media_type') or ''), 'watchedStatus': raw.get('watched_status'), 'date': raw.get('date'), 'duration': _as_int(raw.get('duration')), 'platform': str(raw.get('platform') or ''), 'transcodeDecision': str(raw.get('transcode_decision') or raw.get('transcodeDecision') or '').lower().replace(' ', ''), } @staticmethod def _map_user(raw: dict[str, Any]) -> dict[str, Any]: return { 'userId': str(raw.get('user_id') or raw.get('userId') or ''), 'username': str(raw.get('username') or ''), 'friendlyName': str(raw.get('friendly_name') or raw.get('friendlyName') or raw.get('username') or ''), 'lastSeen': raw.get('last_seen') or raw.get('lastSeen'), 'plays': _as_int(raw.get('plays')), 'duration': _as_int(raw.get('duration')), } @staticmethod def _map_library(raw: dict[str, Any]) -> dict[str, Any]: return { 'sectionId': str(raw.get('section_id') or raw.get('sectionId') or ''), 'name': str(raw.get('section_name') or raw.get('name') or ''), 'type': str(raw.get('section_type') or raw.get('type') or ''), 'count': _as_int(raw.get('count')), 'parentCount': _as_int(raw.get('parent_count') or raw.get('parentCount')), 'childCount': _as_int(raw.get('child_count') or raw.get('childCount')), 'lastAccessed': raw.get('last_accessed') or raw.get('lastAccessed'), } @staticmethod def _map_recent(raw: dict[str, Any]) -> dict[str, Any]: return { 'title': _title(raw), 'mediaType': str(raw.get('media_type') or raw.get('type') or ''), 'library': str(raw.get('section_name') or raw.get('library') or ''), 'addedAt': raw.get('added_at') or raw.get('addedAt'), 'year': raw.get('year'), 'thumb': raw.get('thumb'), } @staticmethod def _map_day(raw: dict[str, Any]) -> dict[str, Any]: return {'date': raw.get('date'), 'plays': _as_int(raw.get('plays')), 'duration': _as_int(raw.get('duration'))} @staticmethod def _map_top_user(raw: dict[str, Any]) -> dict[str, Any]: return {'name': str(raw.get('friendly_name') or raw.get('username') or raw.get('user') or 'unknown'), 'plays': _as_int(raw.get('plays')), 'duration': _as_int(raw.get('duration'))} @staticmethod def _map_top_platform(raw: dict[str, Any]) -> dict[str, Any]: return {'platform': str(raw.get('platform') or 'unknown'), 'plays': _as_int(raw.get('plays')), 'duration': _as_int(raw.get('duration'))} @staticmethod def _error_payload(exc: Exception, **extra: Any) -> dict[str, Any]: reason = redact_message(str(exc)) status = 'auth_failed' if reason.startswith('auth_failed') else ('missing_config' if reason == 'missing_config' else 'unreachable') return {'success': False, 'status': status, 'message': reason, **extra}