Method search_urls

backend/app/services/scrapegraph.py:79–163

Search for URLs using ScrapeGraphAI SearchScraper.

(self, query: str, max_results: int = 5)

    
    
    async def search_urls(self, query: str, max_results: int = 5) -> List[Dict[str, str]]:
        """Search for URLs using ScrapeGraphAI SearchScraper."""
        try:
            async with AsyncClient(api_key=self.api_key) as client:
                # Define schema for structured output
                from pydantic import BaseModel, Field
                from typing import List
                
                class WebsiteInfo(BaseModel):
                    name: str = Field(description="Name of the website")
                    url: str = Field(description="Full URL of the website")
                    description: str = Field(description="Brief description of what the website offers")
                
                class SearchResults(BaseModel):
                    websites: List[WebsiteInfo] = Field(description="List of relevant websites")
                
                # Use searchscraper to find information about the topic
                search_prompt = f"Find the top {max_results} most relevant websites and their URLs for: {query}. Return a list of websites with their URLs and descriptions."
                
                try:
                    # Use searchscraper to find URLs - don't use schema for now
                    result = await client.searchscraper(
                        user_prompt=search_prompt
                    )
                    
                    # Log the result to understand its structure
                    logger.info(f"SearchScraper raw result type: {type(result)}")
                    logger.info(f"SearchScraper raw result: {result}")
                    
                    urls = []
                    
                    # Handle the response - it's a dict when using searchscraper
                    if isinstance(result, dict):
                        # Extract from the result field
                        if 'result' in result and isinstance(result['result'], dict):
                            websites_data = result['result'].get('websites', [])
                            
                            for website in websites_data:
                                if isinstance(website, dict) and 'url' in website:
                                    urls.append({
                                        'url': website['url'],
                                        'description': website.get('description', website.get('name', f'Found via search for: {query}'))
                                    })
                        
                        # Also add reference URLs if they're different
                        if 'reference_urls' in result and result['reference_urls']:
                            for ref_url in result['reference_urls']:
                                if not any(u['url'] == ref_url for u in urls):
                                    urls.append({
                                        'url': ref_url,
                                        'description': f'Reference source for: {query}'
                                    })
                    
                    logger.info(f"Final URLs extracted: {urls}")
                    return urls
                    
                except AttributeError as e:
                    # If searchscraper doesn't exist, try alternative approach