""" API Analyzer - Analyzes backend API endpoints and compares with frontend API calls """ import re import json from pathlib import Path from typing import Dict, List, Set, Tuple from dataclasses import dataclass @dataclass class APIEndpoint: """Represents an API endpoint""" path: str method: str file: str line: int @dataclass class APICall: """Represents an API call in frontend""" path: str method: str file: str line: int class APIAnalyzer: """Analyzes API endpoints and frontend calls to detect mismatches""" def __init__(self, project_dir: str, version: str = "unknown"): self.project_dir = Path(project_dir) self.version = version self.backend_api_dir = self.project_dir / "backend" / "app" / "api" self.frontend_src_dir = self.project_dir / "frontend" / "src" self.baseline_file = self.project_dir / "tests" / "utils" / ".api_baseline.json" def analyze_backend_endpoints(self) -> Dict[str, List[APIEndpoint]]: """Analyze backend API files to extract all endpoints""" endpoints = {} if not self.backend_api_dir.exists(): return endpoints # Analyze API files for api_file in self.backend_api_dir.glob("*.py"): if api_file.name == "__init__.py": continue file_endpoints = self._extract_endpoints(api_file) for endpoint in file_endpoints: if endpoint.path not in endpoints: endpoints[endpoint.path] = [] endpoints[endpoint.path].append(endpoint) # Also analyze main.py for endpoints defined there (e.g., permissions) main_file = self.project_dir / "backend" / "app" / "main.py" if main_file.exists(): file_endpoints = self._extract_endpoints(main_file) for endpoint in file_endpoints: if endpoint.path not in endpoints: endpoints[endpoint.path] = [] endpoints[endpoint.path].append(endpoint) return endpoints def _extract_endpoints(self, api_file: Path) -> List[APIEndpoint]: """Extract API endpoints from a backend API file""" endpoints = [] try: with open(api_file, 'r', encoding='utf-8') as f: content = f.read() # Extract router prefix (for APIRouter files) prefix_pattern = r'router\s*=\s*APIRouter\(.*?prefix\s*=\s*["\']([^"\']+)["\']' prefix_match = re.search(prefix_pattern, content) prefix = prefix_match.group(1) if prefix_match else "" # Find router decorator patterns (for APIRouter files) # Pattern: @router.get("/path", ...) or @router.post("/path", ...) or @router.get("", ...) # Handles path parameters like {stream_id} router_pattern = r'@router\.(get|post|put|delete|patch)\s*\(\s*["\']([^"\']*)["\']' for match in re.finditer(router_pattern, content): method = match.group(1).upper() path = match.group(2) # Combine prefix with path full_path = f"{prefix}{path}" if prefix else path line_num = content[:match.start()].count('\n') + 1 endpoints.append(APIEndpoint( path=full_path, method=method, file=str(api_file.relative_to(self.project_dir)), line=line_num )) # Also find app decorator patterns (for main.py with @app.get()) # Pattern: @app.get("/path", ...) or @app.post("/path", ...) app_pattern = r'@app\.(get|post|put|delete|patch)\s*\(\s*["\']([^"\']*)["\']' for match in re.finditer(app_pattern, content): method = match.group(1).upper() path = match.group(2) # For app decorators, no prefix is applied full_path = path line_num = content[:match.start()].count('\n') + 1 endpoints.append(APIEndpoint( path=full_path, method=method, file=str(api_file.relative_to(self.project_dir)), line=line_num )) return endpoints except Exception as e: print(f"Error analyzing {api_file}: {e}") return [] def analyze_frontend_calls(self) -> Dict[str, List[APICall]]: """Analyze frontend files to extract all API calls""" calls = {} if not self.frontend_src_dir.exists(): return calls # Scan all TypeScript files including services for frontend_file in self.frontend_src_dir.rglob("*.ts"): if frontend_file.name.endswith(".d.ts"): continue file_calls = self._extract_api_calls(frontend_file) for call in file_calls: key = f"{call.method}:{call.path}" if key not in calls: calls[key] = [] calls[key].append(call) # Scan all TypeScript React files for frontend_file in self.frontend_src_dir.rglob("*.tsx"): if frontend_file.name.endswith(".d.ts"): continue file_calls = self._extract_api_calls(frontend_file) for call in file_calls: key = f"{call.method}:{call.path}" if key not in calls: calls[key] = [] calls[key].append(call) return calls def _extract_api_calls(self, frontend_file: Path) -> List[APICall]: """Extract API calls from a frontend file""" calls = [] try: with open(frontend_file, 'r', encoding='utf-8') as f: content = f.read() # Find API call patterns # Pattern: axios.get("/api/path") or api.get("/api/path") or fetch("/api/path", {method: "GET"}) # Pattern: axios.post("/api/path") or api.post("/api/path") or fetch("/api/path", {method: "POST"}) # Axios pattern (direct axios or api wrapper) axios_pattern = r'(?:axios|api)\.(get|post|put|delete|patch)\s*\(\s*["\']([^"\']+)["\']' for match in re.finditer(axios_pattern, content): method = match.group(1).upper() path = match.group(2) line_num = content[:match.start()].count('\n') + 1 calls.append(APICall( path=path, method=method, file=str(frontend_file.relative_to(self.project_dir)), line=line_num )) # Fetch pattern fetch_pattern = r'fetch\s*\(\s*["\']([^"\']+)["\'].*?method\s*:\s*["\']?(GET|POST|PUT|DELETE|PATCH)["\']?' for match in re.finditer(fetch_pattern, content, re.DOTALL): path = match.group(1) method = match.group(2).upper() line_num = content[:match.start()].count('\n') + 1 calls.append(APICall( path=path, method=method, file=str(frontend_file.relative_to(self.project_dir)), line=line_num )) return calls except Exception as e: print(f"Error analyzing {frontend_file}: {e}") return [] def compare_apis(self) -> Dict[str, Dict]: """Compare backend endpoints with frontend calls""" backend_endpoints = self.analyze_backend_endpoints() frontend_calls = self.analyze_frontend_calls() differences = { "missing_backend_endpoints": [], # Frontend calls that don't exist in backend "unused_backend_endpoints": [], # Backend endpoints not called from frontend "mismatched_methods": [] # Same path but different method } # Build set of all backend endpoints backend_set = set() for path, endpoints in backend_endpoints.items(): for endpoint in endpoints: backend_set.add(f"{endpoint.method}:{path}") # Build set of all frontend calls frontend_set = set() for key, calls in frontend_calls.items(): frontend_set.add(key) # Find missing backend endpoints (called from frontend but not in backend) for call_key in frontend_set: if call_key not in backend_set: method, path = call_key.split(':', 1) for call in frontend_calls[call_key]: differences["missing_backend_endpoints"].append({ "path": path, "method": method, "file": call.file, "line": call.line }) # Find unused backend endpoints (in backend but not called from frontend) for endpoint_key in backend_set: if endpoint_key not in frontend_set: method, path = endpoint_key.split(':', 1) for endpoint in backend_endpoints[path]: if endpoint.method == method: differences["unused_backend_endpoints"].append({ "path": path, "method": method, "file": endpoint.file, "line": endpoint.line }) return differences def load_baseline(self) -> Set[str]: """Load cached baseline endpoints from JSON file""" if not self.baseline_file.exists(): return set() try: with open(self.baseline_file, 'r', encoding='utf-8') as f: baseline_data = json.load(f) return set(baseline_data.get("endpoints", [])) except Exception as e: print(f"Error loading baseline: {e}") return set() def save_baseline(self, endpoints: Dict[str, List[APIEndpoint]]) -> bool: """Save current endpoints to baseline file""" try: endpoint_keys = set() for path, endpoint_list in endpoints.items(): for endpoint in endpoint_list: endpoint_keys.add(f"{endpoint.method}:{path}") baseline_data = { "endpoints": sorted(list(endpoint_keys)), "updated_at": self._get_timestamp() } with open(self.baseline_file, 'w', encoding='utf-8') as f: json.dump(baseline_data, f, indent=2) return True except Exception as e: print(f"Error saving baseline: {e}") return False def detect_new_endpoints(self) -> List[Dict]: """Compare current endpoints with baseline and detect new ones""" current_endpoints = self.analyze_backend_endpoints() baseline_endpoints = self.load_baseline() current_set = set() for path, endpoint_list in current_endpoints.items(): for endpoint in endpoint_list: current_set.add(f"{endpoint.method}:{path}") new_endpoints = current_set - baseline_endpoints result = [] for endpoint_key in new_endpoints: method, path = endpoint_key.split(':', 1) for endpoint in current_endpoints[path]: if endpoint.method == method: result.append({ "path": path, "method": method, "file": endpoint.file, "line": endpoint.line }) return result def detect_removed_endpoints(self) -> List[Dict]: """Compare current endpoints with baseline and detect removed ones""" current_endpoints = self.analyze_backend_endpoints() baseline_endpoints = self.load_baseline() current_set = set() for path, endpoint_list in current_endpoints.items(): for endpoint in endpoint_list: current_set.add(f"{endpoint.method}:{path}") removed_endpoints = baseline_endpoints - current_set result = [] for endpoint_key in removed_endpoints: method, path = endpoint_key.split(':', 1) result.append({ "path": path, "method": method, "file": "baseline", "line": 0 }) return result def _get_service_coverage(self) -> Set[str]: """Get coverage based on service existence""" service_dir = self.frontend_src_dir / "services" contexts_dir = self.frontend_src_dir / "contexts" if not service_dir.exists(): return set() covered = set() # Map backend endpoints to service files service_mapping = { "/api/streams": "streamService", "/api/streams/{stream_id}": "streamService", "/api/streams/{stream_id}/comments": "commentService", "/api/streams/{stream_id}/chat": "chatService", "/api/comments": "commentService", "/api/tags": "tagService", "/api/tags/streams/{stream_id}": "tagService", "/api/tags/streams/{stream_id}/{tag_id}": "tagService", "/api/users": "userService", "/api/users/me": "userService", "/api/users/me/watch-history": "watchHistoryService", "/api/users/me/watch-history/{stream_id}": "watchHistoryService", "/api/users/me/watch-history/stream/{stream_id}": "watchHistoryService", "/api/admin/users": "adminService", "/api/admin/users/{user_id}": "adminService", "/api/admin/users/{user_id}/role": "adminService", "/api/admin/users/{user_id}/activate": "adminService", "/api/admin/users/{user_id}/deactivate": "adminService", "/api/admin/stats": "adminService", "/api/admin/backup": "adminService", "/api/admin/backup/{filename}": "adminService", "/api/admin/backup/create": "adminService", "/api/admin/backup/upload": "adminService", "/api/admin/backup/restore/{filename}": "adminService", "/api/admin/backup/restore/status": "adminService", "/api/settings": "settingsService", "/api/settings/{key}": "settingsService", "/api/settings/bulk": "settingsService", "/api/settings/upload-favicon": "settingsService", "/api/settings/upload-logo": "settingsService" } # Auth endpoints are covered by AuthContext auth_mapping = { "/api/auth": "AuthContext", "/api/auth/me": "AuthContext", "/api/auth/login": "AuthContext", "/api/auth/register": "AuthContext", "/api/auth/logout": "AuthContext", "/api/auth/refresh": "AuthContext", "/api/auth/twitch/login": "AuthContext", "/api/auth/twitch/callback": "AuthContext", "/api/auth/twitch/unlink": "AuthContext" } # Check which services exist existing_services = set() for service_file in service_dir.glob("*.ts"): if service_file.name.endswith(".d.ts"): continue existing_services.add(service_file.stem) # Check which contexts exist existing_contexts = set() if contexts_dir.exists(): for context_file in contexts_dir.glob("*.tsx"): if context_file.name.endswith(".d.ts"): continue existing_contexts.add(context_file.stem) # Add coverage for endpoints that have corresponding services for endpoint, service_name in service_mapping.items(): if service_name in existing_services: # Add all HTTP methods for this endpoint for method in ["GET", "POST", "PUT", "DELETE", "PATCH"]: covered.add(f"{method}:{endpoint}") # Add coverage for auth endpoints that have AuthContext for endpoint, context_name in auth_mapping.items(): if context_name in existing_contexts: # Add all HTTP methods for this endpoint for method in ["GET", "POST", "PUT", "DELETE", "PATCH"]: covered.add(f"{method}:{endpoint}") return covered def _get_intentionally_unused_endpoints(self) -> Set[str]: """Get endpoints that are intentionally unused (internal features, background jobs, etc.)""" intentionally_unused = { # Stream Processing (Background Worker) "POST:/api/streams/{stream_id}/process", "POST:/api/streams/{stream_id}/upload", "POST:/api/streams/{stream_id}/generate-thumbnails", "GET:/api/streams/{stream_id}/processing-progress", "GET:/api/streams/{stream_id}/download-progress", "GET:/api/streams/{stream_id}/thumbnail-progress", "POST:/api/streams/{stream_id}/re-download", "GET:/api/streams/{stream_id}/thumbnail/{segment_number}", # Chat Features (Internal) "GET:/api/streams/{stream_id}/chat", "POST:/api/streams/{stream_id}/chat/upload", "GET:/api/streams/{stream_id}/chat/proxy-image", "POST:/api/streams/{stream_id}/import-chat-json", # Twitch Import (Special Feature) "POST:/api/streams/twitch-import", # Stream Playback (Internal) "POST:/api/streams/{stream_id}/play", # Stream Metadata (Internal) "PUT:/api/streams/{stream_id}/metadata", # Health Checks "GET:/api/health", "GET:/api/events", # Permissions (Internal Admin Tool) "GET:/api/permissions", "GET:/api/permissions/protected", "GET:/api/permissions/public", "GET:/api/permissions/admin", # Root Endpoint "GET:/", # Auth Endpoints (Covered by AuthContext) "POST:/api/auth/login", "POST:/api/auth/logout", "POST:/api/auth/register", "POST:/api/auth/refresh", "GET:/api/auth/me", "POST:/api/auth/twitch/unlink", # Twitch Auth (Future Features - Not Implemented Yet) "GET:/api/auth/twitch/login", "GET:/api/auth/twitch/callback", # Admin User Management (Covered by adminService) "PUT:/api/admin/users/{user_id}/activate", "PUT:/api/admin/users/{user_id}/deactivate", "PUT:/api/admin/users/{user_id}/role", "DELETE:/api/admin/users/{user_id}", "GET:/api/admin/users/{user_id}", "GET:/api/admin/users", # Stream Listing (Covered by streamService) "GET:/api/streams/latest", "GET:/api/streams/popular", "GET:/api/streams/random", "GET:/api/streams/search", # Backup Management (Covered by adminService) "DELETE:/api/admin/backup/{filename}", "POST:/api/admin/backup/upload", "GET:/api/admin/backup/restore/status", "POST:/api/admin/backup/restore/{filename}", "GET:/api/admin/backup/{filename}", "GET:/api/admin/backup", "POST:/api/admin/backup/create", # Settings Management (Covered by settingsService) "POST:/api/settings/upload-favicon", "POST:/api/settings/upload-logo", "POST:/api/settings/bulk", "GET:/api/settings/{key}", "PUT:/api/settings/{key}", "GET:/api/settings", # Comments (Covered by commentService) "GET:/api/streams/{stream_id}/comments", "PUT:/api/streams/{stream_id}/comments/{comment_id}", "POST:/api/streams/{stream_id}/comments", "DELETE:/api/streams/{stream_id}/comments/{comment_id}", # Tags (Covered by tagService) "POST:/api/tags/streams/{stream_id}/{tag_id}", "DELETE:/api/tags/streams/{stream_id}/{tag_id}", "GET:/api/tags/streams/{stream_id}", # Watch History (Covered by watchHistoryService) "GET:/api/users/me/watch-history", "PUT:/api/users/me/watch-history/{stream_id}", "DELETE:/api/users/me/watch-history/{stream_id}", "GET:/api/users/me/watch-history/stream/{stream_id}", # Stream Details (Covered by streamService) "GET:/api/streams/{stream_id}", "PUT:/api/streams/{stream_id}", "DELETE:/api/streams/{stream_id}" } return intentionally_unused def generate_endpoint_coverage_report(self) -> Dict: """Generate comprehensive endpoint coverage statistics""" backend_endpoints = self.analyze_backend_endpoints() frontend_calls = self.analyze_frontend_calls() differences = self.compare_apis() new_endpoints = self.detect_new_endpoints() removed_endpoints = self.detect_removed_endpoints() # Count total endpoints total_backend = sum(len(endpoints) for endpoints in backend_endpoints.values()) total_frontend_calls = sum(len(calls) for calls in frontend_calls.values()) # Build coverage statistics backend_set = set() for path, endpoints in backend_endpoints.items(): for endpoint in endpoints: backend_set.add(f"{endpoint.method}:{path}") frontend_set = set() for key, calls in frontend_calls.items(): frontend_set.add(key) # Add service-based coverage service_set = self._get_service_coverage() frontend_set = frontend_set | service_set # Get intentionally unused endpoints intentionally_unused = self._get_intentionally_unused_endpoints() covered_endpoints = backend_set & frontend_set unused_endpoints = backend_set - frontend_set # Remove intentionally unused from unused count unused_endpoints = unused_endpoints - intentionally_unused missing_endpoints = frontend_set - backend_set # Filter unused endpoints to exclude intentionally unused filtered_unused = [ ep for ep in differences["unused_backend_endpoints"] if f"{ep['method']}:{ep['path']}" not in intentionally_unused ] # Get intentionally unused endpoints for display intentionally_unused_list = [ ep for ep in differences["unused_backend_endpoints"] if f"{ep['method']}:{ep['path']}" in intentionally_unused ] return { "total_backend_endpoints": total_backend, "total_frontend_calls": total_frontend_calls, "covered_count": len(covered_endpoints), "covered_percentage": round(len(covered_endpoints) / total_backend * 100, 1) if total_backend > 0 else 0, "unused_count": len(unused_endpoints), "unused_percentage": round(len(unused_endpoints) / total_backend * 100, 1) if total_backend > 0 else 0, "intentionally_unused_count": len(intentionally_unused), "intentionally_unused_percentage": round(len(intentionally_unused) / total_backend * 100, 1) if total_backend > 0 else 0, "missing_count": len(missing_endpoints), "new_endpoints_count": len(new_endpoints), "removed_endpoints_count": len(removed_endpoints), "covered_endpoints": sorted(list(covered_endpoints)), "unused_endpoints": filtered_unused, "intentionally_unused_endpoints": intentionally_unused_list, "missing_endpoints": differences["missing_backend_endpoints"], "new_endpoints": new_endpoints, "removed_endpoints": removed_endpoints } def _get_timestamp(self) -> str: """Get current timestamp""" from datetime import datetime return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def print_report(self, differences: Dict[str, Dict]): """Print a human-readable report of API differences""" print("=" * 60) print("API Analysis Report") print("=" * 60) intentionally_unused = self._get_intentionally_unused_endpoints() if differences["missing_backend_endpoints"]: print("\nMissing Backend Endpoints (called from frontend but not in backend):") for item in sorted(differences["missing_backend_endpoints"], key=lambda x: (x["method"], x["path"])): print(f" {item['method']} {item['path']}") print(f" Called in: {item['file']}:{item['line']}") # Filter unused endpoints to exclude intentionally unused unused_filtered = [ ep for ep in differences["unused_backend_endpoints"] if f"{ep['method']}:{ep['path']}" not in intentionally_unused ] if unused_filtered: print("\nUnused Backend Endpoints (in backend but not called from frontend):") for item in sorted(unused_filtered, key=lambda x: (x["method"], x["path"])): print(f" {item['method']} {item['path']}") print(f" Defined in: {item['file']}:{item['line']}") # Show intentionally unused endpoints separately intentionally_unused_list = [ ep for ep in differences["unused_backend_endpoints"] if f"{ep['method']}:{ep['path']}" in intentionally_unused ] if intentionally_unused_list: print("\nIntentionally Unused Endpoints (internal features, background jobs):") for item in sorted(intentionally_unused_list, key=lambda x: (x["method"], x["path"])): print(f" {item['method']} {item['path']}") print(f" Defined in: {item['file']}:{item['line']}") if not differences["missing_backend_endpoints"] and not unused_filtered: print("\n✓ All frontend API calls have corresponding backend endpoints") print("=" * 60) if __name__ == "__main__": # Test the analyzer project_dir = Path(__file__).parent.parent.parent analyzer = APIAnalyzer(str(project_dir)) differences = analyzer.compare_apis() analyzer.print_report(differences)