import os import requests import random import time import re from urllib.parse import urlparse, urljoin from datetime import datetime, timezone from concurrent.futures import ThreadPoolExecutor, as_completed import urllib.robotparser import logging from sqlalchemy import ( create_engine, Column, Integer, String, Text, DateTime, UniqueConstraint, ) from sqlalchemy.orm import declarative_base, sessionmaker from openai import OpenAI from bs4 import BeautifulSoup from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity import numpy as np import json # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler(), logging.FileHandler("crawler.log")], ) logger = logging.getLogger(__name__) # Default Hermes endpoints DEFAULT_HERMES_ENDPOINTS = [ "https://hermes.ai.unturf.com/v1", # 80K tokens "https://hermes2.ai.unturf.com/v1", # 48K tokens ] # Max characters to send to extraction MAX_HTML_INPUT_CHARS = 50000 # Default crawl delay DEFAULT_CRAWL_DELAY = 2.0 # Token limits for Hermes HERMES_TOKEN_LIMIT = 80000 HERMES_INPUT_TOKENS = 72000 HERMES_COMPLETION_TOKENS = 8000 HERMES2_TOKEN_LIMIT = 48000 HERMES2_INPUT_TOKENS = 40000 HERMES2_COMPLETION_TOKENS = 8000 # Max words for chunking MAX_WORDS_LONG = 10000 MAX_WORDS_SHORT = 5000 # Context window limit (words, ~4 chars per token) CONTEXT_WINDOW_WORDS = 20000 # ~80K tokens # Default max results for search DEFAULT_MAX_RESULTS = 25 Base = declarative_base() class Article(Base): __tablename__ = "articles" id = Column(Integer, primary_key=True) url = Column(String, unique=True, nullable=False) title = Column(String, nullable=False) raw_html = Column(Text, nullable=False) extracted_content = Column(Text, nullable=False) summary = Column(Text, nullable=False) linked_domains = Column(Text, nullable=True) # Comma-separated list score = Column(Integer, nullable=True, default=0) # Quality/relevance score (0-100+) fetched_at = Column( DateTime(timezone=True), default=lambda: datetime.now(timezone.utc) ) __table_args__ = (UniqueConstraint("url", name="_url_uc"),) class BaseCrawler: def __init__( self, api_key, model, db_path="data/articles.db", hermes_endpoints=None, user_agent_append="", ): self.hermes_endpoints = hermes_endpoints or DEFAULT_HERMES_ENDPOINTS self.clients = [ OpenAI(base_url=ep, api_key=api_key) for ep in self.hermes_endpoints ] self.api_key = api_key self.model = model self.db_path = db_path default_requests_ua = f"python-requests/{requests.__version__}" self.user_agent = ( f"{default_requests_ua} unturf-deep-research {user_agent_append}".strip() ) self.session = requests.Session() self.session.headers.update({"User-Agent": self.user_agent}) os.makedirs(os.path.dirname(db_path), exist_ok=True) self.engine = create_engine( f"sqlite:///{self.db_path}", echo=False, future=True, connect_args={"check_same_thread": False}, ) Base.metadata.create_all(self.engine) self.SessionLocal = sessionmaker( bind=self.engine, autoflush=False, autocommit=False ) self.robot_parsers = {} self.domain_last_fetched = {} self.domain_crawl_delays = {} self.robots_txt_content = {} self.visited_urls = set() self.linked_domains = set() def _get_domain(self, url): parsed = urlparse(url) return parsed.netloc def _is_subdomain(self, domain, base_domain): if domain == base_domain: return True if domain.endswith("." + base_domain): return True return False def _score_page(self, page_data, query_keywords=None): """ Score a page based on content quality metrics and optional query relevance. Higher scores indicate more valuable content. Scoring factors: - Content length (more content = higher score) - Title quality (descriptive titles = higher score) - URL quality (cleaner URLs = higher score) - Content density (unique words = higher score) - Query relevance (if keywords provided, matching content scores higher) Args: page_data: Dict with 'text', 'title', 'url' keys query_keywords: Optional list of keywords from user's query for relevance scoring Returns: Float score (0-100+, can exceed 100 with relevance bonus) """ score = 0.0 # Factor 1: Content length (0-30 points) text_len = len(page_data.get('text', '')) score += min(30, text_len / 167) # 5000 chars = 30 points # Factor 2: Title quality (0-15 points) title = page_data.get('title', '') if title and title != "Untitled": title_len = len(title) if 10 <= title_len <= 100: score += 15 elif 5 <= title_len < 10 or 100 < title_len <= 150: score += 8 else: score += 3 # Factor 3: URL quality (0-15 points) url = page_data.get('url', '') if url: if '?' in url: score += 3 # Query strings = dynamic content elif '#' in url: score += 8 # Fragments slightly better else: score += 15 # Clean URLs best # Bonus for readable paths path_parts = url.split('/') if any(len(part) > 3 and part.replace('-', '').replace('_', '').isalnum() for part in path_parts): score += 3 # Factor 4: Content density (0-15 points) if text_len > 0: words = page_data.get('text', '').lower().split() unique_words = len(set(words)) if len(words) > 0: uniqueness_ratio = unique_words / len(words) score += uniqueness_ratio * 15 # Factor 5: Query relevance (0-40 points) if query_keywords and len(query_keywords) > 0: text_lower = page_data.get('text', '').lower() title_lower = title.lower() url_lower = url.lower() keyword_matches = 0 keyword_density = 0.0 words = text_lower.split() if text_lower else [] for keyword in query_keywords: keyword_lower = keyword.lower() text_count = text_lower.count(keyword_lower) title_count = title_lower.count(keyword_lower) url_count = url_lower.count(keyword_lower) if text_count > 0: keyword_matches += 1 keyword_density += text_count if title_count > 0: score += 5 * title_count if url_count > 0: score += 3 * url_count if len(query_keywords) > 0: match_ratio = keyword_matches / len(query_keywords) score += match_ratio * 20 if len(words) > 0 and keyword_density > 0: density_score = min(10, (keyword_density / len(words)) * 1000) score += density_score return score def _fetch_robots_txt(self, domain): if domain in self.robot_parsers: logger.info(f"Using cached robots.txt for {domain}") return self.robot_parsers[domain] logger.info(f"Fetching robots.txt for {domain}") robots_url = f"https://{domain}/robots.txt" parser = urllib.robotparser.RobotFileParser() parser.set_url(robots_url) try: resp = self.session.get(robots_url, timeout=5) resp.raise_for_status() self.robots_txt_content[domain] = resp.text parser.parse(resp.text.splitlines()) self.robot_parsers[domain] = parser delay = parser.crawl_delay(self.user_agent) self.domain_crawl_delays[domain] = ( delay if delay is not None else DEFAULT_CRAWL_DELAY ) except Exception as e: logger.error(f"Could not fetch robots.txt for {domain}: {e}") self.robot_parsers[domain] = None self.robots_txt_content[domain] = None self.domain_crawl_delays[domain] = DEFAULT_CRAWL_DELAY return self.robot_parsers[domain] def _can_fetch(self, url): domain = self._get_domain(url) parser = self._fetch_robots_txt(domain) if parser is None: return True can_fetch = parser.can_fetch(self.user_agent, url) if not can_fetch: logger.warning(f"Blocked by robots.txt: {url}") robots_content = self.robots_txt_content.get(domain, "") if robots_content: logger.info(f"Relevant robots.txt rules for {domain}:") current_user_agent = None relevant_rules = [] for line in robots_content.splitlines(): line = line.strip() if not line or line.startswith("#"): continue if line.lower().startswith("user-agent:"): current_user_agent = line[11:].strip() elif line.lower().startswith("disallow:") and current_user_agent: rule = line[9:].strip() if current_user_agent.lower() in (self.user_agent.lower(), "*"): parsed_url = urlparse(url) path = parsed_url.path if rule and ( path.startswith(rule) or (rule.endswith("*") and path.startswith(rule[:-1])) ): relevant_rules.append( f" User-agent: {current_user_agent}\n Disallow: {rule}" ) if relevant_rules: logger.info("\n".join(relevant_rules)) else: logger.info( "No specific Disallow rules found; may be blocked by a broad rule." ) else: logger.info("No robots.txt content available to display rules.") return can_fetch def _enforce_crawl_delay(self, domain): delay = self.domain_crawl_delays.get(domain, DEFAULT_CRAWL_DELAY) last_fetched = self.domain_last_fetched.get(domain, 0) elapsed = time.time() - last_fetched if elapsed < delay: sleep_time = delay - elapsed logger.info( f"Delaying crawl for {domain} by {sleep_time:.2f} seconds due to crawl delay" ) time.sleep(sleep_time) self.domain_last_fetched[domain] = time.time() def fetch_webpage(self, url): if not url.startswith(("http://", "https://")): logger.error(f"Invalid URL scheme: {url}") return None, [] try: if not self._can_fetch(url): return None, [] domain = self._get_domain(url) self._enforce_crawl_delay(domain) r = self.session.get(url, timeout=15) r.raise_for_status() html = r.text soup = BeautifulSoup(html, "html.parser") links = [] for a_tag in soup.find_all("a", href=True): href = a_tag["href"] if href.startswith(("javascript:", "#")): continue absolute_url = urljoin(url, href) parsed = urlparse(absolute_url) if parsed.scheme in ("http", "https"): links.append(absolute_url) return html, links except requests.exceptions.RequestException as e: logger.error(f"Error fetching {url}: {e}") return None, [] def _fanout_call(self, messages, max_tokens, prefer_hermes=False): def estimate_input_tokens(messages): total_chars = sum(len(m["content"]) for m in messages) return (total_chars // 5 + 1) // 3 + 1 def call_client(client, endpoint, messages, max_tokens): try: input_tokens = estimate_input_tokens(messages) total_tokens = input_tokens + max_tokens if ( endpoint == "https://hermes2.ai.unturf.com/v1" and total_tokens > HERMES2_TOKEN_LIMIT ): logger.info( f"Skipping {endpoint}: request exceeds {HERMES2_TOKEN_LIMIT} tokens ({total_tokens})" ) return None, endpoint if ( endpoint == "https://hermes.ai.unturf.com/v1" and total_tokens > HERMES_TOKEN_LIMIT ): logger.info( f"Skipping {endpoint}: request exceeds {HERMES_TOKEN_LIMIT} tokens ({total_tokens})" ) return None, endpoint response = client.chat.completions.create( model=self.model, messages=messages, temperature=0, max_tokens=max_tokens, ) return response.choices[0].message.content, endpoint except Exception as e: logger.error(f"Error calling endpoint {endpoint}: {e}") return None, endpoint with ThreadPoolExecutor(max_workers=len(self.hermes_endpoints)) as executor: futures = [ executor.submit(call_client, client, endpoint, messages, max_tokens) for client, endpoint in zip(self.clients, self.hermes_endpoints) ] for future in as_completed(futures): result, used_endpoint = future.result() if result is not None: logger.info(f"Success using {used_endpoint}") return result raise RuntimeError("All Hermes endpoints failed.") def extract_with_hermes(self, html): text = BeautifulSoup(html, "html.parser").get_text(separator="\n") if len(text) > MAX_HTML_INPUT_CHARS: text = text[:MAX_HTML_INPUT_CHARS] messages = [ { "role": "system", "content": ( "Extract the main article content, preserving all formatting, structure, and relevant details, " "including headers, paragraphs, lists, and key text. Exclude only ads, navigation, and unrelated boilerplate. " "Maximize content retention to capture comprehensive information, ensuring no relevant text is omitted." ), }, {"role": "user", "content": text}, ] try: return self._fanout_call(messages, max_tokens=HERMES_COMPLETION_TOKENS) except RuntimeError: shortened = text[: MAX_HTML_INPUT_CHARS // 2] messages[1]["content"] = shortened return self._fanout_call(messages, max_tokens=HERMES_COMPLETION_TOKENS // 2) def summarize_with_hermes(self, content): messages = [ { "role": "system", "content": ( "Summarize the following article concisely as bullet points, keeping all factual details and structure. Avoid hallucination." ), }, {"role": "user", "content": content}, ] return self._fanout_call(messages, max_tokens=HERMES_COMPLETION_TOKENS) def _split_into_chunks(self, text, content_length): max_words = MAX_WORDS_LONG if content_length > 5000 else MAX_WORDS_SHORT logger.info(f"Using max_words={max_words} for content length={content_length}") words = text.split() chunks = [] current_chunk = [] current_word_count = 0 for word in words: current_chunk.append(word) current_word_count += 1 if current_word_count >= max_words: chunks.append(" ".join(current_chunk)) current_chunk = [] current_word_count = 0 if current_chunk: chunks.append(" ".join(current_chunk)) return chunks if len(chunks) > 1 else [text] def cache_article( self, url, title, html, extracted, summary, linked_domains, force_crawl, score=0 ): with self.SessionLocal() as db: existing = db.query(Article).filter_by(url=url).first() linked_domains_str = ( ",".join(sorted(linked_domains)) if linked_domains else "" ) if existing and force_crawl: existing.title = title existing.raw_html = html existing.extracted_content = extracted existing.summary = summary existing.linked_domains = linked_domains_str existing.score = int(score) existing.fetched_at = datetime.now(timezone.utc) try: db.commit() logger.info( f"Updated cached article: {url} with score={score:.1f}, linked domains: {linked_domains_str}" ) except Exception as e: db.rollback() logger.error(f"Error updating article {url}: {e}") raise elif not existing: art = Article( url=url, title=title, raw_html=html, extracted_content=extracted, summary=summary, linked_domains=linked_domains_str, score=int(score), fetched_at=datetime.now(timezone.utc), ) db.add(art) try: db.commit() logger.info( f"Cached new article: {url} with score={score:.1f}, linked domains: {linked_domains_str}" ) except Exception as e: db.rollback() logger.error(f"Error caching article {url}: {e}") raise else: logger.info(f"Skipping cache update for {url} (force_crawl=False)") def process_url( self, title_url, depth, max_depth, base_domain, trust_subdomains, trust_linked_domains, force_crawl, query_keywords=None, ): title, url = title_url if url in self.visited_urls: logger.info(f"Skipping already visited: {url}") return [] self.visited_urls.add(url) with self.SessionLocal() as db: existing = db.query(Article).filter_by(url=url).first() # If already cached and not forcing re-crawl if existing and not force_crawl: logger.info(f"Already cached: {url}") # If we're not at max depth, still need to extract links for depth traversal if depth < max_depth: logger.info(f"Extracting links from cached page {url} for depth {depth+1} traversal") html, links = self.fetch_webpage(url) if not html: logger.warning(f"Failed to re-fetch {url} for link extraction") return [] # Extract links and return them for further crawling new_urls = [] skipped_robots = 0 skipped_domain = 0 skipped_depth = 0 for link in links: parsed = urlparse(link) if parsed.scheme not in ("http", "https"): continue link_domain = parsed.netloc if self._can_fetch(link): if link_domain: self.linked_domains.add(link_domain) should_crawl = False is_base_or_subdomain = ( self._is_subdomain(link_domain, base_domain) if trust_subdomains else link_domain == base_domain ) if is_base_or_subdomain and depth < max_depth: should_crawl = True elif ( trust_linked_domains and not is_base_or_subdomain and depth == 0 ): should_crawl = True if should_crawl: new_urls.append((f"Linked from {url}", link)) else: if not is_base_or_subdomain: skipped_domain += 1 else: skipped_depth += 1 else: skipped_robots += 1 logger.info(f"Link extraction (cached page): found {len(links)} total, added {len(new_urls)} to crawl queue, skipped {skipped_robots} (robots.txt), skipped {skipped_domain} (wrong domain), skipped {skipped_depth} (max depth reached)") return new_urls else: # At max depth, no need to extract links return [] # Need to fetch and cache this page logger.info(f"Fetching (depth {depth}): {url}") html, links = self.fetch_webpage(url) if not html: return [] soup = BeautifulSoup(html, "html.parser") page_title = soup.title.string.strip() if soup.title else title extracted = self.extract_with_hermes(html) # Calculate quality/relevance score for this page page_data = { 'url': url, 'title': page_title, 'text': extracted } score = self._score_page(page_data, query_keywords=query_keywords) logger.info(f"Scored page {url}: {score:.1f} points") page_linked_domains = set() new_urls = [] skipped_robots = 0 skipped_domain = 0 skipped_depth = 0 for link in links: parsed = urlparse(link) if parsed.scheme not in ("http", "https"): continue link_domain = parsed.netloc if self._can_fetch(link): if link_domain: page_linked_domains.add(link_domain) self.linked_domains.add(link_domain) should_crawl = False is_base_or_subdomain = ( self._is_subdomain(link_domain, base_domain) if trust_subdomains else link_domain == base_domain ) if is_base_or_subdomain and depth < max_depth: should_crawl = True elif ( trust_linked_domains and not is_base_or_subdomain and depth == 0 ): should_crawl = True if should_crawl: new_urls.append((f"Linked from {url}", link)) else: if not is_base_or_subdomain: skipped_domain += 1 else: skipped_depth += 1 else: skipped_robots += 1 logger.info(f"Link extraction: found {len(links)} total, added {len(new_urls)} to crawl queue, skipped {skipped_robots} (robots.txt), skipped {skipped_domain} (wrong domain), skipped {skipped_depth} (max depth reached)") self.cache_article( url, page_title, html, extracted, extracted, page_linked_domains, force_crawl, score=score, ) logger.info( f"Found linked domains (robots.txt compliant): {page_linked_domains}" ) return new_urls def crawl_recursive( self, start_urls, max_depth=2, base_domain=None, trust_subdomains=True, trust_linked_domains=False, force_crawl=False, query_keywords=None, ): to_crawl = [(title_url, 0) for title_url in start_urls] all_urls = [] logger.info(f"Starting crawl with max_depth={max_depth}, initial URLs={len(start_urls)}, query_keywords={query_keywords}") while to_crawl: logger.debug(f"Crawl queue: {len(to_crawl)} URLs remaining, {len(all_urls)} URLs processed so far") current_title_url, depth = to_crawl.pop(0) new_urls = self.process_url( current_title_url, depth, max_depth, base_domain, trust_subdomains, trust_linked_domains, force_crawl, query_keywords=query_keywords, ) all_urls.extend(new_urls) for new_url in new_urls: to_crawl.append((new_url, depth + 1)) return all_urls def _extract_keywords_with_hermes(self, query): messages = [ { "role": "system", "content": ( "Analyze the query and extract up to 5 relevant keywords or phrases " "capturing main topics or entities. Focus on nouns, proper nouns, and key concepts. " "Avoid generic terms like 'what', 'is'. Return as a comma-separated string." ), }, {"role": "user", "content": query}, ] try: result = self._fanout_call(messages, max_tokens=100) keywords = [k.strip() for k in result.split(",") if k.strip()] if len(keywords) > 5: keywords = keywords[:5] logger.info(f"Extracted keywords for query '{query}': {keywords}") return keywords except RuntimeError: logger.warning("Failed to extract keywords, falling back to query split") return query.lower().split() def _search_sqlite( self, query, keywords, max_results=DEFAULT_MAX_RESULTS, allowed_domains=None ): with self.SessionLocal() as db: articles = db.query(Article).all() if not articles: logger.warning("No articles found in SQLite cache.") return [] filtered_articles = [] for article in articles: article_domain = self._get_domain(article.url) article_linked_domains = ( set(article.linked_domains.split(",")) if article.linked_domains else set() ) if allowed_domains: if article_domain in allowed_domains or any( ld in allowed_domains for ld in article_linked_domains ): filtered_articles.append(article) else: filtered_articles.append(article) if not filtered_articles: logger.warning( "No articles match the allowed domains or linked domains." ) return [] core_keywords = keywords + query.lower().split() relevant_articles = [] for article in filtered_articles: content_lower = article.extracted_content.lower() keyword_score = sum( 1 for kw in core_keywords if kw.lower() in content_lower ) article_linked_domains = ( set(article.linked_domains.split(",")) if article.linked_domains else set() ) linked_domain_score = sum( 1 for kw in core_keywords if any(kw.lower() in ld.lower() for ld in article_linked_domains) ) total_score = keyword_score + linked_domain_score if total_score > 0: relevant_articles.append((article, total_score)) relevant_articles.sort(key=lambda x: x[1], reverse=True) top_articles = relevant_articles[:max_results] if top_articles: documents = [article.extracted_content for article, _ in top_articles] vectorizer = TfidfVectorizer(stop_words="english") try: tfidf_matrix = vectorizer.fit_transform(documents + [query]) similarities = cosine_similarity( tfidf_matrix[-1], tfidf_matrix[:-1] )[0] scored_articles = [ (article, score, total_score) for (article, total_score), score in zip( top_articles, similarities ) ] scored_articles.sort( key=lambda x: 0.5 * x[1] + 0.5 * (x[2] / max(1, max(s[2] for s in scored_articles))), reverse=True, ) return [ { "url": article.url, "title": article.title, "content": article.extracted_content, "linked_domains": article.linked_domains, } for article, _, _ in scored_articles ] except ValueError as e: logger.warning( f"TF-IDF failed: {e}, falling back to keyword ranking" ) return [ { "url": article.url, "title": article.title, "content": article.extracted_content, "linked_domains": article.linked_domains, } for article, _ in top_articles ] return [] def aggregate_and_answer( self, query, query_keywords=None, allowed_domains=None, max_results=DEFAULT_MAX_RESULTS, ): cache_results = self._search_sqlite( query, query_keywords, max_results=max_results, allowed_domains=allowed_domains, ) logger.info(f"Retrieved {len(cache_results)} articles from SQLite cache") def estimate_tokens(text): return (len(text) // 5 + 1) // 3 + 1 combined_content = [] total_words = 0 total_tokens = 0 prompt_template = f"Based on these article excerpts:\n{{}}\n\nProvide a comprehensive answer to: {query}" template_tokens = estimate_tokens(prompt_template.format("")) for result in cache_results: content = result["content"] linked_domains = result.get("linked_domains", "") word_count = len(content.split()) content_tokens = estimate_tokens(content) endpoint_limit = HERMES2_INPUT_TOKENS if ( total_words + word_count <= CONTEXT_WINDOW_WORDS and total_tokens + content_tokens + template_tokens <= endpoint_limit ): excerpt = f"From {result['url']} (Title: {result['title']}):\n{content}" if linked_domains: excerpt += f"\nLinked domains: {linked_domains}" combined_content.append(excerpt) total_words += word_count total_tokens += content_tokens logger.info( f"Included cached article {result['url']} (words: {word_count}, tokens: {content_tokens}, linked_domains: {linked_domains})" ) else: if total_tokens + template_tokens < endpoint_limit: remaining_tokens = endpoint_limit - total_tokens - template_tokens remaining_words = min(word_count, remaining_tokens * 3 // 5) trimmed_content = " ".join(content.split()[:remaining_words]) trimmed_tokens = estimate_tokens(trimmed_content) if trimmed_tokens <= remaining_tokens: excerpt = f"From {result['url']} (Title: {result['title']}):\n{trimmed_content}" if linked_domains: excerpt += f"\nLinked domains: {linked_domains}" combined_content.append(excerpt) total_words += remaining_words total_tokens += trimmed_tokens logger.info( f"Included trimmed cached article {result['url']} (words: {remaining_words}, tokens: {trimmed_tokens}, linked_domains: {linked_domains})" ) else: chunks = self._split_into_chunks(content, len(content)) for chunk in chunks: chunk_words = len(chunk.split()) chunk_tokens = estimate_tokens(chunk) if ( total_words + chunk_words <= CONTEXT_WINDOW_WORDS and total_tokens + chunk_tokens + template_tokens <= endpoint_limit ): excerpt = f"From {result['url']} (Title: {result['title']}):\n{chunk}" if linked_domains: excerpt += f"\nLinked domains: {linked_domains}" combined_content.append(excerpt) total_words += chunk_words total_tokens += chunk_tokens logger.info( f"Included chunk from cached article {result['url']} (words: {chunk_words}, tokens: {chunk_tokens}, linked_domains: {linked_domains})" ) else: logger.info( f"Skipping chunk from cached article {result['url']} (exceeds limit: {total_tokens + chunk_tokens + template_tokens})" ) break else: logger.info( f"Skipping cached article {result['url']} (exceeds limit: {total_tokens + content_tokens + template_tokens})" ) if not combined_content: return "No relevant information found to answer the query." combined = "\n\n".join(combined_content) total_tokens = estimate_tokens(combined) + template_tokens logger.info( f"Using {total_tokens} tokens from {len(combined_content)} articles/chunks" ) messages = [{"role": "user", "content": prompt_template.format(combined)}] return self._fanout_call( messages, max_tokens=HERMES_COMPLETION_TOKENS, prefer_hermes=True ) class DuckDuckGoCrawler(BaseCrawler): def fetch_duckduckgo_results(self, query, max_results=10): try: url = f"https://api.duckduckgo.com/?q={query}&format=json" headers = {"User-Agent": self.user_agent} response = self.session.get(url, headers=headers, timeout=10) response.raise_for_status() data = response.json() logger.info(f"DuckDuckGo API status: {response.status_code}, response: {response.text[:1000]}") results = [] for item in data.get("RelatedTopics", [])[:max_results]: if "FirstURL" in item: results.append( (item.get("Text", "Search Result"), item["FirstURL"]) ) logger.info( f"Fetched {len(results)} URLs from DuckDuckGo for query: {query}" ) return results except Exception as e: logger.error(f"Error fetching DuckDuckGo results: {e}") return [] def run( self, query, max_depth=2, trust_subdomains=True, trust_linked_domains=False, force_crawl=False, max_results=DEFAULT_MAX_RESULTS, ): self.visited_urls = set() self.linked_domains = set() logger.info( f"Starting DuckDuckGo crawl for query '{query}' with depth {max_depth}, " f"trust_subdomains={trust_subdomains}, trust_linked_domains={trust_linked_domains}, " f"force_crawl={force_crawl}, max_results={max_results}" ) # Extract query keywords EARLY for scoring during crawl logger.info("Extracting query keywords for relevance scoring...") query_keywords = self._extract_keywords_with_hermes(query) start_urls = self.fetch_duckduckgo_results(query) if not start_urls: logger.warning( "No DuckDuckGo results found. Falling back to cached articles." ) else: # Use the first result's domain as base_domain, or a dummy if none base_domain = ( urlparse(start_urls[0][1]).netloc if start_urls else "example.com" ) self.linked_domains.add(base_domain) self.crawl_recursive( start_urls, max_depth=max_depth, base_domain=base_domain, trust_subdomains=trust_subdomains, trust_linked_domains=trust_linked_domains, force_crawl=force_crawl, query_keywords=query_keywords, ) logger.info(f"Collected linked domains: {self.linked_domains}") logger.info("Generating comprehensive answer...") result = self.aggregate_and_answer( query, query_keywords=query_keywords, allowed_domains=self.linked_domains, max_results=max_results, ) print(result) class DirectTargetCrawler(BaseCrawler): def run( self, query, targets=None, max_depth=2, trust_subdomains=True, trust_linked_domains=False, force_crawl=False, max_results=DEFAULT_MAX_RESULTS, ): self.visited_urls = set() self.linked_domains = set() if not targets: raise ValueError( "No target provided. Please specify a target domain or URLs using the --target flag." ) target_list = [t.strip() for t in targets.split(",") if t.strip()] if not target_list: raise ValueError("No valid target URLs provided") base_domain = urlparse(target_list[0]).netloc self.linked_domains.add(base_domain) start_urls = [(f"Target Page {i+1}", url) for i, url in enumerate(target_list)] logger.info( f"Starting direct crawl from {targets} with depth {max_depth}, " f"trust_subdomains={trust_subdomains}, trust_linked_domains={trust_linked_domains}, " f"force_crawl={force_crawl}, max_results={max_results}" ) # Extract query keywords EARLY for scoring during crawl logger.info("Extracting query keywords for relevance scoring...") query_keywords = self._extract_keywords_with_hermes(query) self.crawl_recursive( start_urls, max_depth=max_depth, base_domain=base_domain, trust_subdomains=trust_subdomains, trust_linked_domains=trust_linked_domains, force_crawl=force_crawl, query_keywords=query_keywords, ) logger.info(f"Collected linked domains: {self.linked_domains}") logger.info("Generating comprehensive answer...") result = self.aggregate_and_answer( query, query_keywords=query_keywords, allowed_domains=self.linked_domains, max_results=max_results, ) print(result) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser( description="Web crawler for answering queries using DuckDuckGo or direct targets." ) parser.add_argument("query", help="Query to answer based on crawled content") parser.add_argument("--api-key", default="dummy-api-key", help="OpenAI API key") parser.add_argument( "--model", default="adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic", help="Model ID", ) parser.add_argument( "--target", help="Starting domain or comma-separated URLs to crawl (e.g., https://example.com or https://example.com,https://blog.example.com/post1)", ) parser.add_argument( "--duckduckgo", action="store_true", help="Use DuckDuckGo search results as starting URLs instead of direct targets", ) parser.add_argument( "--depth", type=int, default=2, help="Maximum crawl depth (default: 2)" ) parser.add_argument( "--no-trust-subdomains", action="store_true", help="Do not crawl subdomains of the base domain (default: crawl subdomains)", ) parser.add_argument( "--trust-linked-domains", action="store_true", help="Crawl linked domains non-recursively", ) parser.add_argument( "--force-crawl", action="store_true", help="Force re-crawling of cached URLs, updating existing articles", ) parser.add_argument( "--max-results", type=int, default=DEFAULT_MAX_RESULTS, help=f"Maximum number of pages to use for answering (default: {DEFAULT_MAX_RESULTS})", ) parser.add_argument( "--user-agent-append", default="", help="String to append to default user agent" ) args = parser.parse_args() if args.duckduckgo and args.target: raise ValueError( "Cannot use both --duckduckgo and --target flags simultaneously." ) crawler = ( DuckDuckGoCrawler( api_key=args.api_key, model=args.model, user_agent_append=args.user_agent_append, ) if args.duckduckgo else DirectTargetCrawler( api_key=args.api_key, model=args.model, user_agent_append=args.user_agent_append, ) ) if args.duckduckgo: crawler.run( args.query, max_depth=args.depth, trust_subdomains=not args.no_trust_subdomains, trust_linked_domains=args.trust_linked_domains, force_crawl=args.force_crawl, max_results=args.max_results, ) else: crawler.run( args.query, args.target, args.depth, trust_subdomains=not args.no_trust_subdomains, trust_linked_domains=args.trust_linked_domains, force_crawl=args.force_crawl, max_results=args.max_results, )