Russell Ballestrini
0589138396
Ported from Discord bot's async web fetcher improvements. Add comprehensive logging to understand crawl behavior: - Log crawl parameters at start (max_depth, URLs, keywords) - Debug log for crawl queue state during processing - Detailed link extraction stats with skip reasons: - Total links found - Links added to crawl queue - Links skipped by robots.txt - Links skipped (wrong domain) - Links skipped (max depth reached) Applied to both: - Fresh page fetching and link extraction - Cached page link extraction for depth traversal This diagnostic logging helps identify why crawlers find fewer pages than expected (e.g., robots.txt blocking, domain filtering, depth limits). No crawl logic changes - purely diagnostic visibility.
1092 lines
43 KiB
Python
1092 lines
43 KiB
Python
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,
|
|
)
|