tools/web_tools.py

# 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'<a rel="nofollow" class="result__a" href="([^"]+)">([^<]+)</a>.*?'
                    r'<a class="result__snippet"[^>]*>([^<]+)</a>',
                    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'<h2><a href="([^"]+)"[^>]*>(.*?)</a></h2>', 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"<script[^>]*>.*?</script>", "", html, flags=re.DOTALL)
                text = re.sub(r"<style[^>]*>.*?</style>", "", 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'<a[^>]+href="([^"]+)"[^>]*>(.*?)</a>', 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'<img[^>]+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"<h([1-6])[^>]*>(.*?)</h\\1>", 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"<table[^>]*>(.*?)</table>", 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")