File size: 3,742 Bytes
b311643 2ddc644 b311643 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | import re
import urllib.parse
import requests
from bs4 import BeautifulSoup
import html2text
import warnings
# Suppress duckduckgo_search rename warning
warnings.filterwarnings("ignore", message=".*duckduckgo_search.*")
from duckduckgo_search import DDGS
# Standard browser headers to avoid getting blocked by websites
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.5",
"Referer": "https://www.google.com/"
}
def clean_text(text: str) -> str:
"""Cleans excess whitespace and formats text nicely."""
# Replace multiple newlines/spaces with single ones
text = re.sub(r'\n+', '\n', text)
text = re.sub(r' +', ' ', text)
return text.strip()
def web_search(query: str, max_results: int = 3) -> list:
"""
Searches DuckDuckGo and returns a list of dictionaries with titles, hrefs, and body snippets.
Falls back gracefully if the search fails.
"""
try:
results = []
with DDGS() as ddgs:
for r in ddgs.text(query, max_results=max_results):
results.append({
"title": r.get("title", "No Title"),
"url": r.get("href", ""),
"snippet": r.get("body", "")
})
return results
except Exception as e:
print(f"Error during DuckDuckGo search: {e}")
return []
def scrape_url(url: str, max_chars: int = 4000) -> str:
"""
Fetches the web page content and converts it to clean markdown.
Truncates the output to fit context windows.
"""
if not url.startswith("http"):
return "Invalid URL format."
try:
response = requests.get(url, headers=HEADERS, timeout=8)
if response.status_code != 200:
return f"Failed to retrieve page. Status code: {response.status_code}"
# Detect and convert content
content_type = response.headers.get('Content-Type', '').lower()
if 'text/html' not in content_type:
return f"Scraping is limited to HTML content. Content-Type received: {content_type}"
# Initialize html2text converter
h = html2text.HTML2Text()
h.ignore_links = False
h.ignore_images = True
h.ignore_emphasis = False
h.body_width = 0 # Wrap lines at infinity
# Extract HTML
html = response.text
markdown_content = h.handle(html)
# Clean text
markdown_content = clean_text(markdown_content)
if len(markdown_content) > max_chars:
return markdown_content[:max_chars] + "\n\n... [Content Truncated due to size constraints] ..."
return markdown_content
except requests.exceptions.Timeout:
return "Scraping error: Connection timed out."
except Exception as e:
return f"Scraping error occurred: {str(e)}"
def format_search_results_for_prompt(query: str, search_results: list) -> str:
"""Formats search results and snippets into a structured text context block."""
if not search_results:
return "No search results returned for the query."
context = f"### WEB SEARCH RESULTS FOR: '{query}'\n"
context += "Below are relevant snippets retrieved from the web. Use these to formulate a factually correct answer:\n\n"
for idx, res in enumerate(search_results, 1):
context += f"Source [{idx}]: {res['title']}\n"
context += f"URL: {res['url']}\n"
context += f"Snippet: {res['snippet']}\n\n"
context += "---\n"
return context
|