app.py

"""
HuggingBot Browser API
----------------------
Server-side web search + page scraping for HuggingBot, run as a HuggingFace Space.

Endpoints (CORS-enabled):
  POST/GET  /search     {query, max_results, source}  -> {status, results:[{title,snippet,url,source}]}
  POST/GET  /scrape     {url, max_length}             -> {status, page_title, final_url, http_status, markdown_content}
  POST      /web_browse {action:"Search"|"Scrape", query, ...}  (compat with the old broadfield-dev/browser-api shape)
  GET       /health
"""
import json
import re
import urllib.parse

import requests
from bs4 import BeautifulSoup

# ----------------------------------------------------------------------------
# HTTP helpers
# ----------------------------------------------------------------------------
HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                  "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9",
}


def http_get(url, params=None, timeout=25):
    return requests.get(url, params=params, headers=HEADERS, timeout=timeout, allow_redirects=True)


def clean_snippet(text, limit=320):
    text = re.sub(r"\s+", " ", text or "").strip()
    return text[:limit]


# ----------------------------------------------------------------------------
# SEARCH ENGINES (server-side -> no CORS problems)
# ----------------------------------------------------------------------------
def parse_ddg_html(text):
    soup = BeautifulSoup(text, "html.parser")
    results = []
    for a in soup.select("a.result__a"):
        title = a.get_text(" ", strip=True)
        if not title:
            continue
        href = a.get("href", "")
        m = re.search(r"uddg=([^&]+)", href)
        if m:
            href = urllib.parse.unquote(m.group(1))
        parent = a.find_parent(("div", "article"))
        snippet = ""
        if parent is not None:
            node = parent.select_one(".result__snippet")
            if node is not None:
                snippet = node.get_text(" ", strip=True)
        results.append({"title": title, "url": href, "snippet": clean_snippet(snippet)})
    return results


def search_ddg(query, n=5):
    r = http_get("https://html.duckduckgo.com/html/", params={"q": query})
    r.raise_for_status()
    return parse_ddg_html(r.text)[:n]


def search_bing(query, n=5):
    r = http_get("https://www.bing.com/search", params={"q": query})
    r.raise_for_status()
    soup = BeautifulSoup(r.text, "html.parser")
    results = []
    for li in soup.select("li.b_algo"):
        a = li.select_one("h2 a")
        if a is None:
            continue
        href = a.get("href", "")
        if not href or not href.startswith("http"):
            continue
        p = li.select_one("p, .b_caption p")
        snippet = p.get_text(" ", strip=True) if p is not None else ""
        results.append({
            "title": a.get_text(" ", strip=True),
            "url": href,
            "snippet": clean_snippet(snippet),
        })
    return results[:n]


def search_bing_news(query, n=5):
    r = http_get("https://www.bing.com/news/search", params={"q": query})
    r.raise_for_status()
    soup = BeautifulSoup(r.text, "html.parser")
    results = []
    for card in soup.select(".news-card, .newsitem"):
        a = card.select_one("a.title, h3 a, a")
        if a is None:
            continue
        href = a.get("href", "")
        title = a.get_text(" ", strip=True)
        if not title or not href:
            continue
        sn = card.select_one(".snippet")
        snippet = sn.get_text(" ", strip=True) if sn is not None else ""
        results.append({"title": title, "url": href, "snippet": clean_snippet(snippet)})
    return results[:n]


def search_wikipedia(query, n=5):
    r = http_get("https://en.wikipedia.org/w/api.php", params={
        "action": "query", "list": "search", "srsearch": query,
        "format": "json", "srlimit": n,
    })
    r.raise_for_status()
    data = r.json()
    results = []
    for item in data.get("query", {}).get("search", []):
        title = item.get("title", "")
        if not title:
            continue
        snippet = re.sub("<[^>]+>", "", item.get("snippet", "")).strip()
        results.append({
            "title": title,
            "url": "https://en.wikipedia.org/wiki/" + urllib.parse.quote(title.replace(" ", "_")),
            "snippet": clean_snippet(snippet),
        })
    return results


def search_web(query, max_results=5, source="auto"):
    source = (source or "auto").lower()
    try:
        n = max(1, min(int(max_results), 10))
    except (TypeError, ValueError):
        n = 5

    attempts = []
    if source in ("wikipedia", "wiki"):
        attempts.append(("wikipedia", lambda: search_wikipedia(query, n)))
    elif source == "news":
        attempts.append(("bing-news", lambda: search_bing_news(query, n)))
    elif source == "reddit":
        attempts.append(("duckduckgo", lambda: search_ddg("site:reddit.com " + query, n)))
        attempts.append(("bing", lambda: search_bing(query + " reddit", n)))
    else:  # auto / web
        attempts.append(("duckduckgo", lambda: search_ddg(query, n)))
        attempts.append(("bing", lambda: search_bing(query, n)))
        attempts.append(("wikipedia", lambda: search_wikipedia(query, n)))

    last_error = "no results found"
    for name, fn in attempts:
        try:
            results = fn()
            if results:
                return {"engine": name, "results": results}
        except Exception as e:  # noqa: BLE001
            last_error = "%s: %s" % (name, e)
    return {"engine": "none", "results": [], "error": last_error}


# ----------------------------------------------------------------------------
# SCRAPING
# ----------------------------------------------------------------------------
REMOVE_SELECTORS = [
    "script", "style", "noscript", "svg", "canvas", "iframe", "form", "button",
    "nav", "footer", "aside",
    "[role='navigation']", "[role='search']", "[role='banner']", "[aria-hidden='true']",
    ".nav", ".navbar", ".menu", ".sidebar", ".header", ".footer",
    ".ads", ".advertisement", ".cookie", ".popup", ".modal", ".share", ".social",
]


def html_to_markdown(node):
    lines = []
    for el in node.find_all(["h1", "h2", "h3", "h4", "h5", "h6", "p", "li", "pre", "blockquote", "hr"]):
        tag = el.name
        if tag in ("h1", "h2", "h3", "h4", "h5", "h6"):
            level = int(tag[1])
            text = el.get_text(" ", strip=True)
            if text:
                lines.append("\n" + "#" * level + " " + text + "\n")
        elif tag == "p":
            text = el.get_text(" ", strip=True)
            if text:
                lines.append(text + "\n")
        elif tag == "li":
            text = el.get_text(" ", strip=True)
            if text:
                lines.append("* " + text)
        elif tag == "pre":
            lines.append("\n```\n" + el.get_text("\n", strip=True) + "\n```\n")
        elif tag == "blockquote":
            text = el.get_text(" ", strip=True)
            if text:
                lines.append("> " + text + "\n")
        elif tag == "hr":
            lines.append("\n---\n")
    md = re.sub(r"\n{3,}", "\n\n", "\n".join(lines))
    return md.strip()


def scrape_url(url, max_len=50000):
    try:
        max_len = max(100, min(int(max_len), 100000))
    except (TypeError, ValueError):
        max_len = 50000

    url = (url or "").strip()
    if not url:
        return {"status": "error", "error_message": "Missing 'url'"}
    if not url.lower().startswith(("http://", "https://")):
        url = "https://" + url

    try:
        r = http_get(url, timeout=35)
    except Exception as e:  # noqa: BLE001
        return {"status": "error", "url": url, "error_message": str(e)}

    final_url = r.url or url
    status = r.status_code
    ctype = r.headers.get("content-type", "").lower()
    base = {"status": "success", "url": url, "final_url": final_url, "http_status": status}

    if status >= 400:
        base["status"] = "error"
        base["error_message"] = "HTTP %d: the page could not be fetched" % status
        return base

    raw = r.text or ""

    # JSON responses -> return pretty JSON
    if "json" in ctype:
        try:
            content = json.dumps(r.json(), ensure_ascii=False, indent=2)
            if len(content) > max_len:
                content = content[:max_len] + "\n\n...[truncated]"
            base.update({"page_title": "JSON response", "description": "", "content": content,
                         "markdown_content": content, "links": []})
            return base
        except Exception:  # noqa: BLE001
            pass

    soup = BeautifulSoup(raw, "html.parser")
    for selector in REMOVE_SELECTORS:
        for el in soup.select(selector):
            el.decompose()

    title = soup.title.get_text(" ", strip=True) if soup.title else ""
    desc = ""
    og = soup.select_one('meta[property="og:description"]')
    if og is not None:
        desc = (og.get("content") or "").strip()
    meta = soup.select_one('meta[name="description"]')
    if not desc and meta is not None:
        desc = (meta.get("content") or "").strip()

    node = soup.find("main") or soup.find("article") or soup.find("body") or soup
    md = html_to_markdown(node)

    # collect links
    links, seen = [], set()
    for a in soup.find_all("a", href=True):
        href = (a.get("href") or "").strip()
        if not href or href.startswith("#") or href.startswith("javascript:"):
            continue
        full = urllib.parse.urljoin(final_url, href)
        if full in seen:
            continue
        seen.add(full)
        text = a.get_text(" ", strip=True)[:80]
        links.append({"url": full, "title": text})
        if len(links) >= 20:
            break

    if links:
        md += "\n\n**Links:**\n" + "\n".join(
            "- [%s](%s)" % (l["title"] or l["url"], l["url"]) for l in links)

    if len(md) > max_len:
        md = md[:max_len] + "\n\n...[truncated]"

    base.update({
        "page_title": title,
        "description": desc,
        "content": md,
        "markdown_content": md,
        "links": links,
    })
    return base


# ----------------------------------------------------------------------------
# FastAPI app + CORS + Gradio UI
# ----------------------------------------------------------------------------
import uvicorn
import gradio as gr
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(title="HuggingBot Browser API")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=False,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/health")
def health():
    return {"status": "ok", "service": "huggingbot-browser"}


def _search_args(body, params):
    if body:
        q = body.get("query") or body.get("q")
        n = body.get("max_results") or body.get("max") or 5
        s = body.get("source") or body.get("search_engine") or "auto"
    else:
        q = params.get("q") or params.get("query")
        n = params.get("max_results") or params.get("max") or 5
        s = params.get("source") or "auto"
    return q, n, s


@app.get("/search")
@app.post("/search")
async def endpoint_search(request: Request):
    body = None
    if request.method == "POST":
        try:
            body = await request.json()
        except Exception:  # noqa: BLE001
            body = None
    q, n, s = _search_args(body, request.query_params)
    if not q:
        return {"status": "error", "error_message": "Missing required parameter: query"}
    data = search_web(q, n, s)
    ok = bool(data.get("results"))
    return {
        "status": "success" if ok else "error",
        "query": q,
        "source": s,
        "engine": data.get("engine", "none"),
        "results": data.get("results", []),
        "error_message": None if ok else data.get("error"),
    }


@app.get("/scrape")
@app.post("/scrape")
async def endpoint_scrape(request: Request):
    body = None
    if request.method == "POST":
        try:
            body = await request.json()
        except Exception:  # noqa: BLE001
            body = None
    if body:
        url = body.get("url") or body.get("query")
        mlen = body.get("max_length") or body.get("max_len") or 50000
    else:
        url = request.query_params.get("url") or request.query_params.get("q")
        mlen = request.query_params.get("max_length") or request.query_params.get("max_len") or 50000
    if not url:
        return {"status": "error", "error_message": "Missing required parameter: url"}
    return scrape_url(url, mlen)


@app.post("/web_browse")
async def endpoint_web_browse(request: Request):
    """Compatibility endpoint mirroring the old broadfield-dev/browser-api shape."""
    try:
        body = await request.json()
    except Exception:  # noqa: BLE001
        return {"status": "error", "error_message": "Expected JSON body"}
    action = (body.get("action") or "").strip().lower()
    query = body.get("query") or body.get("url") or body.get("q")
    if not query:
        return {"status": "error", "error_message": "Missing 'query'"}
    if action in ("search", "scrape url", "scrape") or action == "search":
        q_action = "search" if "search" in action else "scrape"
    else:
        return {"status": "error", "error_message": "action must be 'Search' or 'Scrape'"}

    if action == "search":
        src = body.get("search_engine_name") or body.get("search_engine") or "auto"
        data = search_web(query, body.get("max_results") or 5, src)
        results = data.get("results", [])
        lines = ['## Search results for "%s"' % query, ""]
        for i, r in enumerate(results, 1):
            lines.append("### %d. %s" % (i, r["title"]))
            if r.get("snippet"):
                lines.append("> " + r["snippet"])
            if r.get("url"):
                lines.append("🔗 " + r["url"])
            lines.append("")
        return {
            "status": "success" if results else "error",
            "query": query,
            "action": "Search",
            "final_url": "",
            "page_title": "Search results",
            "http_status": 200,
            "markdown_content": "\n".join(lines),
            "results": results,
            "error_message": None if results else data.get("error"),
        }
    else:
        mlen = body.get("max_length") or body.get("max_len") or 50000
        data = scrape_url(query, mlen)
        data["query"] = query
        data["action"] = "Scrape"
        data["status"] = data.get("status")
        return data


# Minimal Gradio UI so the Space behaves like a proper Gradio app too
def ui_search(q, source, n):
    data = search_web(q, n, source)
    return json.dumps({k: v for k, v in data.items()}, ensure_ascii=False, indent=2)


def ui_scrape(url, mlen):
    data = scrape_url(url, mlen)
    out = data.get("markdown_content") or data.get("error_message") or "no content"
    return "# %s\n\n%s\n" % (data.get("page_title", "Page"), out)


demo = gr.Blocks(title="HuggingBot Browser API")
with demo:
    gr.Markdown("# HuggingBot Browser API\n\n"
                "Server-side **search** + **scrape** for HuggingBot.\n\n"
                "- `POST /search`  → `{query, max_results, source}`\n"
                "- `POST /scrape`  → `{url, max_length}`\n"
                "- `POST /web_browse` → legacy-compatible shape\n"
                "- `GET /health`")
    with gr.Tab("Search"):
        with gr.Row():
            q = gr.Textbox(label="Query", value="latest AI news 2025", scale=3)
            src = gr.Dropdown(["auto", "web", "wikipedia", "news", "reddit"], value="auto", label="Source")
            n = gr.Slider(1, 10, value=5, step=1, label="Max results")
        btn = gr.Button("Search")
        out = gr.Code(label="JSON", language="json")
        btn.click(ui_search, inputs=[q, src, n], outputs=out)
    with gr.Tab("Scrape"):
        with gr.Row():
            u = gr.Textbox(label="URL", value="https://en.wikipedia.org/wiki/Web_scraping", scale=3)
            ml = gr.Slider(1000, 50000, value=5000, step=500, label="Max length")
        btn2 = gr.Button("Scrape")
        out2 = gr.Markdown()
        btn2.click(ui_scrape, inputs=[u, ml], outputs=out2)

app.mount("/", demo.app)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)