# File 8: tools/web_tools.py
with open(f"{base_dir}/tools/web_tools.py", "w") as f:
f.write('''"""Web Search and Network Tools"""
import json
import urllib.request
import urllib.parse
import ssl
import re
from typing import Dict, List, Optional
from config import Config
from utils import Logger, truncate
class WebTools:
"""Web search, scraping, and network operations"""
def __init__(self):
self.logger = Logger()
self.ssl_context = ssl.create_default_context()
self.ssl_context.check_hostname = False
self.ssl_context.verify_mode = ssl.CERT_NONE
def _fetch_url(self, url: str, headers: Dict = None, timeout: int = 15) -> tuple:
"""Helper to fetch URL content"""
default_headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
if headers:
default_headers.update(headers)
req = urllib.request.Request(url, headers=default_headers)
with urllib.request.urlopen(req, timeout=timeout, context=self.ssl_context) as response:
content_type = response.headers.get("Content-Type", "")
data = response.read()
if "text" in content_type or "json" in content_type or "html" in content_type:
try:
text = data.decode("utf-8", errors="replace")
return text, content_type, response.status
except:
pass
return data, content_type, response.status
def web_search(self, query: str, engine: str = "duckduckgo",
max_results: int = 10, safe_search: bool = True) -> Dict:
"""Search the web"""
try:
results = []
if engine == "duckduckgo":
search_url = f"https://html.duckduckgo.com/html/?q={urllib.parse.quote(query)}"
html, _, status = self._fetch_url(search_url)
result_blocks = re.findall(
r'([^<]+).*?'
r']*>([^<]+)',
html, re.DOTALL
)
for i, (url, title, snippet) in enumerate(result_blocks[:max_results]):
url = urllib.parse.unquote(url)
if url.startswith("//"):
url = "https:" + url
results.append({
"title": re.sub(r"<[^>]+>", "", title).strip(),
"url": url,
"snippet": re.sub(r"<[^>]+>", "", snippet).strip(),
"source": "duckduckgo"
})
elif engine == "bing":
search_url = f"https://www.bing.com/search?q={urllib.parse.quote(query)}"
html, _, _ = self._fetch_url(search_url)
titles = re.findall(r'', html)
for i, (url, title) in enumerate(titles[:max_results]):
results.append({
"title": re.sub(r"<[^>]+>", "", title).strip(),
"url": url,
"snippet": "",
"source": "bing"
})
self.logger.log("web_search", {"query": query, "engine": engine, "results": len(results)})
return {
"success": True,
"query": query,
"engine": engine,
"results": results,
"total": len(results)
}
except Exception as e:
return {"success": False, "error": str(e)}
def web_fetch(self, url: str, method: str = "GET",
headers: Dict = None, data: str = None,
timeout: int = 15, max_size: int = 100000) -> Dict:
"""Fetch content from URL"""
try:
default_headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
if headers:
default_headers.update(headers)
req = urllib.request.Request(
url,
data=data.encode() if data else None,
headers=default_headers,
method=method
)
with urllib.request.urlopen(req, timeout=timeout, context=self.ssl_context) as response:
content_type = response.headers.get("Content-Type", "")
data_bytes = response.read(max_size)
try:
text = data_bytes.decode("utf-8", errors="replace")
except:
text = data_bytes.hex()[:1000]
self.logger.log("web_fetch", {"url": url, "method": method, "size": len(data_bytes)})
return {
"success": True,
"url": url,
"status": response.status,
"content_type": content_type,
"headers": dict(response.headers),
"content": truncate(text, 3000),
"size": len(data_bytes),
"truncated": len(data_bytes) >= max_size
}
except Exception as e:
return {"success": False, "error": str(e)}
def web_scrape(self, url: str, extract: str = "text",
selector: str = "", max_length: int = 5000) -> Dict:
"""Scrape structured data from web page"""
try:
html, content_type, status = self._fetch_url(url)
if not isinstance(html, str):
return {"success": False, "error": "Binary content, cannot scrape"}
result = {"url": url, "extract": extract}
if extract == "text":
text = re.sub(r"]*>.*?", "", html, flags=re.DOTALL)
text = re.sub(r"]*>.*?", "", text, flags=re.DOTALL)
text = re.sub(r"<[^>]+>", " ", text)
text = re.sub(r"\s+", " ", text).strip()
result["content"] = text[:max_length]
result["total_length"] = len(text)
elif extract == "links":
links = re.findall(r']+href="([^"]+)"[^>]*>(.*?)', html, re.DOTALL)
result["links"] = [
{
"url": urllib.parse.urljoin(url, href),
"text": re.sub(r"<[^>]+>", "", text).strip()[:100]
}
for href, text in links[:50]
]
result["total"] = len(links)
elif extract == "images":
images = re.findall(r']+src="([^"]+)"[^>]*>', html)
result["images"] = [
{"url": urllib.parse.urljoin(url, src)}
for src in images[:30]
]
result["total"] = len(images)
elif extract == "headings":
headings = re.findall(r"]*>(.*?)", html, re.DOTALL)
result["headings"] = [
{"level": int(level), "text": re.sub(r"<[^>]+>", "", text).strip()}
for level, text in headings
]
elif extract == "tables":
tables = re.findall(r"]*>(.*?)", html, re.DOTALL)
result["tables"] = len(tables)
result["first_table"] = truncate(tables[0], 2000) if tables else ""
self.logger.log("web_scrape", {"url": url, "extract": extract})
result["success"] = True
return result
except Exception as e:
return {"success": False, "error": str(e)}
def api_request(self, url: str, method: str = "GET",
headers: Dict = None, params: Dict = None,
json_data: Dict = None, timeout: int = 15) -> Dict:
"""Make API requests with JSON handling"""
try:
if params:
url += "?" + urllib.parse.urlencode(params)
default_headers = {
"User-Agent": "Swarm-CLI/1.0",
"Accept": "application/json"
}
if headers:
default_headers.update(headers)
data = None
if json_data:
data = json.dumps(json_data).encode()
default_headers["Content-Type"] = "application/json"
req = urllib.request.Request(
url,
data=data,
headers=default_headers,
method=method
)
with urllib.request.urlopen(req, timeout=timeout, context=self.ssl_context) as response:
content = response.read().decode("utf-8", errors="replace")
try:
parsed = json.loads(content)
except:
parsed = content
self.logger.log("api_request", {"url": url, "method": method, "status": response.status})
return {
"success": 200 <= response.status < 300,
"status": response.status,
"headers": dict(response.headers),
"data": parsed,
"raw": truncate(content, 2000)
}
except Exception as e:
return {"success": False, "error": str(e)}
def dns_lookup(self, hostname: str, record_type: str = "A") -> Dict:
"""DNS lookup"""
try:
import socket
if record_type == "A":
ip = socket.gethostbyname(hostname)
return {
"success": True,
"hostname": hostname,
"record_type": "A",
"result": ip
}
else:
import subprocess
result = subprocess.run(
f"nslookup -type={record_type} {hostname}",
shell=True, capture_output=True, text=True, timeout=10
)
return {
"success": result.returncode == 0,
"hostname": hostname,
"record_type": record_type,
"output": result.stdout or result.stderr
}
except Exception as e:
return {"success": False, "error": str(e)}
def port_scan(self, host: str, ports: List[int] = None,
timeout: float = 1.0) -> Dict:
"""Scan ports on a host"""
try:
import socket
if not ports:
ports = [21, 22, 23, 25, 53, 80, 110, 143, 443, 993, 995, 3306, 3389, 5432, 8080, 8443]
open_ports = []
for port in ports:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
result = sock.connect_ex((host, port))
if result == 0:
try:
service = socket.getservbyport(port, "tcp")
except:
service = "unknown"
open_ports.append({"port": port, "service": service, "status": "open"})
sock.close()
return {
"success": True,
"host": host,
"scanned": len(ports),
"open_ports": open_ports,
"closed": len(ports) - len(open_ports)
}
except Exception as e:
return {"success": False, "error": str(e)}
def download_file(self, url: str, filename: str = "",
verify_checksum: str = "", chunk_size: int = 8192) -> Dict:
"""Download file with progress"""
try:
import hashlib
from config import Config
if not filename:
filename = url.split("/")[-1].split("?")[0] or "download"
save_path = os.path.join(Config.SANDBOX_PATH, filename)
os.makedirs(os.path.dirname(save_path), exist_ok=True)
req = urllib.request.Request(url, headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
})
hasher = hashlib.sha256()
downloaded = 0
with urllib.request.urlopen(req, context=self.ssl_context) as response:
total_size = int(response.headers.get("Content-Length", 0))
with open(save_path, "wb") as f:
while True:
chunk = response.read(chunk_size)
if not chunk:
break
f.write(chunk)
hasher.update(chunk)
downloaded += len(chunk)
checksum = hasher.hexdigest()
verified = True
if verify_checksum:
verified = checksum == verify_checksum
self.logger.log("download_file", {"url": url, "filename": filename, "size": downloaded})
return {
"success": True,
"url": url,
"filename": filename,
"path": save_path,
"size": downloaded,
"sha256": checksum,
"verified": verified if verify_checksum else None
}
except Exception as e:
return {"success": False, "error": str(e)}
''')
print("tools/web_tools.py done")