470 lines
19 KiB
Python
470 lines
19 KiB
Python
import os
|
|
import requests
|
|
import random
|
|
import time
|
|
import re
|
|
from urllib.parse import quote_plus, urlparse, parse_qs, unquote
|
|
from datetime import datetime, timezone
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
import urllib.robotparser
|
|
import threading
|
|
|
|
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
|
|
|
|
# Default Hermes endpoints
|
|
DEFAULT_HERMES_ENDPOINTS = [
|
|
"https://hermes.ai.unturf.com/v1", # 80k context window
|
|
"https://hermes2.ai.unturf.com/v1", # Smaller context window
|
|
]
|
|
# Max characters to send to extraction to avoid context overflow
|
|
MAX_HTML_INPUT_CHARS = 50000
|
|
# Default crawl delay if robots.txt is missing or doesn't specify
|
|
DEFAULT_CRAWL_DELAY = 2.0 # seconds
|
|
# Token limits for Hermes endpoints
|
|
HERMES_TOKEN_LIMIT = 80000 # hermes.ai.unturf.com
|
|
HERMES2_TOKEN_LIMIT = 48000 # hermes2.ai.unturf.com
|
|
HERMES_INPUT_TOKENS = 72000
|
|
HERMES_COMPLETION_TOKENS = 8000
|
|
HERMES2_INPUT_TOKENS = 40000
|
|
HERMES2_COMPLETION_TOKENS = 8000
|
|
|
|
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) # Stores extracted_content unless summarized
|
|
fetched_at = Column(
|
|
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
|
|
)
|
|
__table_args__ = (UniqueConstraint("url", name="_url_uc"),)
|
|
|
|
|
|
class SQLAlchemyDuckDuckGoCrawler:
|
|
def __init__(
|
|
self,
|
|
api_key,
|
|
model,
|
|
db_path="data/articles.db",
|
|
hermes_endpoints=None,
|
|
user_agent_append="",
|
|
):
|
|
# Setup Hermes clients
|
|
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
|
|
|
|
# Configure user agent
|
|
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})
|
|
|
|
# Initialize DB
|
|
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
|
|
)
|
|
|
|
# Robots.txt parser and domain tracking
|
|
self.robot_parsers = {}
|
|
self.domain_last_fetched = {}
|
|
self.domain_crawl_delays = {}
|
|
self.robots_txt_content = {}
|
|
|
|
def _get_domain(self, url):
|
|
"""Extract the domain from a URL."""
|
|
parsed = urlparse(url)
|
|
return parsed.netloc
|
|
|
|
def _fetch_robots_txt(self, domain):
|
|
"""Fetch and parse robots.txt for a domain."""
|
|
if domain in self.robot_parsers:
|
|
return self.robot_parsers[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:
|
|
print(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):
|
|
"""Check if crawling the URL is allowed per robots.txt."""
|
|
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:
|
|
print(f"Blocked by robots.txt: {url}")
|
|
robots_content = self.robots_txt_content.get(domain, "")
|
|
if robots_content:
|
|
print(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:
|
|
print("\n".join(relevant_rules))
|
|
else:
|
|
print(
|
|
" No specific Disallow rules found; may be blocked by a broad rule."
|
|
)
|
|
else:
|
|
print(" No robots.txt content available to display rules.")
|
|
return can_fetch
|
|
|
|
def _enforce_crawl_delay(self, domain):
|
|
"""Enforce crawl delay for the 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
|
|
print(
|
|
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 search_duckduckgo(self, query, max_results=10):
|
|
"""Search DuckDuckGo and extract target URLs from redirect links."""
|
|
encoded = quote_plus(query)
|
|
url = f"https://html.duckduckgo.com/html/?q={encoded}"
|
|
resp = self.session.get(url, timeout=10)
|
|
resp.raise_for_status()
|
|
soup = BeautifulSoup(resp.text, "html.parser")
|
|
results = []
|
|
for link in soup.select(".result__title a"):
|
|
if len(results) >= max_results:
|
|
break
|
|
href = link.get("href")
|
|
title = link.get_text(strip=True)
|
|
if href and href.startswith("//duckduckgo.com/l/?uddg="):
|
|
parsed = urlparse(href)
|
|
query_params = parse_qs(parsed.query)
|
|
target_url = query_params.get("uddg", [None])[0]
|
|
if target_url:
|
|
target_url = unquote(target_url)
|
|
if target_url.startswith("//"):
|
|
target_url = "https:" + target_url
|
|
elif not target_url.startswith(("http://", "https://")):
|
|
target_url = "https://" + target_url
|
|
results.append((title, target_url))
|
|
else:
|
|
print(f"Skipping invalid redirect URL: {href}")
|
|
else:
|
|
if href and not href.startswith(("javascript:", "#")):
|
|
if href.startswith("//"):
|
|
href = "https:" + href
|
|
elif not href.startswith(("http://", "https://")):
|
|
href = "https://" + href
|
|
results.append((title, href))
|
|
else:
|
|
print(f"Skipping invalid URL: {href}")
|
|
return results
|
|
|
|
def fetch_webpage(self, url):
|
|
"""Fetch a webpage, validating the URL first."""
|
|
if not url.startswith(("http://", "https://")):
|
|
print(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()
|
|
return r.text
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"Error fetching {url}: {e}")
|
|
return None
|
|
|
|
def _fanout_call(self, messages, max_tokens, prefer_hermes=False):
|
|
"""Call Hermes endpoints, preferring hermes.ai.unturf.com for aggregate_and_answer."""
|
|
|
|
def call_client(client, endpoint):
|
|
try:
|
|
adjusted_max_tokens = max_tokens
|
|
if endpoint == "https://hermes2.ai.unturf.com/v1":
|
|
input_tokens = sum(len(m["content"]) // 4 + 1 for m in messages)
|
|
if input_tokens + max_tokens > HERMES2_TOKEN_LIMIT:
|
|
adjusted_max_tokens = max(
|
|
1000, HERMES2_TOKEN_LIMIT - input_tokens
|
|
)
|
|
print(
|
|
f"Adjusted max_tokens to {adjusted_max_tokens} for {endpoint} due to context limit"
|
|
)
|
|
response = client.chat.completions.create(
|
|
model=self.model,
|
|
messages=messages,
|
|
temperature=0,
|
|
max_tokens=adjusted_max_tokens,
|
|
)
|
|
return response.choices[0].message.content, endpoint
|
|
except Exception as e:
|
|
print(f"Error calling endpoint {endpoint}: {e}")
|
|
return None, endpoint
|
|
|
|
if prefer_hermes:
|
|
hermes_endpoint = "https://hermes.ai.unturf.com/v1"
|
|
hermes_client = next(
|
|
(
|
|
c
|
|
for c, e in zip(self.clients, self.hermes_endpoints)
|
|
if e == hermes_endpoint
|
|
),
|
|
None,
|
|
)
|
|
if hermes_client:
|
|
print(f"Attempting {hermes_endpoint} for final answer")
|
|
result, used_endpoint = call_client(hermes_client, hermes_endpoint)
|
|
if result is not None:
|
|
print(f"Success using {used_endpoint}")
|
|
return result
|
|
print(f"Fallback: {hermes_endpoint} failed, trying other endpoints")
|
|
other_clients = [
|
|
(c, e)
|
|
for c, e in zip(self.clients, self.hermes_endpoints)
|
|
if e != hermes_endpoint
|
|
]
|
|
with ThreadPoolExecutor(max_workers=len(other_clients)) as executor:
|
|
futures = [
|
|
executor.submit(call_client, c, e) for c, e in other_clients
|
|
]
|
|
for future in as_completed(futures):
|
|
result, used_endpoint = future.result()
|
|
if result is not None:
|
|
print(f"Success using {used_endpoint}")
|
|
return result
|
|
else:
|
|
print(
|
|
f"Warning: {hermes_endpoint} not in endpoints, using parallel execution"
|
|
)
|
|
else:
|
|
with ThreadPoolExecutor(max_workers=len(self.clients)) as executor:
|
|
futures = [
|
|
executor.submit(call_client, c, e)
|
|
for c, e in zip(self.clients, self.hermes_endpoints)
|
|
]
|
|
for future in as_completed(futures):
|
|
result, used_endpoint = future.result()
|
|
if result is not None:
|
|
print(f"Success using {used_endpoint}")
|
|
return result
|
|
raise RuntimeError("All Hermes endpoints failed.")
|
|
|
|
def extract_with_hermes(self, html):
|
|
"""Extract main article content using Hermes."""
|
|
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 formatting and structure, "
|
|
"excluding ads and navigation. Return only the full text."
|
|
),
|
|
},
|
|
{"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):
|
|
"""Summarize content as bullet points, used only when token limit is exceeded."""
|
|
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 cache_article(self, url, title, html, extracted, summary):
|
|
"""Cache article in database, using extracted_content as summary."""
|
|
with self.SessionLocal() as db:
|
|
art = Article(
|
|
url=url,
|
|
title=title,
|
|
raw_html=html,
|
|
extracted_content=extracted,
|
|
summary=summary, # Stores extracted_content unless summarized
|
|
fetched_at=datetime.now(timezone.utc),
|
|
)
|
|
db.add(art)
|
|
try:
|
|
db.commit()
|
|
except Exception as e:
|
|
db.rollback()
|
|
print(f"Error caching article {url}: {e}")
|
|
raise
|
|
|
|
def process_url(self, title_url):
|
|
"""Process a URL, extracting content but not summarizing."""
|
|
title, url = title_url
|
|
with self.SessionLocal() as db:
|
|
if db.query(Article).filter_by(url=url).first():
|
|
print(f"Already cached: {url}")
|
|
return
|
|
print(f"Fetching: {url}")
|
|
html = self.fetch_webpage(url)
|
|
if not html:
|
|
return
|
|
extracted = self.extract_with_hermes(html)
|
|
# Skip summarization; use extracted content as summary
|
|
self.cache_article(url, title, html, extracted, extracted)
|
|
|
|
def aggregate_and_answer(self, query):
|
|
"""Aggregate article content and generate an answer, summarizing if needed."""
|
|
with self.SessionLocal() as db:
|
|
articles = db.query(Article).all()
|
|
content_list = [
|
|
(art.summary, art.extracted_content) for art in articles
|
|
] # summary is extracted_content
|
|
|
|
def estimate_tokens(text):
|
|
return len(text) // 4 + 1
|
|
|
|
# Try with full extracted content
|
|
combined = "\n\n".join(content for content, _ in content_list)
|
|
prompt_template = f"Based on these article contents:\n{{}}\n\nProvide a comprehensive answer to: {query}"
|
|
template_tokens = estimate_tokens(prompt_template.format(""))
|
|
combined_tokens = estimate_tokens(combined)
|
|
|
|
if combined_tokens + template_tokens <= HERMES_INPUT_TOKENS:
|
|
print(f"Using full content ({combined_tokens} tokens)")
|
|
messages = [{"role": "user", "content": prompt_template.format(combined)}]
|
|
return self._fanout_call(
|
|
messages, max_tokens=HERMES_COMPLETION_TOKENS, prefer_hermes=True
|
|
)
|
|
|
|
# Summarize if token limit exceeded
|
|
print(
|
|
f"Warning: Input exceeds token limit ({combined_tokens + template_tokens} tokens). Summarizing content."
|
|
)
|
|
summaries = []
|
|
for _, extracted_content in content_list:
|
|
summary = self.summarize_with_hermes(extracted_content)
|
|
summaries.append(summary)
|
|
|
|
combined = "\n\n".join(summaries)
|
|
combined_tokens = estimate_tokens(combined)
|
|
if combined_tokens + template_tokens > HERMES_INPUT_TOKENS:
|
|
print(
|
|
f"Warning: Summarized input still exceeds token limit ({combined_tokens + template_tokens} tokens). Truncating."
|
|
)
|
|
max_chars = (HERMES_INPUT_TOKENS - template_tokens) * 4
|
|
combined = combined[:max_chars]
|
|
last_newline = combined.rfind("\n\n", 0, max_chars)
|
|
if last_newline != -1:
|
|
combined = combined[:last_newline]
|
|
combined_tokens = estimate_tokens(combined)
|
|
print(f"Truncated to {combined_tokens} tokens.")
|
|
|
|
messages = [{"role": "user", "content": prompt_template.format(combined)}]
|
|
return self._fanout_call(
|
|
messages, max_tokens=HERMES_COMPLETION_TOKENS, prefer_hermes=True
|
|
)
|
|
|
|
def run(self, query, max_results=10):
|
|
hits = self.search_duckduckgo(query, max_results)
|
|
with ThreadPoolExecutor(max_workers=4) as executor:
|
|
futures = [executor.submit(self.process_url, hit) for hit in hits]
|
|
for future in as_completed(futures):
|
|
future.result()
|
|
print("Generating comprehensive answer...")
|
|
print(self.aggregate_and_answer(query))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("query", help="Search and deep-query prompt")
|
|
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(
|
|
"--max-results", type=int, default=10, help="Max search results to process"
|
|
)
|
|
parser.add_argument(
|
|
"--user-agent-append", default="", help="String to append to default user agent"
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
crawler = SQLAlchemyDuckDuckGoCrawler(
|
|
api_key=args.api_key, model=args.model, user_agent_append=args.user_agent_append
|
|
)
|
|
crawler.run(args.query, args.max_results)
|