sync/tools.py

import subprocess
from pathlib import Path

try:
    import httpx
    _http = httpx.Client(timeout=30, follow_redirects=True)
except ImportError:
    _http = None


def search(query: str) -> str:
    """Search the web and return relevant results."""
    if _http is None:
        return (
            f"[search stub for '{query}']: "
            f"install httpx for real search")
    resp = _http.get(
        "https://html.duckduckgo.com/html/",
        params={"q": query})
    from html.parser import HTMLParser
    results = []

    class DDGParser(HTMLParser):
        in_result = False
        current = ""

        def handle_starttag(self, tag, attrs):
            attrs = dict(attrs)
            if tag == "a" and "result__a" in attrs.get(
                    "class", ""):
                self.in_result = True
                self.current = attrs.get("href", "")

        def handle_data(self, data):
            if self.in_result:
                results.append(
                    f"{data.strip()} ({self.current})")
                self.in_result = False

    DDGParser().feed(resp.text)
    return "\n".join(results[:10]) or "No results found."


def fetch(url: str) -> str:
    """Fetch a website URL and return its text content."""
    if _http is None:
        return f"[fetch stub for {url}]: install httpx"
    resp = _http.get(url)
    text = resp.text
    if len(text) > 50000:
        text = text[:50000] + "\n... (truncated)"
    return text


def read_file(path: str) -> str:
    """Read a file and return its contents."""
    return Path(path).read_text()


def write_file(path: str, content: str) -> str:
    """Write content to a file."""
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    Path(path).write_text(content)
    return f"Wrote {len(content)} bytes to {path}"


def run_command(command: str) -> str:
    """Run a shell command and return stdout+stderr."""
    result = subprocess.run(
        command, shell=True,
        capture_output=True, text=True, timeout=60)
    output = result.stdout
    if result.stderr:
        output += f"\nSTDERR: {result.stderr}"
    if result.returncode != 0:
        output += f"\nEXIT CODE: {result.returncode}"
    return output


def bash(script: str) -> str:
    """Run a multi-line bash script."""
    result = subprocess.run(
        ["bash", "-c", script],
        capture_output=True, text=True, timeout=300)
    output = result.stdout
    if result.stderr:
        output += f"\nSTDERR: {result.stderr}"
    if result.returncode != 0:
        output += f"\nEXIT CODE: {result.returncode}"
    return output