`windlass.providers.loaders.web`¶

web ¶

HTML, web-page and YouTube loaders.

The HTML loader works with no dependencies (falling back to a regex-based tag stripper) but does a considerably better job with BeautifulSoup installed. The web loader adds HTTP fetching, and the YouTube loader pulls transcripts.

Install the optional pieces with::

pip install "windlass[loaders]"

Example

HTMLLoader().load(b"
Hi
")[0].content 'Hi'

HTMLLoader ¶

HTMLLoader(
    *,
    selector: str | None = None,
    remove_tags: tuple[str, ...] = _NOISE_TAGS,
    keep_links: bool = False,
    **config: Any
)

Bases: Loader

Extracts readable text from HTML.

Parameters:

Name	Type	Description	Default
`selector`	`str \| None`	CSS selector for the content region. When omitted the loader tries :data:`_CONTENT_SELECTORS` in order.	`None`
`remove_tags`	`tuple[str, ...]`	Elements to strip entirely before extraction.	`_NOISE_TAGS`
`keep_links`	`bool`	Render links as `text (url)` so the model can cite them.	`False`
`**config`	`Any`	Forwarded to :class:`~windlass.interfaces.loader.Loader`.	`{}`

Note

Works without beautifulsoup4 using a regex stripper, which is fine for well-formed markup. Install windlass[loaders] for real parsing, boilerplate removal and CSS selectors.

Source code in src\windlass\providers\loaders\web.py

def __init__(
    self,
    *,
    selector: str | None = None,
    remove_tags: tuple[str, ...] = _NOISE_TAGS,
    keep_links: bool = False,
    **config: Any,
) -> None:
    super().__init__(**config)
    self.selector = selector
    self.remove_tags = remove_tags
    self.keep_links = keep_links

aload_source `async` ¶

aload_source(source: SourceLike) -> list[Document]

Extract text from one HTML document.

Parameters:

Name	Type	Description	Default
`source`	`SourceLike`	A path or a bytes payload.	required

Returns:

Type	Description
`list[Document]`	A single-element list, empty when no text could be extracted.

Source code in src\windlass\providers\loaders\web.py

async def aload_source(self, source: SourceLike) -> list[Document]:
    """Extract text from one HTML document.

    Args:
        source: A path or a bytes payload.

    Returns:
        A single-element list, empty when no text could be extracted.
    """
    raw = self._read_bytes(source)
    html = raw.decode("utf-8", errors="replace")
    path = str(source) if not isinstance(source, bytes) else None
    base = self._base_metadata(source) if not isinstance(source, bytes) else {}
    text, meta = await to_thread(self.extract, html)
    if not text.strip():
        return []
    return [
        Document(
            content=text,
            metadata={**base, **meta},
            source=path,
            mimetype="text/html",
        )
    ]

extract ¶

extract(html: str) -> tuple[str, dict[str, Any]]

Turn HTML into text plus metadata.

Parameters:

Name	Type	Description	Default
`html`	`str`	The HTML source.	required

Returns:

Type	Description
`str`	A `(text, metadata)` pair. Metadata may include `title`,
`dict[str, Any]`	`description`, `headings` and `links`.

Example

text, meta = HTMLLoader().extract("T
Body
") meta["title"], text ('T', 'Body')

Source code in src\windlass\providers\loaders\web.py

def extract(self, html: str) -> tuple[str, dict[str, Any]]:
    """Turn HTML into text plus metadata.

    Args:
        html: The HTML source.

    Returns:
        A ``(text, metadata)`` pair. Metadata may include ``title``,
        ``description``, ``headings`` and ``links``.

    Example:
        >>> text, meta = HTMLLoader().extract("<title>T</title><p>Body</p>")
        >>> meta["title"], text
        ('T', 'Body')
    """
    if not is_available("bs4"):
        title = re.search(r"<title[^>]*>(.*?)</title>", html, re.IGNORECASE | re.DOTALL)
        fallback: dict[str, Any] = (
            {"title": normalize_whitespace(title.group(1))} if title else {}
        )
        return strip_html(html), fallback

    bs4 = require("bs4", extra="loaders", feature="The HTML loader")
    parser = "lxml" if is_available("lxml") else "html.parser"
    soup = bs4.BeautifulSoup(html, parser)

    meta: dict[str, Any] = {}
    if soup.title and soup.title.string:
        meta["title"] = normalize_whitespace(str(soup.title.string))
    description = soup.find("meta", attrs={"name": "description"})
    if description and description.get("content"):
        meta["description"] = normalize_whitespace(str(description["content"]))

    for tag in soup(list(self.remove_tags)):
        tag.decompose()

    root = None
    selectors = [self.selector] if self.selector else list(_CONTENT_SELECTORS)
    for candidate in selectors:
        if not candidate:
            continue
        found = soup.select_one(candidate)
        if found is not None and found.get_text(strip=True):
            root = found
            break
    root = root or soup

    headings = [
        normalize_whitespace(h.get_text())
        for h in root.find_all(["h1", "h2", "h3"])
        if h.get_text(strip=True)
    ]
    if headings:
        meta["headings"] = headings[:50]

    if self.keep_links:
        links = []
        for anchor in root.find_all("a", href=True):
            label = normalize_whitespace(anchor.get_text())
            if label:
                anchor.replace_with(f"{label} ({anchor['href']})")
                links.append(anchor["href"])
        if links:
            meta["links"] = links[:100]

    return normalize_whitespace(root.get_text(separator="\n")), meta

WebLoader ¶

WebLoader(
    *,
    headers: dict[str, str] | None = None,
    follow_redirects: bool = True,
    timeout: float | None = None,
    max_bytes: int = 10000000,
    **config: Any
)

Bases: HTMLLoader

Fetches URLs over HTTP and extracts their text.

Parameters:

Name	Type	Description	Default
`headers`	`dict[str, str] \| None`	Extra request headers. A browser-like `User-Agent` is sent by default, because many sites reject the Python default.	`None`
`follow_redirects`	`bool`	Follow 3xx responses.	`True`
`timeout`	`float \| None`	Per-request timeout in seconds.	`None`
`max_bytes`	`int`	Refuse responses larger than this, so one enormous page cannot exhaust memory.	`10000000`
`**config`	`Any`	Forwarded to :class:`HTMLLoader`.	`{}`

Raises:

Type	Description
`IngestionError`	On a network failure or a non-2xx response.

Performance

URLs are fetched concurrently by the base class, bounded by the global max_concurrency setting. Be considerate of the sites you crawl.

Source code in src\windlass\providers\loaders\web.py

def __init__(
    self,
    *,
    headers: dict[str, str] | None = None,
    follow_redirects: bool = True,
    timeout: float | None = None,
    max_bytes: int = 10_000_000,
    **config: Any,
) -> None:
    super().__init__(**config)
    self.headers = {
        "User-Agent": "Mozilla/5.0 (compatible; WindlassBot/0.1; +https://github.com/windlass)",
        "Accept": "text/html,application/xhtml+xml",
        **(headers or {}),
    }
    self.follow_redirects = follow_redirects
    self.timeout = timeout or settings().request_timeout
    self.max_bytes = max_bytes

can_handle `classmethod` ¶

can_handle(source: SourceLike) -> bool

Return whether source looks like an HTTP(S) URL.

Source code in src\windlass\providers\loaders\web.py

@classmethod
def can_handle(cls, source: SourceLike) -> bool:
    """Return whether ``source`` looks like an HTTP(S) URL."""
    return isinstance(source, str) and source.startswith(("http://", "https://"))

aload_source `async` ¶

aload_source(source: SourceLike) -> list[Document]

Fetch a URL and extract its text.

Parameters:

Name	Type	Description	Default
`source`	`SourceLike`	The URL to fetch.	required

Returns:

Type	Description
`list[Document]`	A single-element list, empty when the page had no extractable text.

Raises:

Type	Description
`IngestionError`	On a network error or an HTTP error status.

Source code in src\windlass\providers\loaders\web.py

async def aload_source(self, source: SourceLike) -> list[Document]:
    """Fetch a URL and extract its text.

    Args:
        source: The URL to fetch.

    Returns:
        A single-element list, empty when the page had no extractable text.

    Raises:
        IngestionError: On a network error or an HTTP error status.
    """
    url = str(source)
    try:
        async with httpx.AsyncClient(
            headers=self.headers,
            follow_redirects=self.follow_redirects,
            timeout=self.timeout,
        ) as client:
            response = await client.get(url)
            response.raise_for_status()
            if len(response.content) > self.max_bytes:
                raise IngestionError(
                    f"{url} returned {len(response.content)} bytes, "
                    f"over the {self.max_bytes} limit."
                )
            html = response.text
    except httpx.HTTPStatusError as exc:
        raise IngestionError(
            f"{url} returned HTTP {exc.response.status_code}.",
            context={"url": url, "status": exc.response.status_code},
        ) from exc
    except httpx.HTTPError as exc:
        raise IngestionError(f"Could not fetch {url}: {exc}", context={"url": url}) from exc

    text, meta = await to_thread(self.extract, html)
    if not text.strip():
        return []
    parsed = urlparse(url)
    return [
        Document(
            content=text,
            metadata={**meta, "url": url, "domain": parsed.netloc, "source": url},
            source=url,
            mimetype="text/html",
        )
    ]

acrawl `async` ¶

acrawl(urls: list[str], *, concurrency: int | None = None) -> list[Document]

Fetch several URLs concurrently.

Parameters:

Name	Type	Description	Default
`urls`	`list[str]`	The URLs to fetch.	required
`concurrency`	`int \| None`	Maximum simultaneous requests.	`None`

Returns:

Type	Description
`list[Document]`	Every document that loaded successfully; failures are logged and
`list[Document]`	skipped.

Source code in src\windlass\providers\loaders\web.py

async def acrawl(self, urls: list[str], *, concurrency: int | None = None) -> list[Document]:
    """Fetch several URLs concurrently.

    Args:
        urls: The URLs to fetch.
        concurrency: Maximum simultaneous requests.

    Returns:
        Every document that loaded successfully; failures are logged and
        skipped.
    """
    limit = concurrency or settings().max_concurrency
    results = await gather_bounded(
        [self.aload_source(u) for u in urls], limit=limit, return_exceptions=True
    )
    documents: list[Document] = []
    for url, result in zip(urls, results, strict=True):
        if isinstance(result, BaseException):
            self._log.warning("Skipping %s: %s", url, result)
            continue
        documents.extend(result)
    return documents

YouTubeLoader ¶

YouTubeLoader(
    *,
    languages: tuple[str, ...] = ("en",),
    chunk_by_time: bool = False,
    segment_seconds: int = 300,
    **config: Any
)

Bases: Loader

Loads transcripts for YouTube videos.

Parameters:

Name	Type	Description	Default
`languages`	`tuple[str, ...]`	Preferred transcript languages, in order of preference.	`('en',)`
`chunk_by_time`	`bool`	Emit one document per `segment_seconds` window, so citations carry a timestamp you can link to.	`False`
`segment_seconds`	`int`	Window length when `chunk_by_time` is on.	`300`
`**config`	`Any`	Forwarded to :class:`~windlass.interfaces.loader.Loader`.	`{}`

Raises:

Type	Description
`MissingDependencyError`	When `youtube-transcript-api` is not installed.
`IngestionError`	When the video has no accessible transcript.

Note

Only videos with captions (auto-generated or human) can be loaded. There is no audio transcription here — use the audio loader for that.

Source code in src\windlass\providers\loaders\web.py

def __init__(
    self,
    *,
    languages: tuple[str, ...] = ("en",),
    chunk_by_time: bool = False,
    segment_seconds: int = 300,
    **config: Any,
) -> None:
    super().__init__(**config)
    self.languages = tuple(languages)
    self.chunk_by_time = chunk_by_time
    self.segment_seconds = segment_seconds

can_handle `classmethod` ¶

can_handle(source: SourceLike) -> bool

Return whether source is a YouTube URL or a bare video id.

Source code in src\windlass\providers\loaders\web.py

@classmethod
def can_handle(cls, source: SourceLike) -> bool:
    """Return whether ``source`` is a YouTube URL or a bare video id."""
    if not isinstance(source, str):
        return False
    return bool(_YOUTUBE_ID_RE.search(source)) or (
        len(source) == 11 and source.replace("-", "").replace("_", "").isalnum()
    )

aload_source `async` ¶

aload_source(source: SourceLike) -> list[Document]

Fetch one video's transcript.

Parameters:

Name	Type	Description	Default
`source`	`SourceLike`	A YouTube URL or an 11-character video id.	required

Returns:

Type	Description
`list[Document]`	One document per time segment, or one for the whole transcript.

Raises:

Type	Description
`IngestionError`	When no transcript is available.

Source code in src\windlass\providers\loaders\web.py

async def aload_source(self, source: SourceLike) -> list[Document]:
    """Fetch one video's transcript.

    Args:
        source: A YouTube URL or an 11-character video id.

    Returns:
        One document per time segment, or one for the whole transcript.

    Raises:
        IngestionError: When no transcript is available.
    """
    api = require("youtube_transcript_api", extra="loaders", feature="The YouTube loader")
    video_id = extract_video_id(str(source))

    def _fetch() -> list[dict[str, Any]]:
        try:
            return api.YouTubeTranscriptApi.get_transcript(
                video_id, languages=list(self.languages)
            )
        except Exception as exc:
            raise IngestionError(
                f"No transcript available for video {video_id}: {exc}",
                hint="The video may have captions disabled, be private, or not "
                f"offer any of {self.languages}.",
                context={"video_id": video_id},
            ) from exc

    entries = await to_thread(_fetch)
    url = f"https://www.youtube.com/watch?v={video_id}"
    base = {"video_id": video_id, "url": url, "source": url}

    if not self.chunk_by_time:
        text = " ".join(e["text"].strip() for e in entries if e.get("text"))
        duration = entries[-1]["start"] + entries[-1].get("duration", 0) if entries else 0
        return [
            Document(
                content=normalize_whitespace(text),
                metadata={**base, "duration_seconds": round(duration)},
                source=url,
                mimetype="text/plain",
            )
        ]

    documents: list[Document] = []
    buffer: list[str] = []
    window_start = 0.0
    for entry in entries:
        start = float(entry.get("start", 0.0))
        if buffer and start - window_start >= self.segment_seconds:
            documents.append(self._segment(buffer, window_start, base, url))
            buffer, window_start = [], start
        buffer.append(entry.get("text", "").strip())
    if buffer:
        documents.append(self._segment(buffer, window_start, base, url))
    return documents

extract_video_id ¶

extract_video_id(source: str) -> str

Pull the 11-character video id out of a YouTube URL.

Parameters:

Name	Type	Description	Default
`source`	`str`	A YouTube URL in any of its common shapes, or a bare id.	required

Returns:

Type	Description
`str`	The video id.

Raises:

Type	Description
`IngestionError`	When no id can be found.

Example

extract_video_id("https://youtu.be/dQw4w9WgXcQ") 'dQw4w9WgXcQ' extract_video_id("dQw4w9WgXcQ") 'dQw4w9WgXcQ'

Source code in src\windlass\providers\loaders\web.py

def extract_video_id(source: str) -> str:
    """Pull the 11-character video id out of a YouTube URL.

    Args:
        source: A YouTube URL in any of its common shapes, or a bare id.

    Returns:
        The video id.

    Raises:
        IngestionError: When no id can be found.

    Example:
        >>> extract_video_id("https://youtu.be/dQw4w9WgXcQ")
        'dQw4w9WgXcQ'
        >>> extract_video_id("dQw4w9WgXcQ")
        'dQw4w9WgXcQ'
    """
    match = _YOUTUBE_ID_RE.search(source)
    if match:
        return match.group(1)
    if len(source) == 11:
        return source
    raise IngestionError(
        f"Could not find a YouTube video id in {source!r}.",
        hint="Pass a watch/embed/short URL or the bare 11-character id.",
    )

windlass.providers.loaders.web¶

web ¶

HTMLLoader ¶

aload_source async ¶

extract ¶

WebLoader ¶

can_handle classmethod ¶

aload_source async ¶

acrawl async ¶

YouTubeLoader ¶

can_handle classmethod ¶

aload_source async ¶

extract_video_id ¶

`windlass.providers.loaders.web`¶

aload_source `async` ¶

can_handle `classmethod` ¶

aload_source `async` ¶

acrawl `async` ¶

can_handle `classmethod` ¶

aload_source `async` ¶