scripts/fetch_docs.py

from __future__ import annotations

import argparse
import hashlib
import re
from pathlib import Path
from urllib.parse import urljoin, urlparse

import requests
from bs4 import BeautifulSoup
from markdownify import markdownify as html_to_markdown
from tqdm import tqdm


DEFAULT_SOURCE = "https://docs.edgeimpulse.com/llms.txt"
DEFAULT_OUT = Path("data/raw_docs")
TIMEOUT = 30


def safe_name(url: str) -> str:
    parsed = urlparse(url)
    path = parsed.path.strip("/") or "index"
    stem = re.sub(r"[^A-Za-z0-9_.-]+", "_", path)
    digest = hashlib.sha1(url.encode("utf-8")).hexdigest()[:10]
    return f"{stem}_{digest}.md"


def get_text(url: str) -> str:
    response = requests.get(url, timeout=TIMEOUT)
    response.raise_for_status()
    return response.text


def extract_urls(source_url: str, source_text: str) -> list[str]:
    urls: set[str] = set()
    for match in re.finditer(r"https?://[^\s)>\"]+", source_text):
        urls.add(match.group(0).rstrip(".,"))

    for match in re.finditer(r"\[[^\]]+\]\(([^)]+)\)", source_text):
        href = match.group(1).strip()
        if href and not href.startswith("#"):
            urls.add(urljoin(source_url, href))

    return sorted(url for url in urls if "edgeimpulse.com" in url)


def page_to_markdown(url: str, text: str) -> str:
    content_type_hint = urlparse(url).path.lower()
    if content_type_hint.endswith((".md", ".mdx", ".txt")):
        body = text
    else:
        soup = BeautifulSoup(text, "html.parser")
        for tag in soup(["script", "style", "nav", "footer"]):
            tag.decompose()
        main = soup.find("main") or soup.body or soup
        body = html_to_markdown(str(main), heading_style="ATX")

    return f"---\nsource: {url}\n---\n\n{body.strip()}\n"


def fetch_docs(source: str, out_dir: Path, limit: int | None = None) -> None:
    out_dir.mkdir(parents=True, exist_ok=True)
    source_text = get_text(source)
    (out_dir / "llms.txt").write_text(source_text, encoding="utf-8")

    urls = extract_urls(source, source_text)
    if limit:
        urls = urls[:limit]

    manifest: list[str] = []
    for url in tqdm(urls, desc="Fetching docs"):
        try:
            text = get_text(url)
            markdown = page_to_markdown(url, text)
            filename = safe_name(url)
            (out_dir / filename).write_text(markdown, encoding="utf-8")
            manifest.append(f"{filename}\t{url}")
        except Exception as exc:
            manifest.append(f"ERROR\t{url}\t{exc}")

    (out_dir / "manifest.tsv").write_text("\n".join(manifest) + "\n", encoding="utf-8")
    print(f"Wrote {len(manifest)} fetched entries to {out_dir}")


def main() -> None:
    parser = argparse.ArgumentParser(description="Fetch Edge Impulse docs from llms.txt links.")
    parser.add_argument("--source", default=DEFAULT_SOURCE)
    parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
    parser.add_argument("--limit", type=int, default=None, help="Optional small test limit.")
    args = parser.parse_args()
    fetch_docs(args.source, args.out, args.limit)


if __name__ == "__main__":
    main()