""" Web scraping module for topic research. Uses DuckDuckGo (no API key needed) + trafilatura for content extraction. """ import requests from bs4 import BeautifulSoup import trafilatura import re import time import urllib.parse HEADERS = { "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.9", } TRUSTED_DOMAINS = [ "wikipedia.org", "bbc.com", "reuters.com", "cnn.com", "theguardian.com", "nytimes.com", "sciencedaily.com", "nationalgeographic.com", "britannica.com", "nature.com", "smithsonianmag.com", "history.com", "nasa.gov", "scientificamerican.com", "wired.com", "techcrunch.com", ] def search_duckduckgo(query: str, max_results: int = 8) -> list[str]: """Search DuckDuckGo HTML and extract result URLs.""" encoded = urllib.parse.quote_plus(query) url = f"https://html.duckduckgo.com/html/?q={encoded}" try: resp = requests.get(url, headers=HEADERS, timeout=15) resp.raise_for_status() soup = BeautifulSoup(resp.text, "lxml") urls = [] for link in soup.select("a.result__a"): href = link.get("href", "") if "uddg=" in href: actual = urllib.parse.parse_qs(urllib.parse.urlparse(href).query).get("uddg", [None])[0] if actual: href = actual if href and href.startswith("http"): urls.append(href) if len(urls) >= max_results: break return urls except Exception as e: print(f"DuckDuckGo search error: {e}") return [] def scrape_url(url: str) -> str: """Extract clean text from a URL using trafilatura with BS4 fallback.""" try: downloaded = trafilatura.fetch_url(url) if downloaded: text = trafilatura.extract( downloaded, include_comments=False, include_tables=False, no_fallback=False, favor_recall=True, ) if text and len(text) > 100: return text[:3000] resp = requests.get(url, headers=HEADERS, timeout=10) resp.raise_for_status() soup = BeautifulSoup(resp.text, "lxml") for tag in soup(["script", "style", "nav", "footer", "header", "aside", "iframe"]): tag.decompose() text = soup.get_text(separator=" ", strip=True) text = re.sub(r'\s+', ' ', text) return text[:3000] except Exception as e: print(f"Scrape error for {url}: {e}") return "" def research_topic(topic: str, additional_notes: str = "", progress_callback=None) -> dict: """Thoroughly research a topic by scraping multiple sources.""" results = { "topic": topic, "sources": [], "key_facts": [], "full_text": "", } queries = [ f"{topic}", f"{topic} facts information explained", f"{topic} latest news developments", ] if additional_notes: queries.append(f"{topic} {additional_notes}") all_urls = [] for i, query in enumerate(queries): if progress_callback: progress_callback(f"🔍 Searching: {query[:50]}...") urls = search_duckduckgo(query, max_results=5) for url in urls: if url not in all_urls: all_urls.append(url) time.sleep(0.5) scraped_texts = [] for i, url in enumerate(all_urls[:10]): if progress_callback: progress_callback(f"📄 Reading source {i+1}/{min(len(all_urls), 10)}...") text = scrape_url(url) if text and len(text) > 200: scraped_texts.append(text) results["sources"].append({ "url": url, "text_length": len(text), "preview": text[:200], }) time.sleep(0.3) results["full_text"] = "\n\n---\n\n".join(scraped_texts) for text in scraped_texts[:5]: sentences = re.split(r'[.!?]\s+', text) facts = [s.strip() for s in sentences if len(s.strip()) > 30 and len(s.strip()) < 300] results["key_facts"].extend(facts[:5]) seen = set() unique_facts = [] for fact in results["key_facts"]: fact_lower = fact.lower().strip() if fact_lower not in seen: seen.add(fact_lower) unique_facts.append(fact) results["key_facts"] = unique_facts[:20] return results def get_search_keywords(topic: str, scenes_data: list) -> list[str]: """Extract search keywords from topic and scenes for image searching.""" keywords = [topic] for scene in scenes_data: visual = scene.get("visual", "") if visual: keywords.append(visual) return keywords