Api Design Reviewer

#!/usr/bin/env python3
"""
API Linter - Analyzes OpenAPI/Swagger specifications for REST conventions and best practices.

This script validates API designs against established conventions including:
- Resource naming conventions (kebab-case resources, camelCase fields)
- HTTP method usage patterns
- URL structure consistency
- Error response format standards
- Documentation completeness
- Pagination patterns
- Versioning compliance

Supports both OpenAPI JSON specifications and raw endpoint definition JSON.
"""

import argparse
import json
import re
import sys
from typing import Any, Dict, List, Tuple, Optional, Set
from urllib.parse import urlparse
from dataclasses import dataclass, field


@dataclass
class LintIssue:
    """Represents a linting issue found in the API specification."""
    severity: str  # 'error', 'warning', 'info'
    category: str
    message: str
    path: str
    suggestion: str = ""
    line_number: Optional[int] = None


@dataclass
class LintReport:
    """Complete linting report with issues and statistics."""
    issues: List[LintIssue] = field(default_factory=list)
    total_endpoints: int = 0
    endpoints_with_issues: int = 0
    score: float = 0.0
    
    def add_issue(self, issue: LintIssue) -> None:
        """Add an issue to the report."""
        self.issues.append(issue)
    
    def get_issues_by_severity(self) -> Dict[str, List[LintIssue]]:
        """Group issues by severity level."""
        grouped = {'error': [], 'warning': [], 'info': []}
        for issue in self.issues:
            if issue.severity in grouped:
                grouped[issue.severity].append(issue)
        return grouped
    
    def calculate_score(self) -> float:
        """Calculate overall API quality score (0-100)."""
        if self.total_endpoints == 0:
            return 100.0
        
        error_penalty = len([i for i in self.issues if i.severity == 'error']) * 10
        warning_penalty = len([i for i in self.issues if i.severity == 'warning']) * 3
        info_penalty = len([i for i in self.issues if i.severity == 'info']) * 1
        
        total_penalty = error_penalty + warning_penalty + info_penalty
        base_score = 100.0
        
        # Penalty per endpoint to normalize across API sizes
        penalty_per_endpoint = total_penalty / self.total_endpoints if self.total_endpoints > 0 else total_penalty
        
        self.score = max(0.0, base_score - penalty_per_endpoint)
        return self.score


class APILinter:
    """Main API linting engine."""
    
    def __init__(self):
        self.report = LintReport()
        self.openapi_spec: Optional[Dict] = None
        self.raw_endpoints: Optional[Dict] = None
        
        # Regex patterns for naming conventions
        self.kebab_case_pattern = re.compile(r'^[a-z]+(?:-[a-z0-9]+)*$')
        self.camel_case_pattern = re.compile(r'^[a-z][a-zA-Z0-9]*$')
        self.snake_case_pattern = re.compile(r'^[a-z]+(?:_[a-z0-9]+)*$')
        self.pascal_case_pattern = re.compile(r'^[A-Z][a-zA-Z0-9]*$')
        
        # Standard HTTP methods
        self.http_methods = {'GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', 'OPTIONS'}
        
        # Standard HTTP status codes by method
        self.standard_status_codes = {
            'GET': {200, 304, 404},
            'POST': {200, 201, 400, 409, 422},
            'PUT': {200, 204, 400, 404, 409},
            'PATCH': {200, 204, 400, 404, 409},
            'DELETE': {200, 204, 404},
            'HEAD': {200, 404},
            'OPTIONS': {200}
        }
        
        # Common error status codes
        self.common_error_codes = {400, 401, 403, 404, 405, 409, 422, 429, 500, 502, 503}

    def lint_openapi_spec(self, spec: Dict[str, Any]) -> LintReport:
        """Lint an OpenAPI/Swagger specification."""
        self.openapi_spec = spec
        self.report = LintReport()
        
        # Basic structure validation
        self._validate_openapi_structure()
        
        # Info section validation
        self._validate_info_section()
        
        # Server section validation
        self._validate_servers_section()
        
        # Paths validation (main linting logic)
        self._validate_paths_section()
        
        # Components validation
        self._validate_components_section()
        
        # Security validation
        self._validate_security_section()
        
        # Calculate final score
        self.report.calculate_score()
        
        return self.report

    def lint_raw_endpoints(self, endpoints: Dict[str, Any]) -> LintReport:
        """Lint raw endpoint definitions."""
        self.raw_endpoints = endpoints
        self.report = LintReport()
        
        # Validate raw endpoint structure
        self._validate_raw_endpoint_structure()
        
        # Lint each endpoint
        for endpoint_path, endpoint_data in endpoints.get('endpoints', {}).items():
            self._lint_raw_endpoint(endpoint_path, endpoint_data)
        
        self.report.calculate_score()
        return self.report

    def _validate_openapi_structure(self) -> None:
        """Validate basic OpenAPI document structure."""
        required_fields = ['openapi', 'info', 'paths']
        
        for field in required_fields:
            if field not in self.openapi_spec:
                self.report.add_issue(LintIssue(
                    severity='error',
                    category='structure',
                    message=f"Missing required field: {field}",
                    path=f"/{field}",
                    suggestion=f"Add the '{field}' field to the root of your OpenAPI specification"
                ))

    def _validate_info_section(self) -> None:
        """Validate the info section of OpenAPI spec."""
        if 'info' not in self.openapi_spec:
            return
            
        info = self.openapi_spec['info']
        required_info_fields = ['title', 'version']
        recommended_info_fields = ['description', 'contact']
        
        for field in required_info_fields:
            if field not in info:
                self.report.add_issue(LintIssue(
                    severity='error',
                    category='documentation',
                    message=f"Missing required info field: {field}",
                    path=f"/info/{field}",
                    suggestion=f"Add a '{field}' field to the info section"
                ))
        
        for field in recommended_info_fields:
            if field not in info:
                self.report.add_issue(LintIssue(
                    severity='warning',
                    category='documentation',
                    message=f"Missing recommended info field: {field}",
                    path=f"/info/{field}",
                    suggestion=f"Consider adding a '{field}' field to improve API documentation"
                ))
        
        # Validate version format
        if 'version' in info:
            version = info['version']
            if not re.match(r'^\d+\.\d+(\.\d+)?(-\w+)?$', version):
                self.report.add_issue(LintIssue(
                    severity='warning',
                    category='versioning',
                    message=f"Version format '{version}' doesn't follow semantic versioning",
                    path="/info/version",
                    suggestion="Use semantic versioning format (e.g., '1.0.0', '2.1.3-beta')"
                ))

    def _validate_servers_section(self) -> None:
        """Validate the servers section."""
        if 'servers' not in self.openapi_spec:
            self.report.add_issue(LintIssue(
                severity='warning',
                category='configuration',
                message="Missing servers section",
                path="/servers",
                suggestion="Add a servers section to specify API base URLs"
            ))
            return
        
        servers = self.openapi_spec['servers']
        if not isinstance(servers, list) or len(servers) == 0:
            self.report.add_issue(LintIssue(
                severity='warning',
                category='configuration',
                message="Empty servers section",
                path="/servers",
                suggestion="Add at least one server URL"
            ))

    def _validate_paths_section(self) -> None:
        """Validate all API paths and operations."""
        if 'paths' not in self.openapi_spec:
            return
            
        paths = self.openapi_spec['paths']
        if not paths:
            self.report.add_issue(LintIssue(
                severity='error',
                category='structure',
                message="No paths defined in API specification",
                path="/paths",
                suggestion="Define at least one API endpoint"
            ))
            return
        
        self.report.total_endpoints = sum(
            len([method for method in path_obj.keys() if method.upper() in self.http_methods])
            for path_obj in paths.values() if isinstance(path_obj, dict)
        )
        
        endpoints_with_issues = set()
        
        for path, path_obj in paths.items():
            if not isinstance(path_obj, dict):
                continue
                
            # Validate path structure
            path_issues = self._validate_path_structure(path)
            if path_issues:
                endpoints_with_issues.add(path)
            
            # Validate each operation in the path
            for method, operation in path_obj.items():
                if method.upper() not in self.http_methods:
                    continue
                    
                operation_issues = self._validate_operation(path, method.upper(), operation)
                if operation_issues:
                    endpoints_with_issues.add(path)
        
        self.report.endpoints_with_issues = len(endpoints_with_issues)

    def _validate_path_structure(self, path: str) -> bool:
        """Validate REST path structure and naming conventions."""
        has_issues = False
        
        # Check if path starts with slash
        if not path.startswith('/'):
            self.report.add_issue(LintIssue(
                severity='error',
                category='url_structure',
                message=f"Path must start with '/' character: {path}",
                path=f"/paths/{path}",
                suggestion=f"Change '{path}' to '/{path.lstrip('/')}'"
            ))
            has_issues = True
        
        # Split path into segments
        segments = [seg for seg in path.split('/') if seg]
        
        # Check for empty segments (double slashes)
        if '//' in path:
            self.report.add_issue(LintIssue(
                severity='error',
                category='url_structure',
                message=f"Path contains empty segments: {path}",
                path=f"/paths/{path}",
                suggestion="Remove double slashes from the path"
            ))
            has_issues = True
        
        # Validate each segment
        for i, segment in enumerate(segments):
            # Skip parameter segments
            if segment.startswith('{') and segment.endswith('}'):
                # Validate parameter naming
                param_name = segment[1:-1]
                if not self.camel_case_pattern.match(param_name) and not self.kebab_case_pattern.match(param_name):
                    self.report.add_issue(LintIssue(
                        severity='warning',
                        category='naming',
                        message=f"Path parameter '{param_name}' should use camelCase or kebab-case",
                        path=f"/paths/{path}",
                        suggestion=f"Use camelCase (e.g., 'userId') or kebab-case (e.g., 'user-id')"
                    ))
                    has_issues = True
                continue
            
            # Check for resource naming conventions
            if not self.kebab_case_pattern.match(segment):
                # Allow version segments like 'v1', 'v2'
                if not re.match(r'^v\d+$', segment):
                    self.report.add_issue(LintIssue(
                        severity='warning',
                        category='naming',
                        message=f"Resource segment '{segment}' should use kebab-case",
                        path=f"/paths/{path}",
                        suggestion=f"Use kebab-case for '{segment}' (e.g., 'user-profiles', 'order-items')"
                    ))
                    has_issues = True
            
            # Check for verb usage in URLs (anti-pattern)
            common_verbs = {'get', 'post', 'put', 'delete', 'create', 'update', 'remove', 'add'}
            if segment.lower() in common_verbs:
                self.report.add_issue(LintIssue(
                    severity='warning',
                    category='rest_conventions',
                    message=f"Avoid verbs in URLs: '{segment}' in {path}",
                    path=f"/paths/{path}",
                    suggestion="Use HTTP methods instead of verbs in URLs. Use nouns for resources."
                ))
                has_issues = True
        
        # Check path depth (avoid over-nesting)
        if len(segments) > 6:
            self.report.add_issue(LintIssue(
                severity='warning',
                category='url_structure',
                message=f"Path has excessive nesting ({len(segments)} levels): {path}",
                path=f"/paths/{path}",
                suggestion="Consider flattening the resource hierarchy or using query parameters"
            ))
            has_issues = True
        
        # Check for consistent versioning
        if any('v' + str(i) in segments for i in range(1, 10)):
            version_segments = [seg for seg in segments if re.match(r'^v\d+$', seg)]
            if len(version_segments) > 1:
                self.report.add_issue(LintIssue(
                    severity='error',
                    category='versioning',
                    message=f"Multiple version segments in path: {path}",
                    path=f"/paths/{path}",
                    suggestion="Use only one version segment per path"
                ))
                has_issues = True
        
        return has_issues

    def _validate_operation(self, path: str, method: str, operation: Dict[str, Any]) -> bool:
        """Validate individual operation (HTTP method + path combination)."""
        has_issues = False
        operation_path = f"/paths/{path}/{method.lower()}"
        
        # Check for required operation fields
        if 'responses' not in operation:
            self.report.add_issue(LintIssue(
                severity='error',
                category='structure',
                message=f"Missing responses section for {method} {path}",
                path=f"{operation_path}/responses",
                suggestion="Define expected responses for this operation"
            ))
            has_issues = True
        
        # Check for operation documentation
        if 'summary' not in operation:
            self.report.add_issue(LintIssue(
                severity='warning',
                category='documentation',
                message=f"Missing summary for {method} {path}",
                path=f"{operation_path}/summary",
                suggestion="Add a brief summary describing what this operation does"
            ))
            has_issues = True
        
        if 'description' not in operation:
            self.report.add_issue(LintIssue(
                severity='info',
                category='documentation',
                message=f"Missing description for {method} {path}",
                path=f"{operation_path}/description",
                suggestion="Add a detailed description for better API documentation"
            ))
            has_issues = True
        
        # Validate HTTP method usage patterns
        method_issues = self._validate_http_method_usage(path, method, operation)
        if method_issues:
            has_issues = True
        
        # Validate responses
        if 'responses' in operation:
            response_issues = self._validate_responses(path, method, operation['responses'])
            if response_issues:
                has_issues = True
        
        # Validate parameters
        if 'parameters' in operation:
            param_issues = self._validate_parameters(path, method, operation['parameters'])
            if param_issues:
                has_issues = True
        
        # Validate request body
        if 'requestBody' in operation:
            body_issues = self._validate_request_body(path, method, operation['requestBody'])
            if body_issues:
                has_issues = True
        
        return has_issues

    def _validate_http_method_usage(self, path: str, method: str, operation: Dict[str, Any]) -> bool:
        """Validate proper HTTP method usage patterns."""
        has_issues = False
        
        # GET requests should not have request body
        if method == 'GET' and 'requestBody' in operation:
            self.report.add_issue(LintIssue(
                severity='error',
                category='rest_conventions',
                message=f"GET request should not have request body: {method} {path}",
                path=f"/paths/{path}/{method.lower()}/requestBody",
                suggestion="Remove requestBody from GET request or use POST if body is needed"
            ))
            has_issues = True
        
        # DELETE requests typically should not have request body
        if method == 'DELETE' and 'requestBody' in operation:
            self.report.add_issue(LintIssue(
                severity='warning',
                category='rest_conventions',
                message=f"DELETE request typically should not have request body: {method} {path}",
                path=f"/paths/{path}/{method.lower()}/requestBody",
                suggestion="Consider using query parameters or path parameters instead"
            ))
            has_issues = True
        
        # POST/PUT/PATCH should typically have request body (except for actions)
        if method in ['POST', 'PUT', 'PATCH'] and 'requestBody' not in operation:
            # Check if this is an action endpoint
            if not any(action in path.lower() for action in ['activate', 'deactivate', 'reset', 'confirm']):
                self.report.add_issue(LintIssue(
                    severity='info',
                    category='rest_conventions',
                    message=f"{method} request typically should have request body: {method} {path}",
                    path=f"/paths/{path}/{method.lower()}",
                    suggestion=f"Consider adding requestBody for {method} operation or use GET if no data is being sent"
                ))
                has_issues = True
        
        return has_issues

    def _validate_responses(self, path: str, method: str, responses: Dict[str, Any]) -> bool:
        """Validate response definitions."""
        has_issues = False
        
        # Check for success response
        success_codes = {'200', '201', '202', '204'}
        has_success = any(code in responses for code in success_codes)
        
        if not has_success:
            self.report.add_issue(LintIssue(
                severity='error',
                category='responses',
                message=f"Missing success response for {method} {path}",
                path=f"/paths/{path}/{method.lower()}/responses",
                suggestion="Define at least one success response (200, 201, 202, or 204)"
            ))
            has_issues = True
        
        # Check for error responses
        has_error_responses = any(code.startswith('4') or code.startswith('5') for code in responses.keys())
        
        if not has_error_responses:
            self.report.add_issue(LintIssue(
                severity='warning',
                category='responses',
                message=f"Missing error responses for {method} {path}",
                path=f"/paths/{path}/{method.lower()}/responses",
                suggestion="Define common error responses (400, 404, 500, etc.)"
            ))
            has_issues = True
        
        # Validate individual response codes
        for status_code, response in responses.items():
            if status_code == 'default':
                continue
                
            try:
                code_int = int(status_code)
            except ValueError:
                self.report.add_issue(LintIssue(
                    severity='error',
                    category='responses',
                    message=f"Invalid status code '{status_code}' for {method} {path}",
                    path=f"/paths/{path}/{method.lower()}/responses/{status_code}",
                    suggestion="Use valid HTTP status codes (e.g., 200, 404, 500)"
                ))
                has_issues = True
                continue
            
            # Check if status code is appropriate for the method
            expected_codes = self.standard_status_codes.get(method, set())
            common_codes = {400, 401, 403, 404, 429, 500}  # Always acceptable
            
            if expected_codes and code_int not in expected_codes and code_int not in common_codes:
                self.report.add_issue(LintIssue(
                    severity='info',
                    category='responses',
                    message=f"Uncommon status code {status_code} for {method} {path}",
                    path=f"/paths/{path}/{method.lower()}/responses/{status_code}",
                    suggestion=f"Consider using standard codes for {method}: {sorted(expected_codes)}"
                ))
                has_issues = True
        
        return has_issues

    def _validate_parameters(self, path: str, method: str, parameters: List[Dict[str, Any]]) -> bool:
        """Validate parameter definitions."""
        has_issues = False
        
        for i, param in enumerate(parameters):
            param_path = f"/paths/{path}/{method.lower()}/parameters[{i}]"
            
            # Check required fields
            if 'name' not in param:
                self.report.add_issue(LintIssue(
                    severity='error',
                    category='parameters',
                    message=f"Parameter missing name field in {method} {path}",
                    path=f"{param_path}/name",
                    suggestion="Add a name field to the parameter"
                ))
                has_issues = True
                continue
            
            if 'in' not in param:
                self.report.add_issue(LintIssue(
                    severity='error',
                    category='parameters',
                    message=f"Parameter '{param['name']}' missing 'in' field in {method} {path}",
                    path=f"{param_path}/in",
                    suggestion="Specify parameter location (query, path, header, cookie)"
                ))
                has_issues = True
            
            # Validate parameter naming
            param_name = param['name']
            param_location = param.get('in', '')
            
            if param_location == 'query':
                # Query parameters should use camelCase or kebab-case
                if not self.camel_case_pattern.match(param_name) and not self.kebab_case_pattern.match(param_name):
                    self.report.add_issue(LintIssue(
                        severity='warning',
                        category='naming',
                        message=f"Query parameter '{param_name}' should use camelCase or kebab-case in {method} {path}",
                        path=f"{param_path}/name",
                        suggestion="Use camelCase (e.g., 'pageSize') or kebab-case (e.g., 'page-size')"
                    ))
                    has_issues = True
            
            elif param_location == 'path':
                # Path parameters should use camelCase or kebab-case
                if not self.camel_case_pattern.match(param_name) and not self.kebab_case_pattern.match(param_name):
                    self.report.add_issue(LintIssue(
                        severity='warning',
                        category='naming',
                        message=f"Path parameter '{param_name}' should use camelCase or kebab-case in {method} {path}",
                        path=f"{param_path}/name",
                        suggestion="Use camelCase (e.g., 'userId') or kebab-case (e.g., 'user-id')"
                    ))
                    has_issues = True
                
                # Path parameters must be required
                if not param.get('required', False):
                    self.report.add_issue(LintIssue(
                        severity='error',
                        category='parameters',
                        message=f"Path parameter '{param_name}' must be required in {method} {path}",
                        path=f"{param_path}/required",
                        suggestion="Set required: true for path parameters"
                    ))
                    has_issues = True
        
        return has_issues

    def _validate_request_body(self, path: str, method: str, request_body: Dict[str, Any]) -> bool:
        """Validate request body definition."""
        has_issues = False
        
        if 'content' not in request_body:
            self.report.add_issue(LintIssue(
                severity='error',
                category='request_body',
                message=f"Request body missing content for {method} {path}",
                path=f"/paths/{path}/{method.lower()}/requestBody/content",
                suggestion="Define content types for the request body"
            ))
            has_issues = True
        
        return has_issues

    def _validate_components_section(self) -> None:
        """Validate the components section."""
        if 'components' not in self.openapi_spec:
            self.report.add_issue(LintIssue(
                severity='info',
                category='structure',
                message="Missing components section",
                path="/components",
                suggestion="Consider defining reusable components (schemas, responses, parameters)"
            ))
            return
        
        components = self.openapi_spec['components']
        
        # Validate schemas
        if 'schemas' in components:
            self._validate_schemas(components['schemas'])

    def _validate_schemas(self, schemas: Dict[str, Any]) -> None:
        """Validate schema definitions."""
        for schema_name, schema in schemas.items():
            # Check schema naming (should be PascalCase)
            if not self.pascal_case_pattern.match(schema_name):
                self.report.add_issue(LintIssue(
                    severity='warning',
                    category='naming',
                    message=f"Schema name '{schema_name}' should use PascalCase",
                    path=f"/components/schemas/{schema_name}",
                    suggestion=f"Use PascalCase for schema names (e.g., 'UserProfile', 'OrderItem')"
                ))
            
            # Validate schema properties
            if isinstance(schema, dict) and 'properties' in schema:
                self._validate_schema_properties(schema_name, schema['properties'])

    def _validate_schema_properties(self, schema_name: str, properties: Dict[str, Any]) -> None:
        """Validate schema property naming."""
        for prop_name, prop_def in properties.items():
            # Properties should use camelCase
            if not self.camel_case_pattern.match(prop_name):
                self.report.add_issue(LintIssue(
                    severity='warning',
                    category='naming',
                    message=f"Property '{prop_name}' in schema '{schema_name}' should use camelCase",
                    path=f"/components/schemas/{schema_name}/properties/{prop_name}",
                    suggestion="Use camelCase for property names (e.g., 'firstName', 'createdAt')"
                ))

    def _validate_security_section(self) -> None:
        """Validate security definitions."""
        if 'security' not in self.openapi_spec and 'components' not in self.openapi_spec:
            self.report.add_issue(LintIssue(
                severity='warning',
                category='security',
                message="No security configuration found",
                path="/security",
                suggestion="Define security schemes and apply them to operations"
            ))

    def _validate_raw_endpoint_structure(self) -> None:
        """Validate structure of raw endpoint definitions."""
        if 'endpoints' not in self.raw_endpoints:
            self.report.add_issue(LintIssue(
                severity='error',
                category='structure',
                message="Missing 'endpoints' field in raw endpoint definition",
                path="/endpoints",
                suggestion="Provide an 'endpoints' object containing endpoint definitions"
            ))
            return
        
        endpoints = self.raw_endpoints['endpoints']
        self.report.total_endpoints = len(endpoints)

    def _lint_raw_endpoint(self, path: str, endpoint_data: Dict[str, Any]) -> None:
        """Lint individual raw endpoint definition."""
        # Validate path structure
        self._validate_path_structure(path)
        
        # Check for required fields
        if 'method' not in endpoint_data:
            self.report.add_issue(LintIssue(
                severity='error',
                category='structure',
                message=f"Missing method field for endpoint {path}",
                path=f"/endpoints/{path}/method",
                suggestion="Specify HTTP method (GET, POST, PUT, PATCH, DELETE)"
            ))
            return
        
        method = endpoint_data['method'].upper()
        if method not in self.http_methods:
            self.report.add_issue(LintIssue(
                severity='error',
                category='structure',
                message=f"Invalid HTTP method '{method}' for endpoint {path}",
                path=f"/endpoints/{path}/method",
                suggestion=f"Use valid HTTP methods: {', '.join(sorted(self.http_methods))}"
            ))

    def generate_json_report(self) -> str:
        """Generate JSON format report."""
        issues_by_severity = self.report.get_issues_by_severity()
        
        report_data = {
            "summary": {
                "total_endpoints": self.report.total_endpoints,
                "endpoints_with_issues": self.report.endpoints_with_issues,
                "total_issues": len(self.report.issues),
                "errors": len(issues_by_severity['error']),
                "warnings": len(issues_by_severity['warning']),
                "info": len(issues_by_severity['info']),
                "score": round(self.report.score, 2)
            },
            "issues": []
        }
        
        for issue in self.report.issues:
            report_data["issues"].append({
                "severity": issue.severity,
                "category": issue.category,
                "message": issue.message,
                "path": issue.path,
                "suggestion": issue.suggestion
            })
        
        return json.dumps(report_data, indent=2)

    def generate_text_report(self) -> str:
        """Generate human-readable text report."""
        issues_by_severity = self.report.get_issues_by_severity()
        
        report_lines = [
            "═══════════════════════════════════════════════════════════════",
            "                      API LINTING REPORT",
            "═══════════════════════════════════════════════════════════════",
            "",
            "SUMMARY:",
            f"  Total Endpoints: {self.report.total_endpoints}",
            f"  Endpoints with Issues: {self.report.endpoints_with_issues}",
            f"  Overall Score: {self.report.score:.1f}/100.0",
            "",
            "ISSUE BREAKDOWN:",
            f"  🔴 Errors: {len(issues_by_severity['error'])}",
            f"  🟡 Warnings: {len(issues_by_severity['warning'])}",
            f"  ℹ️  Info: {len(issues_by_severity['info'])}",
            "",
        ]
        
        if not self.report.issues:
            report_lines.extend([
                "🎉 Congratulations! No issues found in your API specification.",
                ""
            ])
        else:
            # Group issues by category
            issues_by_category = {}
            for issue in self.report.issues:
                if issue.category not in issues_by_category:
                    issues_by_category[issue.category] = []
                issues_by_category[issue.category].append(issue)
            
            for category, issues in issues_by_category.items():
                report_lines.append(f"{'═' * 60}")
                report_lines.append(f"CATEGORY: {category.upper().replace('_', ' ')}")
                report_lines.append(f"{'═' * 60}")
                
                for issue in issues:
                    severity_icon = {"error": "🔴", "warning": "🟡", "info": "ℹ️"}[issue.severity]
                    
                    report_lines.extend([
                        f"{severity_icon} {issue.severity.upper()}: {issue.message}",
                        f"   Path: {issue.path}",
                    ])
                    
                    if issue.suggestion:
                        report_lines.append(f"   💡 Suggestion: {issue.suggestion}")
                    
                    report_lines.append("")
        
        # Add scoring breakdown
        report_lines.extend([
            "═══════════════════════════════════════════════════════════════",
            "SCORING DETAILS:",
            "═══════════════════════════════════════════════════════════════",
            f"Base Score: 100.0",
            f"Errors Penalty: -{len(issues_by_severity['error']) * 10} (10 points per error)",
            f"Warnings Penalty: -{len(issues_by_severity['warning']) * 3} (3 points per warning)",
            f"Info Penalty: -{len(issues_by_severity['info']) * 1} (1 point per info)",
            f"Final Score: {self.report.score:.1f}/100.0",
            ""
        ])
        
        # Add recommendations based on score
        if self.report.score >= 90:
            report_lines.append("🏆 Excellent! Your API design follows best practices.")
        elif self.report.score >= 80:
            report_lines.append("✅ Good API design with minor areas for improvement.")
        elif self.report.score >= 70:
            report_lines.append("⚠️  Fair API design. Consider addressing warnings and errors.")
        elif self.report.score >= 50:
            report_lines.append("❌ Poor API design. Multiple issues need attention.")
        else:
            report_lines.append("🚨 Critical API design issues. Immediate attention required.")
        
        return "\n".join(report_lines)


# Embedded sample OpenAPI spec — intentionally imperfect (a verb in a URL, a
# snake_case property) so --sample produces a representative report.
SAMPLE_OPENAPI_SPEC = {
    "openapi": "3.0.0",
    "info": {"title": "Sample API", "version": "1.0.0"},
    "servers": [{"url": "https://api.example.com"}],
    "paths": {
        "/user-profiles/{userId}": {
            "get": {
                "summary": "Get a user profile",
                "responses": {"200": {"description": "OK"}, "404": {"description": "Not found"}},
                "parameters": [{"name": "userId", "in": "path", "required": True}],
            }
        },
        "/user-profiles/create": {
            "post": {
                "summary": "Create a user profile (verb-in-URL anti-pattern)",
                "responses": {"201": {"description": "Created"}},
            }
        },
    },
    "components": {
        "schemas": {
            "UserProfile": {"properties": {"first_name": {"type": "string"}}}
        }
    },
}


def main():
    """Main CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Analyze OpenAPI/Swagger specifications for REST conventions and best practices",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python api_linter.py openapi.json
  python api_linter.py --format json openapi.json > report.json
  python api_linter.py --raw-endpoints endpoints.json
  python api_linter.py --sample --format json
        """
    )

    parser.add_argument(
        'input_file',
        nargs='?',
        help='Input file: OpenAPI/Swagger JSON file or raw endpoints JSON'
    )

    parser.add_argument(
        '--sample',
        action='store_true',
        help='Lint an embedded sample OpenAPI spec (no input file needed)'
    )
    
    parser.add_argument(
        '--format',
        choices=['text', 'json'],
        default='text',
        help='Output format (default: text)'
    )
    
    parser.add_argument(
        '--raw-endpoints',
        action='store_true',
        help='Treat input as raw endpoint definitions instead of OpenAPI spec'
    )
    
    parser.add_argument(
        '--output',
        help='Output file (default: stdout)'
    )
    
    args = parser.parse_args()

    # Load input data — from the embedded sample or the input file
    if args.sample:
        input_data = SAMPLE_OPENAPI_SPEC
    else:
        if not args.input_file:
            parser.error("input_file is required (or use --sample)")
        try:
            with open(args.input_file, 'r') as f:
                input_data = json.load(f)
        except FileNotFoundError:
            print(f"Error: Input file '{args.input_file}' not found.", file=sys.stderr)
            return 1
        except json.JSONDecodeError as e:
            print(f"Error: Invalid JSON in '{args.input_file}': {e}", file=sys.stderr)
            return 1
    
    # Initialize linter and run analysis
    linter = APILinter()
    
    try:
        if args.raw_endpoints:
            report = linter.lint_raw_endpoints(input_data)
        else:
            report = linter.lint_openapi_spec(input_data)
    except Exception as e:
        print(f"Error during linting: {e}", file=sys.stderr)
        return 1
    
    # Generate report
    if args.format == 'json':
        output = linter.generate_json_report()
    else:
        output = linter.generate_text_report()
    
    # Write output
    if args.output:
        try:
            with open(args.output, 'w') as f:
                f.write(output)
            print(f"Report written to {args.output}")
        except IOError as e:
            print(f"Error writing to '{args.output}': {e}", file=sys.stderr)
            return 1
    else:
        print(output)
    
    # Return appropriate exit code
    error_count = len([i for i in report.issues if i.severity == 'error'])
    return 1 if error_count > 0 else 0


if __name__ == '__main__':
    sys.exit(main())

#!/usr/bin/env python3
"""
Breaking Change Detector - Compares API specification versions to identify breaking changes.

This script analyzes two versions of an API specification and detects potentially
breaking changes including:
- Removed endpoints
- Modified response structures
- Removed or renamed fields
- Field type changes
- New required fields
- HTTP status code changes
- Parameter changes

Generates detailed reports with migration guides for each breaking change.
"""

import argparse
import json
import sys
from typing import Any, Dict, List, Set, Optional, Tuple, Union
from dataclasses import dataclass, field
from enum import Enum


class ChangeType(Enum):
    """Types of API changes."""
    BREAKING = "breaking"
    POTENTIALLY_BREAKING = "potentially_breaking"
    NON_BREAKING = "non_breaking"
    ENHANCEMENT = "enhancement"


class ChangeSeverity(Enum):
    """Severity levels for changes."""
    CRITICAL = "critical"     # Will definitely break clients
    HIGH = "high"            # Likely to break some clients
    MEDIUM = "medium"        # May break clients depending on usage
    LOW = "low"             # Minor impact, unlikely to break clients
    INFO = "info"           # Informational, no breaking impact


@dataclass
class Change:
    """Represents a detected change between API versions."""
    change_type: ChangeType
    severity: ChangeSeverity
    category: str
    path: str
    message: str
    old_value: Any = None
    new_value: Any = None
    migration_guide: str = ""
    impact_description: str = ""
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert change to dictionary for JSON serialization."""
        return {
            "changeType": self.change_type.value,
            "severity": self.severity.value,
            "category": self.category,
            "path": self.path,
            "message": self.message,
            "oldValue": self.old_value,
            "newValue": self.new_value,
            "migrationGuide": self.migration_guide,
            "impactDescription": self.impact_description
        }


@dataclass
class ComparisonReport:
    """Complete comparison report between two API versions."""
    changes: List[Change] = field(default_factory=list)
    summary: Dict[str, int] = field(default_factory=dict)
    
    def add_change(self, change: Change) -> None:
        """Add a change to the report."""
        self.changes.append(change)
    
    def calculate_summary(self) -> None:
        """Calculate summary statistics."""
        self.summary = {
            "total_changes": len(self.changes),
            "breaking_changes": len([c for c in self.changes if c.change_type == ChangeType.BREAKING]),
            "potentially_breaking_changes": len([c for c in self.changes if c.change_type == ChangeType.POTENTIALLY_BREAKING]),
            "non_breaking_changes": len([c for c in self.changes if c.change_type == ChangeType.NON_BREAKING]),
            "enhancements": len([c for c in self.changes if c.change_type == ChangeType.ENHANCEMENT]),
            "critical_severity": len([c for c in self.changes if c.severity == ChangeSeverity.CRITICAL]),
            "high_severity": len([c for c in self.changes if c.severity == ChangeSeverity.HIGH]),
            "medium_severity": len([c for c in self.changes if c.severity == ChangeSeverity.MEDIUM]),
            "low_severity": len([c for c in self.changes if c.severity == ChangeSeverity.LOW]),
            "info_severity": len([c for c in self.changes if c.severity == ChangeSeverity.INFO])
        }
    
    def has_breaking_changes(self) -> bool:
        """Check if report contains any breaking changes."""
        return any(c.change_type in [ChangeType.BREAKING, ChangeType.POTENTIALLY_BREAKING] 
                  for c in self.changes)


class BreakingChangeDetector:
    """Main breaking change detection engine."""
    
    def __init__(self):
        self.report = ComparisonReport()
        self.old_spec: Optional[Dict] = None
        self.new_spec: Optional[Dict] = None
    
    def compare_specs(self, old_spec: Dict[str, Any], new_spec: Dict[str, Any]) -> ComparisonReport:
        """Compare two API specifications and detect changes."""
        self.old_spec = old_spec
        self.new_spec = new_spec
        self.report = ComparisonReport()
        
        # Compare different sections of the API specification
        self._compare_info_section()
        self._compare_servers_section()
        self._compare_paths_section()
        self._compare_components_section()
        self._compare_security_section()
        
        # Calculate summary statistics
        self.report.calculate_summary()
        
        return self.report
    
    def _compare_info_section(self) -> None:
        """Compare API info sections."""
        old_info = self.old_spec.get('info', {})
        new_info = self.new_spec.get('info', {})
        
        # Version comparison
        old_version = old_info.get('version', '')
        new_version = new_info.get('version', '')
        
        if old_version != new_version:
            self.report.add_change(Change(
                change_type=ChangeType.NON_BREAKING,
                severity=ChangeSeverity.INFO,
                category="versioning",
                path="/info/version",
                message=f"API version changed from '{old_version}' to '{new_version}'",
                old_value=old_version,
                new_value=new_version,
                impact_description="Version change indicates API evolution"
            ))
        
        # Title comparison
        old_title = old_info.get('title', '')
        new_title = new_info.get('title', '')
        
        if old_title != new_title:
            self.report.add_change(Change(
                change_type=ChangeType.NON_BREAKING,
                severity=ChangeSeverity.INFO,
                category="metadata",
                path="/info/title",
                message=f"API title changed from '{old_title}' to '{new_title}'",
                old_value=old_title,
                new_value=new_title,
                impact_description="Title change is cosmetic and doesn't affect functionality"
            ))
    
    def _compare_servers_section(self) -> None:
        """Compare server configurations."""
        old_servers = self.old_spec.get('servers', [])
        new_servers = self.new_spec.get('servers', [])
        
        old_urls = {server.get('url', '') for server in old_servers if isinstance(server, dict)}
        new_urls = {server.get('url', '') for server in new_servers if isinstance(server, dict)}
        
        # Removed servers
        removed_urls = old_urls - new_urls
        for url in removed_urls:
            self.report.add_change(Change(
                change_type=ChangeType.BREAKING,
                severity=ChangeSeverity.HIGH,
                category="servers",
                path="/servers",
                message=f"Server URL removed: {url}",
                old_value=url,
                new_value=None,
                migration_guide=f"Update client configurations to use alternative server URLs: {list(new_urls)}",
                impact_description="Clients configured to use removed server URL will fail to connect"
            ))
        
        # Added servers
        added_urls = new_urls - old_urls
        for url in added_urls:
            self.report.add_change(Change(
                change_type=ChangeType.ENHANCEMENT,
                severity=ChangeSeverity.INFO,
                category="servers",
                path="/servers",
                message=f"New server URL added: {url}",
                old_value=None,
                new_value=url,
                impact_description="New server option provides additional deployment flexibility"
            ))
    
    def _compare_paths_section(self) -> None:
        """Compare API paths and operations."""
        old_paths = self.old_spec.get('paths', {})
        new_paths = self.new_spec.get('paths', {})
        
        # Find removed, added, and modified paths
        old_path_set = set(old_paths.keys())
        new_path_set = set(new_paths.keys())
        
        removed_paths = old_path_set - new_path_set
        added_paths = new_path_set - old_path_set
        common_paths = old_path_set & new_path_set
        
        # Handle removed paths
        for path in removed_paths:
            old_operations = self._extract_operations(old_paths[path])
            for method in old_operations:
                self.report.add_change(Change(
                    change_type=ChangeType.BREAKING,
                    severity=ChangeSeverity.CRITICAL,
                    category="endpoints",
                    path=f"/paths{path}",
                    message=f"Endpoint removed: {method.upper()} {path}",
                    old_value=f"{method.upper()} {path}",
                    new_value=None,
                    migration_guide=self._generate_endpoint_removal_migration(path, method, new_paths),
                    impact_description="Clients using this endpoint will receive 404 errors"
                ))
        
        # Handle added paths
        for path in added_paths:
            new_operations = self._extract_operations(new_paths[path])
            for method in new_operations:
                self.report.add_change(Change(
                    change_type=ChangeType.ENHANCEMENT,
                    severity=ChangeSeverity.INFO,
                    category="endpoints",
                    path=f"/paths{path}",
                    message=f"New endpoint added: {method.upper()} {path}",
                    old_value=None,
                    new_value=f"{method.upper()} {path}",
                    impact_description="New functionality available to clients"
                ))
        
        # Handle modified paths
        for path in common_paths:
            self._compare_path_operations(path, old_paths[path], new_paths[path])
    
    def _extract_operations(self, path_object: Dict[str, Any]) -> List[str]:
        """Extract HTTP operations from a path object."""
        http_methods = {'get', 'post', 'put', 'patch', 'delete', 'head', 'options', 'trace'}
        return [method for method in path_object.keys() if method.lower() in http_methods]
    
    def _compare_path_operations(self, path: str, old_path_obj: Dict, new_path_obj: Dict) -> None:
        """Compare operations within a specific path."""
        old_operations = set(self._extract_operations(old_path_obj))
        new_operations = set(self._extract_operations(new_path_obj))
        
        # Removed operations
        removed_ops = old_operations - new_operations
        for method in removed_ops:
            self.report.add_change(Change(
                change_type=ChangeType.BREAKING,
                severity=ChangeSeverity.CRITICAL,
                category="endpoints",
                path=f"/paths{path}/{method}",
                message=f"HTTP method removed: {method.upper()} {path}",
                old_value=f"{method.upper()} {path}",
                new_value=None,
                migration_guide=self._generate_method_removal_migration(path, method, new_operations),
                impact_description="Clients using this method will receive 405 Method Not Allowed errors"
            ))
        
        # Added operations
        added_ops = new_operations - old_operations
        for method in added_ops:
            self.report.add_change(Change(
                change_type=ChangeType.ENHANCEMENT,
                severity=ChangeSeverity.INFO,
                category="endpoints",
                path=f"/paths{path}/{method}",
                message=f"New HTTP method added: {method.upper()} {path}",
                old_value=None,
                new_value=f"{method.upper()} {path}",
                impact_description="New method provides additional functionality for this resource"
            ))
        
        # Modified operations
        common_ops = old_operations & new_operations
        for method in common_ops:
            self._compare_operation_details(path, method, old_path_obj[method], new_path_obj[method])
    
    def _compare_operation_details(self, path: str, method: str, old_op: Dict, new_op: Dict) -> None:
        """Compare details of individual operations."""
        operation_path = f"/paths{path}/{method}"
        
        # Compare parameters
        self._compare_parameters(operation_path, old_op.get('parameters', []), new_op.get('parameters', []))
        
        # Compare request body
        self._compare_request_body(operation_path, old_op.get('requestBody'), new_op.get('requestBody'))
        
        # Compare responses
        self._compare_responses(operation_path, old_op.get('responses', {}), new_op.get('responses', {}))
        
        # Compare security requirements
        self._compare_security_requirements(operation_path, old_op.get('security'), new_op.get('security'))
    
    def _compare_parameters(self, base_path: str, old_params: List[Dict], new_params: List[Dict]) -> None:
        """Compare operation parameters."""
        # Create lookup dictionaries
        old_param_map = {(p.get('name'), p.get('in')): p for p in old_params}
        new_param_map = {(p.get('name'), p.get('in')): p for p in new_params}
        
        old_param_keys = set(old_param_map.keys())
        new_param_keys = set(new_param_map.keys())
        
        # Removed parameters
        removed_params = old_param_keys - new_param_keys
        for param_key in removed_params:
            name, location = param_key
            self.report.add_change(Change(
                change_type=ChangeType.BREAKING,
                severity=ChangeSeverity.HIGH,
                category="parameters",
                path=f"{base_path}/parameters",
                message=f"Parameter removed: {name} (in: {location})",
                old_value=old_param_map[param_key],
                new_value=None,
                migration_guide=f"Remove '{name}' parameter from {location} when calling this endpoint",
                impact_description="Clients sending this parameter may receive validation errors"
            ))
        
        # Added parameters
        added_params = new_param_keys - old_param_keys
        for param_key in added_params:
            name, location = param_key
            new_param = new_param_map[param_key]
            is_required = new_param.get('required', False)
            
            if is_required:
                self.report.add_change(Change(
                    change_type=ChangeType.BREAKING,
                    severity=ChangeSeverity.CRITICAL,
                    category="parameters",
                    path=f"{base_path}/parameters",
                    message=f"New required parameter added: {name} (in: {location})",
                    old_value=None,
                    new_value=new_param,
                    migration_guide=f"Add required '{name}' parameter to {location} when calling this endpoint",
                    impact_description="Clients not providing this parameter will receive 400 Bad Request errors"
                ))
            else:
                self.report.add_change(Change(
                    change_type=ChangeType.NON_BREAKING,
                    severity=ChangeSeverity.INFO,
                    category="parameters",
                    path=f"{base_path}/parameters",
                    message=f"New optional parameter added: {name} (in: {location})",
                    old_value=None,
                    new_value=new_param,
                    impact_description="Optional parameter provides additional functionality"
                ))
        
        # Modified parameters
        common_params = old_param_keys & new_param_keys
        for param_key in common_params:
            name, location = param_key
            old_param = old_param_map[param_key]
            new_param = new_param_map[param_key]
            self._compare_parameter_details(base_path, name, location, old_param, new_param)
    
    def _compare_parameter_details(self, base_path: str, name: str, location: str, 
                                 old_param: Dict, new_param: Dict) -> None:
        """Compare individual parameter details."""
        param_path = f"{base_path}/parameters/{name}"
        
        # Required status change
        old_required = old_param.get('required', False)
        new_required = new_param.get('required', False)
        
        if old_required != new_required:
            if new_required:
                self.report.add_change(Change(
                    change_type=ChangeType.BREAKING,
                    severity=ChangeSeverity.HIGH,
                    category="parameters",
                    path=param_path,
                    message=f"Parameter '{name}' is now required (was optional)",
                    old_value=old_required,
                    new_value=new_required,
                    migration_guide=f"Ensure '{name}' parameter is always provided when calling this endpoint",
                    impact_description="Clients not providing this parameter will receive validation errors"
                ))
            else:
                self.report.add_change(Change(
                    change_type=ChangeType.NON_BREAKING,
                    severity=ChangeSeverity.INFO,
                    category="parameters",
                    path=param_path,
                    message=f"Parameter '{name}' is now optional (was required)",
                    old_value=old_required,
                    new_value=new_required,
                    impact_description="Parameter is now optional, providing more flexibility to clients"
                ))
        
        # Schema/type changes
        old_schema = old_param.get('schema', {})
        new_schema = new_param.get('schema', {})
        
        if old_schema != new_schema:
            self._compare_schemas(param_path, old_schema, new_schema, f"parameter '{name}'")
    
    def _compare_request_body(self, base_path: str, old_body: Optional[Dict], new_body: Optional[Dict]) -> None:
        """Compare request body specifications."""
        body_path = f"{base_path}/requestBody"
        
        # Request body added
        if old_body is None and new_body is not None:
            is_required = new_body.get('required', False)
            if is_required:
                self.report.add_change(Change(
                    change_type=ChangeType.BREAKING,
                    severity=ChangeSeverity.HIGH,
                    category="request_body",
                    path=body_path,
                    message="Required request body added",
                    old_value=None,
                    new_value=new_body,
                    migration_guide="Include request body with appropriate content type when calling this endpoint",
                    impact_description="Clients not providing request body will receive validation errors"
                ))
            else:
                self.report.add_change(Change(
                    change_type=ChangeType.NON_BREAKING,
                    severity=ChangeSeverity.INFO,
                    category="request_body",
                    path=body_path,
                    message="Optional request body added",
                    old_value=None,
                    new_value=new_body,
                    impact_description="Optional request body provides additional functionality"
                ))
        
        # Request body removed
        elif old_body is not None and new_body is None:
            self.report.add_change(Change(
                change_type=ChangeType.BREAKING,
                severity=ChangeSeverity.HIGH,
                category="request_body",
                path=body_path,
                message="Request body removed",
                old_value=old_body,
                new_value=None,
                migration_guide="Remove request body when calling this endpoint",
                impact_description="Clients sending request body may receive validation errors"
            ))
        
        # Request body modified
        elif old_body is not None and new_body is not None:
            self._compare_request_body_details(body_path, old_body, new_body)
    
    def _compare_request_body_details(self, base_path: str, old_body: Dict, new_body: Dict) -> None:
        """Compare request body details."""
        # Required status change
        old_required = old_body.get('required', False)
        new_required = new_body.get('required', False)
        
        if old_required != new_required:
            if new_required:
                self.report.add_change(Change(
                    change_type=ChangeType.BREAKING,
                    severity=ChangeSeverity.HIGH,
                    category="request_body",
                    path=base_path,
                    message="Request body is now required (was optional)",
                    old_value=old_required,
                    new_value=new_required,
                    migration_guide="Always include request body when calling this endpoint",
                    impact_description="Clients not providing request body will receive validation errors"
                ))
            else:
                self.report.add_change(Change(
                    change_type=ChangeType.NON_BREAKING,
                    severity=ChangeSeverity.INFO,
                    category="request_body",
                    path=base_path,
                    message="Request body is now optional (was required)",
                    old_value=old_required,
                    new_value=new_required,
                    impact_description="Request body is now optional, providing more flexibility"
                ))
        
        # Content type changes
        old_content = old_body.get('content', {})
        new_content = new_body.get('content', {})
        self._compare_content_types(base_path, old_content, new_content, "request body")
    
    def _compare_responses(self, base_path: str, old_responses: Dict, new_responses: Dict) -> None:
        """Compare response specifications."""
        responses_path = f"{base_path}/responses"
        
        old_status_codes = set(old_responses.keys())
        new_status_codes = set(new_responses.keys())
        
        # Removed status codes
        removed_codes = old_status_codes - new_status_codes
        for code in removed_codes:
            self.report.add_change(Change(
                change_type=ChangeType.BREAKING,
                severity=ChangeSeverity.HIGH,
                category="responses",
                path=f"{responses_path}/{code}",
                message=f"Response status code {code} removed",
                old_value=old_responses[code],
                new_value=None,
                migration_guide=f"Handle alternative status codes: {list(new_status_codes)}",
                impact_description=f"Clients expecting status code {code} need to handle different responses"
            ))
        
        # Added status codes
        added_codes = new_status_codes - old_status_codes
        for code in added_codes:
            self.report.add_change(Change(
                change_type=ChangeType.NON_BREAKING,
                severity=ChangeSeverity.INFO,
                category="responses",
                path=f"{responses_path}/{code}",
                message=f"New response status code {code} added",
                old_value=None,
                new_value=new_responses[code],
                impact_description="New status code provides more specific response information"
            ))
        
        # Modified responses
        common_codes = old_status_codes & new_status_codes
        for code in common_codes:
            self._compare_response_details(responses_path, code, old_responses[code], new_responses[code])
    
    def _compare_response_details(self, base_path: str, status_code: str, 
                                old_response: Dict, new_response: Dict) -> None:
        """Compare individual response details."""
        response_path = f"{base_path}/{status_code}"
        
        # Compare content types and schemas
        old_content = old_response.get('content', {})
        new_content = new_response.get('content', {})
        
        self._compare_content_types(response_path, old_content, new_content, f"response {status_code}")
    
    def _compare_content_types(self, base_path: str, old_content: Dict, new_content: Dict, context: str) -> None:
        """Compare content types and their schemas."""
        old_types = set(old_content.keys())
        new_types = set(new_content.keys())
        
        # Removed content types
        removed_types = old_types - new_types
        for content_type in removed_types:
            self.report.add_change(Change(
                change_type=ChangeType.BREAKING,
                severity=ChangeSeverity.HIGH,
                category="content_types",
                path=f"{base_path}/content",
                message=f"Content type '{content_type}' removed from {context}",
                old_value=content_type,
                new_value=None,
                migration_guide=f"Use alternative content types: {list(new_types)}",
                impact_description=f"Clients expecting '{content_type}' need to handle different formats"
            ))
        
        # Added content types
        added_types = new_types - old_types
        for content_type in added_types:
            self.report.add_change(Change(
                change_type=ChangeType.ENHANCEMENT,
                severity=ChangeSeverity.INFO,
                category="content_types",
                path=f"{base_path}/content",
                message=f"New content type '{content_type}' added to {context}",
                old_value=None,
                new_value=content_type,
                impact_description=f"Additional format option available for {context}"
            ))
        
        # Modified schemas for common content types
        common_types = old_types & new_types
        for content_type in common_types:
            old_media = old_content[content_type]
            new_media = new_content[content_type]
            
            old_schema = old_media.get('schema', {})
            new_schema = new_media.get('schema', {})
            
            if old_schema != new_schema:
                schema_path = f"{base_path}/content/{content_type}/schema"
                self._compare_schemas(schema_path, old_schema, new_schema, f"{context} ({content_type})")
    
    def _compare_schemas(self, base_path: str, old_schema: Dict, new_schema: Dict, context: str) -> None:
        """Compare schema definitions."""
        # Type changes
        old_type = old_schema.get('type')
        new_type = new_schema.get('type')
        
        if old_type != new_type and old_type is not None and new_type is not None:
            self.report.add_change(Change(
                change_type=ChangeType.BREAKING,
                severity=ChangeSeverity.CRITICAL,
                category="schema",
                path=base_path,
                message=f"Schema type changed from '{old_type}' to '{new_type}' for {context}",
                old_value=old_type,
                new_value=new_type,
                migration_guide=f"Update client code to handle {new_type} instead of {old_type}",
                impact_description="Type change will break client parsing and validation"
            ))
        
        # Property changes for object types
        if old_schema.get('type') == 'object' and new_schema.get('type') == 'object':
            self._compare_object_properties(base_path, old_schema, new_schema, context)
        
        # Array item changes
        if old_schema.get('type') == 'array' and new_schema.get('type') == 'array':
            old_items = old_schema.get('items', {})
            new_items = new_schema.get('items', {})
            if old_items != new_items:
                self._compare_schemas(f"{base_path}/items", old_items, new_items, f"{context} items")
    
    def _compare_object_properties(self, base_path: str, old_schema: Dict, new_schema: Dict, context: str) -> None:
        """Compare object schema properties."""
        old_props = old_schema.get('properties', {})
        new_props = new_schema.get('properties', {})
        old_required = set(old_schema.get('required', []))
        new_required = set(new_schema.get('required', []))
        
        old_prop_names = set(old_props.keys())
        new_prop_names = set(new_props.keys())
        
        # Removed properties
        removed_props = old_prop_names - new_prop_names
        for prop_name in removed_props:
            severity = ChangeSeverity.CRITICAL if prop_name in old_required else ChangeSeverity.HIGH
            self.report.add_change(Change(
                change_type=ChangeType.BREAKING,
                severity=severity,
                category="schema",
                path=f"{base_path}/properties",
                message=f"Property '{prop_name}' removed from {context}",
                old_value=old_props[prop_name],
                new_value=None,
                migration_guide=f"Remove references to '{prop_name}' property in client code",
                impact_description="Clients expecting this property will receive incomplete data"
            ))
        
        # Added properties
        added_props = new_prop_names - old_prop_names
        for prop_name in added_props:
            if prop_name in new_required:
                # This is handled separately in required field changes
                pass
            else:
                self.report.add_change(Change(
                    change_type=ChangeType.NON_BREAKING,
                    severity=ChangeSeverity.INFO,
                    category="schema",
                    path=f"{base_path}/properties",
                    message=f"New optional property '{prop_name}' added to {context}",
                    old_value=None,
                    new_value=new_props[prop_name],
                    impact_description="New property provides additional data without breaking existing clients"
                ))
        
        # Required field changes
        added_required = new_required - old_required
        removed_required = old_required - new_required
        
        for prop_name in added_required:
            self.report.add_change(Change(
                change_type=ChangeType.BREAKING,
                severity=ChangeSeverity.CRITICAL,
                category="schema",
                path=f"{base_path}/properties",
                message=f"Property '{prop_name}' is now required in {context}",
                old_value=False,
                new_value=True,
                migration_guide=f"Ensure '{prop_name}' is always provided when sending {context}",
                impact_description="Clients not providing this property will receive validation errors"
            ))
        
        for prop_name in removed_required:
            self.report.add_change(Change(
                change_type=ChangeType.NON_BREAKING,
                severity=ChangeSeverity.INFO,
                category="schema",
                path=f"{base_path}/properties",
                message=f"Property '{prop_name}' is no longer required in {context}",
                old_value=True,
                new_value=False,
                impact_description="Property is now optional, providing more flexibility"
            ))
        
        # Modified properties
        common_props = old_prop_names & new_prop_names
        for prop_name in common_props:
            old_prop = old_props[prop_name]
            new_prop = new_props[prop_name]
            if old_prop != new_prop:
                self._compare_schemas(f"{base_path}/properties/{prop_name}", 
                                    old_prop, new_prop, f"{context}.{prop_name}")
    
    def _compare_security_requirements(self, base_path: str, old_security: Optional[List], 
                                     new_security: Optional[List]) -> None:
        """Compare security requirements."""
        # Simplified security comparison - could be expanded
        if old_security != new_security:
            severity = ChangeSeverity.HIGH if new_security else ChangeSeverity.CRITICAL
            change_type = ChangeType.BREAKING
            
            if old_security is None and new_security is not None:
                message = "Security requirements added"
                migration_guide = "Ensure proper authentication/authorization when calling this endpoint"
                impact = "Endpoint now requires authentication"
            elif old_security is not None and new_security is None:
                message = "Security requirements removed"
                migration_guide = "Authentication is no longer required for this endpoint"
                impact = "Endpoint is now publicly accessible"
                severity = ChangeSeverity.MEDIUM  # Less severe, more permissive
            else:
                message = "Security requirements modified"
                migration_guide = "Update authentication/authorization method for this endpoint"
                impact = "Different authentication method required"
            
            self.report.add_change(Change(
                change_type=change_type,
                severity=severity,
                category="security",
                path=f"{base_path}/security",
                message=message,
                old_value=old_security,
                new_value=new_security,
                migration_guide=migration_guide,
                impact_description=impact
            ))
    
    def _compare_components_section(self) -> None:
        """Compare components sections."""
        old_components = self.old_spec.get('components', {})
        new_components = self.new_spec.get('components', {})
        
        # Compare schemas
        old_schemas = old_components.get('schemas', {})
        new_schemas = new_components.get('schemas', {})
        
        old_schema_names = set(old_schemas.keys())
        new_schema_names = set(new_schemas.keys())
        
        # Removed schemas
        removed_schemas = old_schema_names - new_schema_names
        for schema_name in removed_schemas:
            self.report.add_change(Change(
                change_type=ChangeType.BREAKING,
                severity=ChangeSeverity.HIGH,
                category="components",
                path=f"/components/schemas/{schema_name}",
                message=f"Schema '{schema_name}' removed from components",
                old_value=old_schemas[schema_name],
                new_value=None,
                migration_guide=f"Remove references to schema '{schema_name}' or use alternative schemas",
                impact_description="References to this schema will fail validation"
            ))
        
        # Added schemas
        added_schemas = new_schema_names - old_schema_names
        for schema_name in added_schemas:
            self.report.add_change(Change(
                change_type=ChangeType.ENHANCEMENT,
                severity=ChangeSeverity.INFO,
                category="components",
                path=f"/components/schemas/{schema_name}",
                message=f"New schema '{schema_name}' added to components",
                old_value=None,
                new_value=new_schemas[schema_name],
                impact_description="New reusable schema available"
            ))
        
        # Modified schemas
        common_schemas = old_schema_names & new_schema_names
        for schema_name in common_schemas:
            old_schema = old_schemas[schema_name]
            new_schema = new_schemas[schema_name]
            if old_schema != new_schema:
                self._compare_schemas(f"/components/schemas/{schema_name}", 
                                    old_schema, new_schema, f"schema '{schema_name}'")
    
    def _compare_security_section(self) -> None:
        """Compare security definitions."""
        old_security_schemes = self.old_spec.get('components', {}).get('securitySchemes', {})
        new_security_schemes = self.new_spec.get('components', {}).get('securitySchemes', {})
        
        if old_security_schemes != new_security_schemes:
            # Simplified comparison - could be more detailed
            self.report.add_change(Change(
                change_type=ChangeType.POTENTIALLY_BREAKING,
                severity=ChangeSeverity.MEDIUM,
                category="security",
                path="/components/securitySchemes",
                message="Security scheme definitions changed",
                old_value=old_security_schemes,
                new_value=new_security_schemes,
                migration_guide="Review authentication implementation for compatibility with new security schemes",
                impact_description="Authentication mechanisms may have changed"
            ))
    
    def _generate_endpoint_removal_migration(self, removed_path: str, method: str, 
                                           remaining_paths: Dict[str, Any]) -> str:
        """Generate migration guide for removed endpoints."""
        # Look for similar endpoints
        similar_paths = []
        path_segments = removed_path.strip('/').split('/')
        
        for existing_path in remaining_paths.keys():
            existing_segments = existing_path.strip('/').split('/')
            if len(existing_segments) == len(path_segments):
                # Check similarity
                similarity = sum(1 for i, seg in enumerate(path_segments) 
                               if i < len(existing_segments) and seg == existing_segments[i])
                if similarity >= len(path_segments) * 0.5:  # At least 50% similar
                    similar_paths.append(existing_path)
        
        if similar_paths:
            return f"Consider using alternative endpoints: {', '.join(similar_paths[:3])}"
        else:
            return "No direct replacement available. Review API documentation for alternative approaches."
    
    def _generate_method_removal_migration(self, path: str, removed_method: str, 
                                         remaining_methods: Set[str]) -> str:
        """Generate migration guide for removed HTTP methods."""
        method_alternatives = {
            'get': ['head'],
            'post': ['put', 'patch'],
            'put': ['post', 'patch'],
            'patch': ['put', 'post'],
            'delete': []
        }
        
        alternatives = []
        for alt_method in method_alternatives.get(removed_method.lower(), []):
            if alt_method in remaining_methods:
                alternatives.append(alt_method.upper())
        
        if alternatives:
            return f"Use alternative methods: {', '.join(alternatives)}"
        else:
            return f"No alternative HTTP methods available for {path}"
    
    def generate_json_report(self) -> str:
        """Generate JSON format report."""
        report_data = {
            "summary": self.report.summary,
            "hasBreakingChanges": self.report.has_breaking_changes(),
            "changes": [change.to_dict() for change in self.report.changes]
        }
        
        return json.dumps(report_data, indent=2)
    
    def generate_text_report(self) -> str:
        """Generate human-readable text report."""
        lines = [
            "═══════════════════════════════════════════════════════════════",
            "                  BREAKING CHANGE ANALYSIS REPORT",
            "═══════════════════════════════════════════════════════════════",
            "",
            "SUMMARY:",
            f"  Total Changes: {self.report.summary.get('total_changes', 0)}",
            f"  🔴 Breaking Changes: {self.report.summary.get('breaking_changes', 0)}",
            f"  🟡 Potentially Breaking: {self.report.summary.get('potentially_breaking_changes', 0)}",
            f"  🟢 Non-Breaking Changes: {self.report.summary.get('non_breaking_changes', 0)}",
            f"  ✨ Enhancements: {self.report.summary.get('enhancements', 0)}",
            "",
            "SEVERITY BREAKDOWN:",
            f"  🚨 Critical: {self.report.summary.get('critical_severity', 0)}",
            f"  ⚠️  High: {self.report.summary.get('high_severity', 0)}",
            f"  ⚪ Medium: {self.report.summary.get('medium_severity', 0)}",
            f"  🔵 Low: {self.report.summary.get('low_severity', 0)}",
            f"  ℹ️  Info: {self.report.summary.get('info_severity', 0)}",
            ""
        ]
        
        if not self.report.changes:
            lines.extend([
                "🎉 No changes detected between the API versions!",
                ""
            ])
        else:
            # Group changes by type and severity
            breaking_changes = [c for c in self.report.changes if c.change_type == ChangeType.BREAKING]
            potentially_breaking = [c for c in self.report.changes if c.change_type == ChangeType.POTENTIALLY_BREAKING]
            non_breaking = [c for c in self.report.changes if c.change_type == ChangeType.NON_BREAKING]
            enhancements = [c for c in self.report.changes if c.change_type == ChangeType.ENHANCEMENT]
            
            # Breaking changes section
            if breaking_changes:
                lines.extend([
                    "🔴 BREAKING CHANGES:",
                    "═" * 60
                ])
                for change in sorted(breaking_changes, key=lambda x: x.severity.value):
                    self._add_change_to_report(lines, change)
                lines.append("")
            
            # Potentially breaking changes section
            if potentially_breaking:
                lines.extend([
                    "🟡 POTENTIALLY BREAKING CHANGES:",
                    "═" * 60
                ])
                for change in sorted(potentially_breaking, key=lambda x: x.severity.value):
                    self._add_change_to_report(lines, change)
                lines.append("")
            
            # Non-breaking changes section
            if non_breaking:
                lines.extend([
                    "🟢 NON-BREAKING CHANGES:",
                    "═" * 60
                ])
                for change in non_breaking:
                    self._add_change_to_report(lines, change)
                lines.append("")
            
            # Enhancements section
            if enhancements:
                lines.extend([
                    "✨ ENHANCEMENTS:",
                    "═" * 60
                ])
                for change in enhancements:
                    self._add_change_to_report(lines, change)
                lines.append("")
        
        # Add overall assessment
        lines.extend([
            "═══════════════════════════════════════════════════════════════",
            "OVERALL ASSESSMENT:",
            "═══════════════════════════════════════════════════════════════"
        ])
        
        if self.report.has_breaking_changes():
            breaking_count = self.report.summary.get('breaking_changes', 0)
            potentially_breaking_count = self.report.summary.get('potentially_breaking_changes', 0)
            
            if breaking_count > 0:
                lines.extend([
                    f"⛔ MAJOR VERSION BUMP REQUIRED",
                    f"   This API version contains {breaking_count} breaking changes that will",
                    f"   definitely break existing clients. A major version bump is required.",
                    ""
                ])
            elif potentially_breaking_count > 0:
                lines.extend([
                    f"⚠️  MINOR VERSION BUMP RECOMMENDED",
                    f"   This API version contains {potentially_breaking_count} potentially breaking",
                    f"   changes. Consider a minor version bump and communicate changes to clients.",
                    ""
                ])
        else:
            lines.extend([
                "✅ PATCH VERSION BUMP ACCEPTABLE",
                "   No breaking changes detected. This version is backward compatible",
                "   with existing clients.",
                ""
            ])
        
        return "\n".join(lines)
    
    def _add_change_to_report(self, lines: List[str], change: Change) -> None:
        """Add a change to the text report."""
        severity_icons = {
            ChangeSeverity.CRITICAL: "🚨",
            ChangeSeverity.HIGH: "⚠️ ",
            ChangeSeverity.MEDIUM: "⚪",
            ChangeSeverity.LOW: "🔵",
            ChangeSeverity.INFO: "ℹ️ "
        }
        
        icon = severity_icons.get(change.severity, "❓")
        
        lines.extend([
            f"{icon} {change.severity.value.upper()}: {change.message}",
            f"   Path: {change.path}",
            f"   Category: {change.category}"
        ])
        
        if change.impact_description:
            lines.append(f"   Impact: {change.impact_description}")
        
        if change.migration_guide:
            lines.append(f"   💡 Migration: {change.migration_guide}")
        
        lines.append("")


def main():
    """Main CLI entry point."""
    parser = argparse.ArgumentParser(
        description="Compare API specification versions to detect breaking changes",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python breaking_change_detector.py v1.json v2.json
  python breaking_change_detector.py --format json v1.json v2.json > changes.json
  python breaking_change_detector.py --output report.txt v1.json v2.json
        """
    )
    
    parser.add_argument(
        'old_spec',
        help='Old API specification file (JSON format)'
    )
    
    parser.add_argument(
        'new_spec',
        help='New API specification file (JSON format)'
    )
    
    parser.add_argument(
        '--format',
        choices=['text', 'json'],
        default='text',
        help='Output format (default: text)'
    )
    
    parser.add_argument(
        '--output',
        help='Output file (default: stdout)'
    )
    
    parser.add_argument(
        '--exit-on-breaking',
        action='store_true',
        help='Exit with code 1 if breaking changes are detected'
    )
    
    args = parser.parse_args()
    
    # Load specification files
    try:
        with open(args.old_spec, 'r') as f:
            old_spec = json.load(f)
    except FileNotFoundError:
        print(f"Error: Old specification file '{args.old_spec}' not found.", file=sys.stderr)
        return 1
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON in '{args.old_spec}': {e}", file=sys.stderr)
        return 1
    
    try:
        with open(args.new_spec, 'r') as f:
            new_spec = json.load(f)
    except FileNotFoundError:
        print(f"Error: New specification file '{args.new_spec}' not found.", file=sys.stderr)
        return 1
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON in '{args.new_spec}': {e}", file=sys.stderr)
        return 1
    
    # Initialize detector and compare specifications
    detector = BreakingChangeDetector()
    
    try:
        report = detector.compare_specs(old_spec, new_spec)
    except Exception as e:
        print(f"Error during comparison: {e}", file=sys.stderr)
        return 1
    
    # Generate report
    if args.format == 'json':
        output = detector.generate_json_report()
    else:
        output = detector.generate_text_report()
    
    # Write output
    if args.output:
        try:
            with open(args.output, 'w') as f:
                f.write(output)
            print(f"Breaking change report written to {args.output}")
        except IOError as e:
            print(f"Error writing to '{args.output}': {e}", file=sys.stderr)
            return 1
    else:
        print(output)
    
    # Exit with appropriate code
    if args.exit_on_breaking and report.has_breaking_changes():
        return 1
    
    return 0


if __name__ == '__main__':
    sys.exit(main())