Create websearch_logic.py

websearch_logic.py

@@ -0,0 +1,281 @@
+import requests
+from bs4 import BeautifulSoup, Comment
+import logging
+import os
+import re
+from duckduckgo_search import DDGS
+from googlesearch import search as google_search_lib # Correct import
+
+logger = logging.getLogger(__name__) 
+
+DEFAULT_TIMEOUT = 10
+MAX_CONTENT_LENGTH_PER_URL = 3000
+MAX_TOTAL_SCRAPED_CONTENT = 9000
+
+# scrape_url function remains the same (as the last version with BeautifulSoup.Comment fix)
+def scrape_url(url_to_scrape, query_filter=None):
+    try:
+        logger.debug(f"SCRAPER_MODULE: Scraping URL: {url_to_scrape}")
+        headers = {
+            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36',
+            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
+            'Accept-Language': 'en-US,en;q=0.9',
+            'Referer': 'https://www.google.com/'
+        }
+        response = requests.get(url_to_scrape, headers=headers, timeout=DEFAULT_TIMEOUT, allow_redirects=True)
+        response.raise_for_status()
+        content_type = response.headers.get('content-type', '').lower()
+        if 'html' not in content_type:
+            logger.info(f"SCRAPER_MODULE: Skipping non-HTML: {url_to_scrape} (type: {content_type})")
+            return {"url": url_to_scrape, "title": url_to_scrape, "error": f"Non-HTML: {content_type}"}
+
+        soup = BeautifulSoup(response.content, 'html.parser')
+        for element_type in ["script", "style", "nav", "footer", "aside", "form", "iframe", "noscript", "header", "menu", "button", "figure", "figcaption", "link", "meta", ".sidebar", ".ad", ".advertisement", ".popup", ".modal", ".share", ".social", ".related-posts", ".comments-area", ".site-footer", ".site-header", ".widget", ".cookie-banner", ".gdpr", "dialog"]:
+            for element in soup.select(element_type): element.decompose()
+        for comment_node in soup.find_all(string=lambda text: isinstance(text, Comment)): # Corrected
+            comment_node.extract()
+        
+        main_content_selectors = ['main', 'article', '.main-content', '.entry-content', '.post-content', '.td-post-content', '.page-content', 'div[role="main"]', 'div[class*="content"]', 'div[class*="article"]', 'div[class*="post"]', 'div[id*="content"]', 'div[id*="main"]']
+        content_area = next((soup.select_one(s) for s in main_content_selectors if soup.select_one(s)), soup.body or soup)
+        
+        text_parts = []
+        if content_area:
+            tags_to_check = ['p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'li', 'td', 'th', 'pre']
+            for element in content_area.find_all(tags_to_check):
+                text = element.get_text(separator=' ', strip=True)
+                if text and len(text) > 10: text_parts.append(text)
+            if not text_parts or len(" ".join(text_parts)) < 200:
+                all_text_from_area = content_area.get_text(separator='\n', strip=True)
+                all_text_from_area = re.sub(r'\n\s*\n+', '\n\n', all_text_from_area)
+                if all_text_from_area and len(all_text_from_area) > len(" ".join(text_parts)):
+                    text_parts = [line for line in all_text_from_area.split('\n\n') if line.strip() and len(line.strip()) > 10]
+        
+        full_text = "\n\n".join(list(dict.fromkeys(text_parts)))
+        if not full_text and hasattr(soup, 'body') and soup.body:
+            full_text = soup.body.get_text(separator='\n', strip=True)
+        elif not full_text: full_text = "Content could not be extracted."
+        
+        full_text = re.sub(r'\s{3,}', ' ', full_text)
+        full_text = re.sub(r'(\n\s*){3,}', '\n\n', full_text)
+        title_tag = soup.find('title')
+        title = title_tag.string.strip() if title_tag and title_tag.string else url_to_scrape
+        if len(full_text) > MAX_CONTENT_LENGTH_PER_URL: full_text = full_text[:MAX_CONTENT_LENGTH_PER_URL] + "..."
+        
+        logger.debug(f"SCRAPER_MODULE: Scraped {len(full_text)} chars from {url_to_scrape}. Title: {title}")
+        return {"url": url_to_scrape, "title": title.strip(), "content": full_text.strip()}
+    except requests.exceptions.HTTPError as e: logger.error(f"SCRAPER_MODULE: HTTP error {url_to_scrape}: {e.response.status_code}"); return {"url": url_to_scrape, "title": url_to_scrape, "error": f"HTTP error: {e.response.status_code}"}
+    except requests.exceptions.Timeout: logger.error(f"SCRAPER_MODULE: Timeout {url_to_scrape}"); return {"url": url_to_scrape, "title": url_to_scrape, "error": "Timeout"}
+    except requests.exceptions.RequestException as e: logger.error(f"SCRAPER_MODULE: Request failed {url_to_scrape}: {e}"); return {"url": url_to_scrape, "title": url_to_scrape, "error": f"Request failed: {e}"}
+    except Exception as e: logger.error(f"SCRAPER_MODULE: Scraping error {url_to_scrape}: {e}", exc_info=True); return {"url": url_to_scrape, "title": url_to_scrape, "error": f"Internal scraping error: {e}"}
+
+# search_and_scrape_duckduckgo remains the same as the last version.
+def search_and_scrape_duckduckgo(search_query, num_results=3):
+    scraped_data_all_urls = []
+    total_content_collected_length = 0
+    try:
+        logger.info(f"SCRAPER_MODULE (DDG): Searching for: '{search_query}' (max {num_results} results)")
+        search_results_urls_info = []
+        with DDGS() as ddgs:
+            ddg_search_results = ddgs.text(search_query, max_results=num_results + 2, region='wt-wt', safesearch='moderate')
+            if ddg_search_results:
+                for result in ddg_search_results[:num_results]: 
+                    if result.get('href'):
+                        search_results_urls_info.append({
+                            "url": result['href'],
+                            "title": result.get('title', 'N/A'),
+                            "description": result.get('body', 'N/A')
+                        })
+        
+        logger.info(f"SCRAPER_MODULE (DDG): Found {len(search_results_urls_info)} URLs: {[r['url'] for r in search_results_urls_info]}")
+        if not search_results_urls_info:
+            return [{"query": search_query, "engine": "DuckDuckGo", "error": "No search results."}]
+
+        for res_info in search_results_urls_info:
+            url_to_scrape = res_info["url"]
+            if total_content_collected_length >= MAX_TOTAL_SCRAPED_CONTENT: break
+            scraped_info = scrape_url(url_to_scrape)
+            if scraped_info:
+                if not scraped_info.get("title") or scraped_info.get("title") == url_to_scrape:
+                    scraped_info["title"] = res_info.get("title", url_to_scrape)
+                current_content = scraped_info.get("content", "")
+                ddg_desc = res_info.get("description")
+                if ddg_desc and (not current_content or len(current_content) < 150 or scraped_info.get("error")):
+                    scraped_info["content"] = f"Search result snippet: {ddg_desc}\n\n(Content from page below {'or error encountered' if  scraped_info.get('error') else ''}):\n{current_content if current_content else 'No content extracted.'}"
+                scraped_data_all_urls.append(scraped_info)
+                if scraped_info.get("content") and not scraped_info.get("error"):
+                    total_content_collected_length += len(scraped_info["content"])
+            else: scraped_data_all_urls.append({"url": url_to_scrape, "title": res_info.get("title", url_to_scrape), "error": "Scraping function returned no data."})
+        return scraped_data_all_urls
+    except Exception as e:
+        logger.error(f"SCRAPER_MODULE (DDG): Error for '{search_query}': {e}", exc_info=True)
+        return [{"query": search_query, "engine": "DuckDuckGo", "error": f"DDG search/scrape failed: {str(e)}"}]
+
+
+def search_and_scrape_google(search_query, num_results=10):
+    """
+    Performs a Google search (via googlesearch-python library) and scrapes the top results.
+    Uses arguments as per the provided documentation snippet.
+    """
+    scraped_data_all_urls = []
+    total_content_collected_length = 0
+    try:
+        logger.info(f"SCRAPER_MODULE (Google): Searching for: '{search_query}' (num_results={num_results})")
+        search_results_info = [] # Will store SearchResult objects or dicts
+        
+        try:
+            # Using arguments based on the provided documentation:
+            # query, lang, num_results, advanced, sleep_interval
+            # The library handles User-Agent internally.
+            # For few results (like 3), sleep_interval for inter-page requests is not triggered,
+            # but the library has an internal default pause between individual fetches.
+            # We can also set a small sleep_interval here for overall politeness for any number of results.
+            # 'advanced=True' is key to get SearchResult objects with title, url, description.
+            
+            # The search function directly returns a list of SearchResult objects if advanced=True
+            results_iterable = google_search_lib(
+                query=search_query,
+                num_results=num_results,
+                lang='en',          # Example language
+                advanced=True,      # To get SearchResult objects
+                sleep_interval=1.0  # Politeness delay between requests the library makes internally
+                                    # This applies if library makes multiple HTTP requests for the results.
+                                    # For num_results=3, it likely makes one request.
+            )
+            
+            # Convert SearchResult objects to our desired dict format
+            for res_obj in results_iterable:
+                if hasattr(res_obj, 'url'): # Check if it's a SearchResult object
+                    search_results_info.append({
+                        "url": res_obj.url,
+                        "title": getattr(res_obj, 'title', 'N/A'),
+                        "description": getattr(res_obj, 'description', 'N/A')
+                    })
+                elif isinstance(res_obj, str): # Fallback if advanced=False or lib changes
+                    search_results_info.append({"url": res_obj, "title": "N/A", "description": "N/A"})
+
+            logger.info(f"SCRAPER_MODULE (Google): Found {len(search_results_info)} result objects: {[r['url'] for r in search_results_info]}")
+
+        except Exception as e_search:
+            logger.error(f"SCRAPER_MODULE (Google): Error during google_search_lib call for '{search_query}': {e_search}", exc_info=True)
+            if "HTTP Error 429" in str(e_search):
+                return [{"query": search_query, "engine": "Google", "error": f"Google search blocked (HTTP 429). Try again later or use DuckDuckGo."}]
+            return [{"query": search_query, "engine": "Google", "error": f"Google search library failed: {str(e_search)}"}]
+
+        if not search_results_info:
+            return [{"query": search_query, "engine": "Google", "error": "No search results retrieved."}]
+
+        for res_info_item in search_results_info:
+            url_to_scrape = res_info_item["url"]
+            if total_content_collected_length >= MAX_TOTAL_SCRAPED_CONTENT:
+                logger.info(f"SCRAPER_MODULE (Google): Reached max total content length. Stopping further scraping.")
+                break
+            
+            scraped_info = scrape_url(url_to_scrape)
+            if scraped_info:
+                # Use Google's title/description if scraping failed to get them or content is short
+                if not scraped_info.get("title") or scraped_info.get("title") == url_to_scrape:
+                    scraped_info["title"] = res_info_item.get("title", url_to_scrape)
+                
+                current_content = scraped_info.get("content", "")
+                google_desc = res_info_item.get("description")
+
+                if google_desc and (not current_content or len(current_content) < 150 or scraped_info.get("error")):
+                     scraped_info["content"] = f"Search result description: {google_desc}\n\n(Content from page below {'or error encountered' if  scraped_info.get('error') else ''}):\n{current_content if current_content else 'No content extracted.'}"
+                elif not current_content and google_desc: # If no content at all was scraped
+                     scraped_info["content"] = f"Could not scrape full content. Search result description: {google_desc}"
+
+                scraped_data_all_urls.append(scraped_info)
+                if scraped_info.get("content") and not scraped_info.get("error"):
+                    total_content_collected_length += len(scraped_info["content"])
+            else:
+                 scraped_data_all_urls.append({"url": url_to_scrape, "title": res_info_item.get("title", "N/A"), "error": "Scraping function (scrape_url) returned no data."})
+        
+        return scraped_data_all_urls
+
+    except Exception as e:
+        logger.error(f"SCRAPER_MODULE (Google): General error during search_and_scrape_google for '{search_query}': {e}", exc_info=True)
+        return [{"query": search_query, "engine": "Google", "error": f"Overall Google search or scraping process failed: {str(e)}"}]
+def search_and_scrape(search_query, num_results=10):
+    """
+    Performs a Google search (via googlesearch-python library) and scrapes the top results.
+    Uses arguments as per the provided documentation snippet.
+    """
+    scraped_data_all_urls = []
+    total_content_collected_length = 0
+    try:
+        logger.info(f"SCRAPER_MODULE (Google): Searching for: '{search_query}' (num_results={num_results})")
+        search_results_info = [] # Will store SearchResult objects or dicts
+        
+        try:
+            # Using arguments based on the provided documentation:
+            # query, lang, num_results, advanced, sleep_interval
+            # The library handles User-Agent internally.
+            # For few results (like 3), sleep_interval for inter-page requests is not triggered,
+            # but the library has an internal default pause between individual fetches.
+            # We can also set a small sleep_interval here for overall politeness for any number of results.
+            # 'advanced=True' is key to get SearchResult objects with title, url, description.
+            
+            # The search function directly returns a list of SearchResult objects if advanced=True
+            results_iterable = google_search_lib(
+                query=search_query,
+                num_results=num_results,
+                lang='en',          # Example language
+                advanced=True,      # To get SearchResult objects
+                sleep_interval=1.0  # Politeness delay between requests the library makes internally
+                                    # This applies if library makes multiple HTTP requests for the results.
+                                    # For num_results=3, it likely makes one request.
+            )
+            
+            # Convert SearchResult objects to our desired dict format
+            for res_obj in results_iterable:
+                if hasattr(res_obj, 'url'): # Check if it's a SearchResult object
+                    search_results_info.append({
+                        "url": res_obj.url,
+                        "title": getattr(res_obj, 'title', 'N/A'),
+                        "description": getattr(res_obj, 'description', 'N/A')
+                    })
+                elif isinstance(res_obj, str): # Fallback if advanced=False or lib changes
+                    search_results_info.append({"url": res_obj, "title": "N/A", "description": "N/A"})
+
+            logger.info(f"SCRAPER_MODULE (Google): Found {len(search_results_info)} result objects: {[r['url'] for r in search_results_info]}")
+
+        except Exception as e_search:
+            logger.error(f"SCRAPER_MODULE (Google): Error during google_search_lib call for '{search_query}': {e_search}", exc_info=True)
+            if "HTTP Error 429" in str(e_search):
+                return [{"query": search_query, "engine": "Google", "error": f"Google search blocked (HTTP 429). Try again later or use DuckDuckGo."}]
+            return [{"query": search_query, "engine": "Google", "error": f"Google search library failed: {str(e_search)}"}]
+
+        if not search_results_info:
+            return [{"query": search_query, "engine": "Google", "error": "No search results retrieved."}]
+
+        for res_info_item in search_results_info:
+            url_to_scrape = res_info_item["url"]
+            if total_content_collected_length >= MAX_TOTAL_SCRAPED_CONTENT:
+                logger.info(f"SCRAPER_MODULE (Google): Reached max total content length. Stopping further scraping.")
+                break
+            
+            scraped_info = scrape_url(url_to_scrape)
+            if scraped_info:
+                # Use Google's title/description if scraping failed to get them or content is short
+                if not scraped_info.get("title") or scraped_info.get("title") == url_to_scrape:
+                    scraped_info["title"] = res_info_item.get("title", url_to_scrape)
+                
+                current_content = scraped_info.get("content", "")
+                google_desc = res_info_item.get("description")
+
+                if google_desc and (not current_content or len(current_content) < 150 or scraped_info.get("error")):
+                     scraped_info["content"] = f"Search result description: {google_desc}\n\n(Content from page below {'or error encountered' if  scraped_info.get('error') else ''}):\n{current_content if current_content else 'No content extracted.'}"
+                elif not current_content and google_desc: # If no content at all was scraped
+                     scraped_info["content"] = f"Could not scrape full content. Search result description: {google_desc}"
+
+                scraped_data_all_urls.append(scraped_info)
+                if scraped_info.get("content") and not scraped_info.get("error"):
+                    total_content_collected_length += len(scraped_info["content"])
+            else:
+                 scraped_data_all_urls.append({"url": url_to_scrape, "title": res_info_item.get("title", "N/A"), "error": "Scraping function (scrape_url) returned no data."})
+        
+        return scraped_data_all_urls
+
+    except Exception as e:
+        logger.error(f"SCRAPER_MODULE (Google): General error during search_and_scrape_google for '{search_query}': {e}", exc_info=True)
+        return [{"query": search_query, "engine": "Google", "error": f"Overall Google search or scraping process failed: {str(e)}"}]