improve scraper

scraper.py

@@ -0,0 +1,107 @@
+import asyncio
+import aiohttp
+import random
+from bs4 import BeautifulSoup
+from googlesearch import search
+from urllib.parse import urlparse
+import os
+
+
+async def fetch(url, session, retries=3, delay=1):
+    if retries == 0:
+        print("Maximum retries exceeded. Failed to fetch URL.")
+        return
+    try:
+        async with session.get(url) as response:
+            return await response.text(encoding='latin')
+    except aiohttp.ClientResponseError as e:
+        if e.status == 429:  # HTTP 429: Too Many Requests
+            print(f"Too many requests. Retrying in {delay} seconds...")
+            await asyncio.sleep(delay)
+            return fetch(url, session, retries=retries - 1, delay=delay * 2)
+        print(f"Error fetching URL: {e}")
+
+
+async def scrape_websites(topic, num_results_per_link=10):
+    outputs = await scrape_google(topic, num_results_per_link)
+    # Select random links based on the user's input
+    selected_links = random.sample(outputs, min(num_results_per_link, len(outputs)))
+    # return the list of strings as a single string
+    # return "\n".join(selected_links)
+    return list(selected_links)
+
+
+async def scrape_google(topic, num_results=15) -> list[str]:
+    # Limit search results to 15 if num_results exceeds 15
+    num_results = min(num_results, 15)
+    # Asynchronous HTTP session
+    async with aiohttp.ClientSession() as session:
+        # Perform Google search
+        search_results = search(topic, num=3 * num_results)
+        search_results = remove_duplicate_results(search_results)
+        # Shuffle search results order
+        random.shuffle(search_results)
+        outlines = []
+        i = 0
+        # Keep scraping url's until you collect wanted num_results, or till you have tried all search results
+        while len(outlines) < num_results and i < len(search_results):
+            result = await scrape_url(search_results[i], session)
+            if result:
+                outlines.append(result)
+            i += 1
+    return outlines
+
+
+async def scrape_url(url, session) -> str:
+    try:
+        # Fetch HTML content asynchronously
+        html_content = await fetch(url, session)
+        # Parse HTML content with BeautifulSoup
+        soup = BeautifulSoup(html_content, 'html.parser')
+        # Extract outlines from parsed HTML
+        outlines = extract_outlines(soup)
+
+        # If outlines exist, accumulate them with website info
+        if outlines:
+            return get_website_info_str(url, outlines)
+        return ''
+
+    except Exception as e:
+        print(f"Error '{e}' while processing URL:{url}")
+        return ''
+
+
+# Minimum length threshold for relevant outlines
+MIN_LENGTH_THRESHOLD = 20
+is_irrelevant_outline = lambda outline: len(outline) < MIN_LENGTH_THRESHOLD
+extract_main_domain_from_url = lambda url: urlparse(url).netloc.split('www.')[-1].split('/')[0]
+extract_website_name_from_url = lambda url: urlparse(url).netloc
+extract_title_from_url = lambda url: os.path.basename(urlparse(url).path)
+
+
+def remove_duplicate_results(search_result) -> list:
+    unique_domains = set()
+    unique_websites = []
+    for url in search_result:
+        domain = extract_main_domain_from_url(url)
+        if domain not in unique_domains:
+            unique_domains.add(domain)
+            unique_websites.append(url)
+    return unique_websites
+
+
+def get_website_info_str(url, outlines) -> str:
+    website_name = extract_website_name_from_url(url)
+    info = f"Website: {website_name}\nURL: {url}\nOutlines:\n." + '\n.'.join(outlines)
+    return info
+
+
+def filter_outlines(outlines) -> list[str]:
+    return [outline for outline in set(outlines) if not is_irrelevant_outline(outline)]
+
+
+def extract_outlines(soup) -> list[str]:
+    headings = soup.find_all(['h1', 'h2', 'h3', 'h4'])
+    outlines = [heading.text.strip() for heading in headings]
+    if len(outlines) >= 3:
+        return filter_outlines(outlines)