Comprehensive comparison for Search technology in Software Development applications

import boto3
import json
from typing import Dict, List, Optional
from datetime import datetime
import logging
from botocore.exceptions import ClientError

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

class CodeSearchService:
    """
    Production-ready AWS CloudSearch service for searching code repositories,
    documentation, and technical resources in a software development context.
    """
    
    def __init__(self, search_endpoint: str, document_endpoint: str):
        self.search_endpoint = search_endpoint
        self.document_endpoint = document_endpoint
        self.cloudsearch_domain = boto3.client('cloudsearchdomain',
                                               endpoint_url=search_endpoint)
        self.cloudsearch_doc = boto3.client('cloudsearchdomain',
                                            endpoint_url=document_endpoint)
    
    def search_code_repositories(self, query: str, filters: Optional[Dict] = None,
                                 page: int = 0, size: int = 10) -> Dict:
        """
        Search code repositories with advanced filtering and pagination.
        
        Args:
            query: Search query string
            filters: Optional filters (language, stars, last_updated)
            page: Page number for pagination
            size: Results per page
        
        Returns:
            Dictionary containing search results and metadata
        """
        try:
            # Build query with structured query syntax
            query_options = {
                'query': query,
                'queryParser': 'structured',
                'size': size,
                'start': page * size,
                'return': '_all_fields',
                'sort': 'stars desc'
            }
            
            # Add filter query if filters provided
            if filters:
                filter_expressions = []
                if filters.get('language'):
                    filter_expressions.append(f"language:'{filters['language']}'")
                if filters.get('min_stars'):
                    filter_expressions.append(f"stars:>={filters['min_stars']}")
                if filters.get('updated_after'):
                    filter_expressions.append(f"last_updated:>='{filters['updated_after']}'")
                
                if filter_expressions:
                    query_options['filterQuery'] = ' AND '.join(filter_expressions)
            
            # Execute search
            response = self.cloudsearch_domain.search(**query_options)
            
            # Process and return results
            return {
                'success': True,
                'total_results': response['hits']['found'],
                'results': [
                    {
                        'id': hit['id'],
                        'fields': hit['fields'],
                        'score': hit.get('_score', 0)
                    }
                    for hit in response['hits']['hit']
                ],
                'page': page,
                'page_size': size,
                'query_time_ms': response['status']['timems']
            }
            
        except ClientError as e:
            logger.error(f"CloudSearch error: {e.response['Error']['Message']}")
            return {
                'success': False,
                'error': 'Search service unavailable',
                'error_code': e.response['Error']['Code']
            }
        except Exception as e:
            logger.error(f"Unexpected error during search: {str(e)}")
            return {
                'success': False,
                'error': 'Internal search error'
            }
    
    def index_repository(self, repo_data: Dict) -> bool:
        """
        Index a code repository into CloudSearch.
        
        Args:
            repo_data: Repository metadata to index
        
        Returns:
            Boolean indicating success
        """
        try:
            document = {
                'type': 'add',
                'id': repo_data['id'],
                'fields': {
                    'name': repo_data['name'],
                    'description': repo_data.get('description', ''),
                    'language': repo_data.get('language', 'Unknown'),
                    'stars': repo_data.get('stars', 0),
                    'forks': repo_data.get('forks', 0),
                    'last_updated': repo_data.get('last_updated', datetime.utcnow().isoformat()),
                    'topics': repo_data.get('topics', []),
                    'owner': repo_data.get('owner', ''),
                    'url': repo_data.get('url', '')
                }
            }
            
            response = self.cloudsearch_doc.upload_documents(
                documents=json.dumps([document]),
                contentType='application/json'
            )
            
            if response['status'] == 'success':
                logger.info(f"Successfully indexed repository: {repo_data['id']}")
                return True
            else:
                logger.error(f"Failed to index repository: {response}")
                return False
                
        except Exception as e:
            logger.error(f"Error indexing repository: {str(e)}")
            return False