docs/scripts/refresh_integration_downloads.py

#!/usr/bin/env python3
"""Regenerate integration download tables (Python + TypeScript).

Reads each integration MDX page's frontmatter for:

    integration:
      name: OpenAIEmbeddings        # class / display name
      npm: "@langchain/openai"      # TypeScript (omit for N/A downloads)
      pypi: langchain-openai        # Python (omit for N/A downloads)
      featured: true                # optional; include in featured table
      deprecated: true              # optional
      # Chat-only capability keys (omit when unknown):
      stream: true
      tool_calling: true
      structured_output: true
      multimodal: true
      # Middleware-only text columns (omit when unknown):
      available: Prompt caching
      source: "[`org/repo`](https://github.com/org/repo)"
      # Retriever-only columns (omit when unknown):
      self_host: true
      cloud_offering: true
      package_md: "[`langchain-aws`](https://reference.langchain.com/python/langchain-aws/...)"
      # Vectorstore capability columns (omit when unknown):
      delete_by_id: true
      filtering: true
      search_by_vector: true
      search_with_score: true
      async_api: true
      passes_standard_tests: true
      multi_tenancy: true
      ids_in_add_documents: true

Also merges third-party rows from scripts/data/integration_external_docs.yaml.
Those rows stay in the same tables but link the name column to docs_url
(partner docs > GitHub > PyPI/npm) instead of a hosted guide.

docs_url values must be https://, http://, or a site-relative path starting
with a single / (protocol-relative //host URLs are rejected). Validate with:

    uv run python scripts/refresh_integration_downloads.py --check-docs-urls

Chat tables include capability columns. Middleware tables include
Provider, Middleware available, Source, and Downloads. Retriever tables
include Retriever, Self-host, Cloud offering, Package, and Downloads.
Vectorstore tables include feature-comparison columns only when at least one
page sets those frontmatter keys; otherwise Vectorstore + Downloads.
Other components use Integration + Downloads. Both featured and all-models
tables share columns.

Usage (from repo root):

    uv run python scripts/refresh_integration_downloads.py
    uv run python scripts/refresh_integration_downloads.py --write
    uv run python scripts/refresh_integration_downloads.py --write --component embeddings
    uv run python scripts/refresh_integration_downloads.py --check-docs-urls
"""

from __future__ import annotations

import argparse
import re
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Optional
from urllib.parse import quote

import requests
import yaml

_HTTP_HEADERS = {"User-Agent": "langchain-docs-download-refresh/1.0"}
_REQUEST_TIMEOUT = 20

_SCRIPT_DIR = Path(__file__).resolve().parent
_REPO_ROOT = _SCRIPT_DIR.parent
_EXTERNAL_DOCS_PATH = _SCRIPT_DIR / "data" / "integration_external_docs.yaml"

LANGUAGES = ("javascript", "python")

# Component directories under src/oss/{lang}/integrations/ to scan.
# Chat keeps capability columns. Middleware adds available/source text.
# Other components are Integration + Downloads only.
COMPONENTS: dict[str, dict[str, Any]] = {
    "chat": {"capabilities": True, "link_label": "Model"},
    "embeddings": {"capabilities": False, "link_label": "Integration"},
    "vectorstores": {
        "capabilities": False,
        "link_label": "Vectorstore",
        "vectorstore_columns": True,
    },
    "tools": {"capabilities": False, "link_label": "Integration"},
    "llms": {"capabilities": False, "link_label": "Integration"},
    "retrievers": {
        "capabilities": False,
        "link_label": "Retriever",
        "retriever_columns": True,
    },
    "document_loaders": {"capabilities": False, "link_label": "Integration"},
    "document_transformers": {"capabilities": False, "link_label": "Integration"},
    "document_compressors": {"capabilities": False, "link_label": "Integration"},
    "stores": {"capabilities": False, "link_label": "Integration"},
    "graphs": {"capabilities": False, "link_label": "Integration"},
    "sandboxes": {"capabilities": False, "link_label": "Integration"},
    "caches": {"capabilities": False, "link_label": "Integration"},
    "callbacks": {"capabilities": False, "link_label": "Integration"},
    "splitters": {"capabilities": False, "link_label": "Integration"},
    "chat_message_histories": {"capabilities": False, "link_label": "Integration"},
    "llm_caching": {"capabilities": False, "link_label": "Integration"},
    "middleware": {
        "capabilities": False,
        "link_label": "Provider",
        "middleware_columns": True,
    },
}

SKIP_FILES = {"index.mdx", "TEMPLATE.mdx"}

HEADER = (
    "{/* Generated by scripts/refresh_integration_downloads.py. "
    "Do not edit by hand. */}\n\n"
)

CHAT_CAPABILITY_KEYS = (
    "stream",
    "tool_calling",
    "structured_output",
    "multimodal",
)

VECTORSTORE_CAPABILITY_KEYS = (
    ("delete_by_id", "Delete by ID"),
    ("filtering", "Filtering"),
    ("search_by_vector", "Search by Vector"),
    ("search_with_score", "Search with score"),
    ("async_api", "Async"),
    ("passes_standard_tests", "Passes Standard Tests"),
    ("multi_tenancy", "Multi Tenancy"),
    ("ids_in_add_documents", "IDs in add Documents"),
)


@dataclass(frozen=True)
class IntegrationRow:
    rel_path: str  # e.g. chat/openai or document_loaders/file_loaders/json
    name: str
    package: Optional[str]
    registry: Optional[str]
    downloads: Optional[int]
    featured: bool
    deprecated: bool
    stream: Optional[bool]
    tool_calling: Optional[bool]
    structured_output: Optional[bool]
    multimodal: Optional[bool]
    # When set, the name column links here instead of a hosted docs page.
    docs_url: Optional[str] = None
    # Middleware-only optional columns from frontmatter.
    available: Optional[str] = None
    source: Optional[str] = None
    # Retriever-only optional columns from frontmatter.
    self_host: Optional[bool] = None
    cloud_offering: Optional[bool] = None
    package_md: Optional[str] = None
    # Vectorstore capability columns from frontmatter.
    delete_by_id: Optional[bool] = None
    filtering: Optional[bool] = None
    search_by_vector: Optional[bool] = None
    search_with_score: Optional[bool] = None
    async_api: Optional[bool] = None
    passes_standard_tests: Optional[bool] = None
    multi_tenancy: Optional[bool] = None
    ids_in_add_documents: Optional[bool] = None


def _integrations_dir(language: str) -> Path:
    return _REPO_ROOT / "src" / "oss" / language / "integrations"


def _load_external_docs() -> dict[str, Any]:
    if not _EXTERNAL_DOCS_PATH.is_file():
        return {}
    data = yaml.safe_load(_EXTERNAL_DOCS_PATH.read_text(encoding="utf-8"))
    return data if isinstance(data, dict) else {}


def _snippet_path(language: str, component: str, kind: str) -> Path:
    # kind: "all" | "featured"
    suffix = "downloads" if kind == "all" else "featured"
    return (
        _REPO_ROOT
        / "src"
        / "snippets"
        / "oss"
        / f"{language}-{component}-{suffix}.mdx"
    )


def _parse_frontmatter(text: str) -> dict[str, Any]:
    if not text.startswith("---"):
        return {}
    end = text.find("\n---", 3)
    if end == -1:
        return {}
    block = text[3:end].strip()
    data = yaml.safe_load(block)
    return data if isinstance(data, dict) else {}


def _parse_pepy_count(raw: str) -> int:
    latest = raw.replace(",", "").strip()
    if latest.endswith(("k", "K")):
        return int(float(latest[:-1]) * 1_000)
    if latest.endswith(("m", "M")):
        return int(float(latest[:-1]) * 1_000_000)
    return int(float(latest))


def _mark(value: Optional[bool]) -> str:
    # data-sort-value enables client-side column sorting (see
    # src/integration-downloads-table.js). Prefer true > false > unknown.
    if value is True:
        return '<span data-sort-value="2">✅</span>'
    if value is False:
        return '<span data-sort-value="1">❌</span>'
    return '<span data-sort-value="0"></span>'


def _pypi_url(package: str) -> str:
    return f"https://pypi.org/project/{package}/"


def _npm_url(package: str) -> str:
    return f"https://www.npmjs.com/package/{package}"


def _downloads_badge(row: IntegrationRow) -> str:
    # data-sort-value is the initial monthly count (-1 for N/A) for first paint
    # and offline fallback. The client re-fetches live pepy/shields badge SVGs
    # and rewrites these values so sort order matches the badges (see
    # src/integration-downloads-table.js).
    if not row.package or not row.registry or row.downloads is None:
        return '<span data-sort-value="-1">N/A</span>'
    sort_value = row.downloads
    if row.registry == "pypi":
        href = _pypi_url(row.package)
        img = f"https://static.pepy.tech/badge/{row.package}/month"
        badge = (
            f'<a href="{href}" target="_blank"><img src="{img}" '
            'alt="Downloads per month" noZoom class="rounded not-prose" /></a>'
        )
    else:
        href = _npm_url(row.package)
        encoded = quote(row.package, safe="@/")
        img = (
            f"https://img.shields.io/npm/dm/{encoded}"
            "?style=flat-square&label=%20&"
        )
        badge = (
            f'<a href="{href}" target="_blank"><img src="{img}" '
            'alt="Downloads per month" noZoom class="rounded not-prose" /></a>'
        )
    return f'<span data-sort-value="{sort_value}">{badge}</span>'


def fetch_npm_downloads(package: str) -> int:
    encoded = quote(package, safe="@/")
    url = f"https://api.npmjs.org/downloads/point/last-month/{encoded}"
    last_error: Exception | None = None
    for attempt in range(6):
        response = requests.get(url, headers=_HTTP_HEADERS, timeout=_REQUEST_TIMEOUT)
        if response.status_code == 429:
            sleep_for = min(2 ** attempt, 30)
            print(
                f"rate limited for npm:{package}; retrying in {sleep_for}s",
                file=sys.stderr,
            )
            time.sleep(sleep_for)
            last_error = requests.HTTPError(
                f"429 for {package}", response=response
            )
            continue
        response.raise_for_status()
        return int(response.json()["downloads"])
    assert last_error is not None
    raise last_error


def fetch_pypi_downloads(package: str) -> int:
    url = f"https://pepy.tech/badge/{package}/month"
    last_error: Exception | None = None
    for attempt in range(6):
        response = requests.get(url, headers=_HTTP_HEADERS, timeout=_REQUEST_TIMEOUT)
        if response.status_code == 429:
            sleep_for = min(2 ** attempt, 30)
            print(
                f"rate limited for pypi:{package}; retrying in {sleep_for}s",
                file=sys.stderr,
            )
            time.sleep(sleep_for)
            last_error = requests.HTTPError(
                f"429 for {package}", response=response
            )
            continue
        response.raise_for_status()
        texts = re.findall(r"<text[^>]*>([^<]+)</text>", response.text)
        if not texts:
            raise ValueError(f"No download text found in pepy badge for {package}")
        return _parse_pepy_count(texts[-1])
    assert last_error is not None
    raise last_error


def _as_bool(value: Any) -> Optional[bool]:
    if isinstance(value, bool):
        return value
    return None


def _page_files(language: str, component: str) -> list[Path]:
    root = _integrations_dir(language) / component
    if not root.is_dir():
        return []
    return sorted(
        p
        for p in root.rglob("*.mdx")
        if p.name not in SKIP_FILES
        and "TEMPLATE" not in p.name
        and "example_data" not in p.parts
    )


def _rel_integration_path(language: str, path: Path) -> str:
    return path.relative_to(_integrations_dir(language)).with_suffix("").as_posix()


def _resolve_downloads(
    language: str,
    npm: Any,
    pypi: Any,
    package_cache: dict[tuple[str, str], int],
) -> tuple[Optional[str], Optional[str], Optional[int]]:
    package: Optional[str] = None
    registry: Optional[str] = None
    downloads: Optional[int] = None

    if language == "javascript" and isinstance(npm, str) and npm.strip():
        candidate = npm.strip()
        if not candidate.startswith("-"):
            package = candidate
            registry = "npm"
    elif language == "python" and isinstance(pypi, str) and pypi.strip():
        candidate = pypi.strip()
        if not candidate.startswith("-"):
            package = candidate
            registry = "pypi"

    if not package or not registry:
        return package, registry, downloads

    cache_key = (registry, package)
    if cache_key not in package_cache:
        try:
            if registry == "npm":
                package_cache[cache_key] = fetch_npm_downloads(package)
            else:
                package_cache[cache_key] = fetch_pypi_downloads(package)
            print(f"{registry}:{package} -> {package_cache[cache_key]}")
            time.sleep(0.15)
        except (requests.RequestException, ValueError, KeyError) as exc:
            print(
                f"warn: failed to fetch {registry} downloads for {package}: {exc}",
                file=sys.stderr,
            )
            return None, None, None
    return package, registry, package_cache[cache_key]


def _is_safe_docs_url(url: str) -> bool:
    """Return True if url is safe to embed in a Markdown link href.

    Allows https://, http://, and site-relative paths that start with a
    single /. Rejects javascript:, data:, and protocol-relative //host URLs.
    """
    cleaned = url.strip()
    if not cleaned:
        return False
    # Site-relative only; reject protocol-relative URLs like //evil.example
    if cleaned.startswith("/") and not cleaned.startswith("//"):
        return True
    lower = cleaned.casefold()
    return lower.startswith("https://") or lower.startswith("http://")


def _normalize_docs_url(url: Any, *, label: str) -> Optional[str]:
    """Strip and validate docs_url; return None when missing or unsafe."""
    if not isinstance(url, str):
        return None
    cleaned = url.strip() or None
    if cleaned is None:
        return None
    if not _is_safe_docs_url(cleaned):
        print(
            f"warn: rejecting unsafe docs_url for {label}: {cleaned!r}",
            file=sys.stderr,
        )
        return None
    return cleaned


def validate_external_docs_urls(
    data: Optional[dict[str, Any]] = None,
) -> list[str]:
    """Return errors for missing or unsafe docs_url values in the YAML."""
    if data is None:
        data = _load_external_docs()
    errors: list[str] = []
    for language, components in data.items():
        if not isinstance(components, dict):
            continue
        for component, items in components.items():
            if not isinstance(items, list):
                continue
            for index, item in enumerate(items):
                if not isinstance(item, dict):
                    continue
                name = item.get("name")
                label = (
                    f"{language}/{component}/{name}"
                    if isinstance(name, str) and name.strip()
                    else f"{language}/{component}[{index}]"
                )
                docs_url = item.get("docs_url")
                if not isinstance(docs_url, str) or not docs_url.strip():
                    errors.append(f"{label}: missing docs_url")
                    continue
                if not _is_safe_docs_url(docs_url):
                    errors.append(
                        f"{label}: unsafe docs_url {docs_url.strip()!r} "
                        "(allowed: https://, http://, or site-relative /path)"
                    )
    return errors


def _row_from_integration_dict(
    *,
    rel_path: str,
    integration: dict[str, Any],
    language: str,
    package_cache: dict[tuple[str, str], int],
    docs_url: Optional[str] = None,
) -> Optional[IntegrationRow]:
    name = integration.get("name")
    if not name or not isinstance(name, str):
        return None

    package, registry, downloads = _resolve_downloads(
        language,
        integration.get("npm"),
        integration.get("pypi"),
        package_cache,
    )

    available = integration.get("available")
    source = integration.get("source")
    package_md = integration.get("package_md")
    external_docs = _normalize_docs_url(
        docs_url if docs_url is not None else integration.get("docs_url"),
        label=repr(name),
    )

    return IntegrationRow(
        rel_path=rel_path,
        name=name,
        package=package,
        registry=registry,
        downloads=downloads,
        featured=bool(integration.get("featured")),
        deprecated=bool(integration.get("deprecated")),
        stream=_as_bool(integration.get("stream")),
        tool_calling=_as_bool(integration.get("tool_calling")),
        structured_output=_as_bool(integration.get("structured_output")),
        multimodal=_as_bool(integration.get("multimodal")),
        docs_url=external_docs,
        available=available.strip()
        if isinstance(available, str) and available.strip()
        else None,
        source=source.strip()
        if isinstance(source, str) and source.strip()
        else None,
        self_host=_as_bool(integration.get("self_host")),
        cloud_offering=_as_bool(integration.get("cloud_offering")),
        package_md=package_md.strip()
        if isinstance(package_md, str) and package_md.strip()
        else None,
        delete_by_id=_as_bool(integration.get("delete_by_id")),
        filtering=_as_bool(integration.get("filtering")),
        search_by_vector=_as_bool(integration.get("search_by_vector")),
        search_with_score=_as_bool(integration.get("search_with_score")),
        async_api=_as_bool(integration.get("async_api")),
        passes_standard_tests=_as_bool(integration.get("passes_standard_tests")),
        multi_tenancy=_as_bool(integration.get("multi_tenancy")),
        ids_in_add_documents=_as_bool(integration.get("ids_in_add_documents")),
    )


def _collect_external_rows(
    language: str,
    component: str,
    package_cache: dict[tuple[str, str], int],
) -> list[IntegrationRow]:
    entries = _load_external_docs().get(language, {})
    if not isinstance(entries, dict):
        return []
    items = entries.get(component, [])
    if not isinstance(items, list):
        return []

    rows: list[IntegrationRow] = []
    for item in items:
        if not isinstance(item, dict):
            continue
        docs_url = item.get("docs_url")
        if not isinstance(docs_url, str) or not docs_url.strip():
            print(
                f"warn: external {language}/{component} entry missing docs_url",
                file=sys.stderr,
            )
            continue
        if not _is_safe_docs_url(docs_url):
            raise ValueError(
                f"unsafe docs_url in {_EXTERNAL_DOCS_PATH.name} "
                f"({language}/{component}): {docs_url.strip()!r}. "
                "Only https://, http://, or site-relative / paths are allowed."
            )
        row = _row_from_integration_dict(
            rel_path=f"{component}/external",
            integration=item,
            language=language,
            package_cache=package_cache,
            docs_url=docs_url.strip(),
        )
        if row is None:
            print(
                f"warn: external {language}/{component} entry missing name",
                file=sys.stderr,
            )
            continue
        rows.append(row)
    return rows


def _collect_rows(
    language: str,
    component: str,
    package_cache: dict[tuple[str, str], int],
) -> list[IntegrationRow]:
    rows: list[IntegrationRow] = []

    for path in _page_files(language, component):
        meta = _parse_frontmatter(path.read_text(encoding="utf-8"))
        integration = meta.get("integration")
        if not isinstance(integration, dict):
            print(
                f"warn: {path.relative_to(_REPO_ROOT)} missing integration frontmatter",
                file=sys.stderr,
            )
            continue

        row = _row_from_integration_dict(
            rel_path=_rel_integration_path(language, path),
            integration=integration,
            language=language,
            package_cache=package_cache,
        )
        if row is None:
            print(
                f"warn: {path.relative_to(_REPO_ROOT)} missing integration.name",
                file=sys.stderr,
            )
            continue
        rows.append(row)

    rows.extend(_collect_external_rows(language, component, package_cache))

    rows.sort(
        key=lambda row: (
            row.downloads is None,
            -(row.downloads or 0),
            row.name.lower(),
        )
    )
    return rows


def _model_link(row: IntegrationRow) -> str:
    if row.docs_url and _is_safe_docs_url(row.docs_url):
        link = f"[`{row.name}`]({row.docs_url})"
    else:
        link = f"[`{row.name}`](/oss/integrations/{row.rel_path})"
    if row.deprecated:
        return f"{link} (deprecated)"
    return link


def _vectorstore_caps_in_use(
    rows: list[IntegrationRow],
) -> tuple[tuple[str, str], ...]:
    """Return capability columns that have at least one known value.

    Omits empty feature columns when no page documents that capability
    (common for TypeScript until frontmatter is filled in).
    """
    return tuple(
        (key, title)
        for key, title in VECTORSTORE_CAPABILITY_KEYS
        if any(getattr(row, key) is not None for row in rows)
    )


def _table_header(
    component: str,
    *,
    vectorstore_caps: tuple[tuple[str, str], ...] = (),
) -> tuple[str, str]:
    label = COMPONENTS[component]["link_label"]
    if COMPONENTS[component]["capabilities"]:
        if component != "chat":
            raise ValueError(f"capabilities only supported for chat, got {component}")
        # Prefer JS-style labels for javascript snippets is handled by caller via language;
        # use neutral chat headers matching javascript (links work for both).
        header = (
            f"| {label} | Stream | [Tool Calling](/oss/langchain/tools/) "
            "| [`withStructuredOutput()`](/oss/langchain/models#structured-output) "
            "| [`Multimodal`](/oss/langchain/messages#multimodal) | Downloads |"
        )
        sep = "| :--- | :--- | :--- | :--- | :--- | :--- |"
        return header, sep
    if COMPONENTS[component].get("middleware_columns"):
        header = f"| {label} | Middleware available | Source | Downloads |"
        sep = "| :--- | :--- | :--- | :--- |"
        return header, sep
    if COMPONENTS[component].get("retriever_columns"):
        header = f"| {label} | Self-host | Cloud offering | Package | Downloads |"
        sep = "| :--- | :--- | :--- | :--- | :--- |"
        return header, sep
    if COMPONENTS[component].get("vectorstore_columns") and vectorstore_caps:
        caps = " | ".join(title for _, title in vectorstore_caps)
        header = f"| {label} | {caps} | Downloads |"
        sep = (
            "| :--- | "
            + " | ".join(":---" for _ in vectorstore_caps)
            + " | :--- |"
        )
        return header, sep
    header = f"| {label} | Downloads |"
    sep = "| :--- | :--- |"
    return header, sep


def _chat_table_header(language: str) -> tuple[str, str]:
    if language == "javascript":
        header = (
            "| Model | Stream | [Tool Calling](/oss/langchain/tools/) "
            "| [`withStructuredOutput()`](/oss/langchain/models#structured-output) "
            "| [`Multimodal`](/oss/langchain/messages#multimodal) | Downloads |"
        )
    else:
        header = (
            "| Model | Stream | [Tool calling](/oss/langchain/tools) "
            "| [Structured output](/oss/langchain/structured-output/) "
            "| [Multimodal](/oss/langchain/messages#multimodal) | Downloads |"
        )
    sep = "| :--- | :--- | :--- | :--- | :--- | :--- |"
    return header, sep


def _normalize_prose(text: str) -> str:
    """Normalize prose for Vale (no whitespace around em/en dashes)."""
    return re.sub(r"\s*([—–])\s*", r"\1", text)


def _escape_cell(text: str) -> str:
    return _normalize_prose(text).replace("|", "\\|")


def _package_cell(row: IntegrationRow) -> str:
    if row.package_md:
        return row.package_md
    if row.package and row.registry == "pypi":
        return f"[`{row.package}`](https://pypi.org/project/{row.package}/)"
    if row.package and row.registry == "npm":
        return f"[`{row.package}`](https://www.npmjs.com/package/{row.package})"
    return ""


def _render_row(
    component: str,
    row: IntegrationRow,
    *,
    vectorstore_caps: tuple[tuple[str, str], ...] = (),
) -> str:
    if COMPONENTS[component]["capabilities"]:
        cells = [
            _model_link(row),
            _mark(row.stream),
            _mark(row.tool_calling),
            _mark(row.structured_output),
            _mark(row.multimodal),
            _downloads_badge(row),
        ]
    elif COMPONENTS[component].get("middleware_columns"):
        cells = [
            _model_link(row),
            _escape_cell(row.available) if row.available else "",
            row.source or "",
            _downloads_badge(row),
        ]
    elif COMPONENTS[component].get("retriever_columns"):
        cells = [
            _model_link(row),
            _mark(row.self_host),
            _mark(row.cloud_offering),
            _package_cell(row),
            _downloads_badge(row),
        ]
    elif COMPONENTS[component].get("vectorstore_columns") and vectorstore_caps:
        cells = [_model_link(row)]
        for key, _ in vectorstore_caps:
            cells.append(_mark(getattr(row, key)))
        cells.append(_downloads_badge(row))
    else:
        cells = [_model_link(row), _downloads_badge(row)]
    return "| " + " | ".join(cells) + " |"


def _render_table(
    language: str, component: str, rows: list[IntegrationRow]
) -> str:
    vectorstore_caps: tuple[tuple[str, str], ...] = ()
    if COMPONENTS[component].get("vectorstore_columns"):
        vectorstore_caps = _vectorstore_caps_in_use(rows)
    if COMPONENTS[component]["capabilities"]:
        header, sep = _chat_table_header(language)
    else:
        header, sep = _table_header(component, vectorstore_caps=vectorstore_caps)
    # Wrapper class hooks client-side column sorting
    # (src/integration-downloads-table.js).
    lines = [
        HEADER.rstrip("\n"),
        "",
        '<div class="integration-downloads-table">',
        "",
        header,
        sep,
    ]
    for row in rows:
        lines.append(
            _render_row(component, row, vectorstore_caps=vectorstore_caps)
        )
    lines.extend(["", "</div>", ""])
    return "\n".join(lines)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--write",
        action="store_true",
        help="Write snippet files (default: print to stdout)",
    )
    parser.add_argument(
        "--check-docs-urls",
        action="store_true",
        help=(
            "Validate docs_url schemes in integration_external_docs.yaml "
            "and exit (no network, no writes)"
        ),
    )
    parser.add_argument(
        "--language",
        choices=[*LANGUAGES, "all"],
        default="all",
    )
    parser.add_argument(
        "--component",
        choices=[*COMPONENTS.keys(), "all"],
        default="all",
    )
    args = parser.parse_args()

    if args.check_docs_urls:
        errors = validate_external_docs_urls()
        if errors:
            for error in errors:
                print(f"error: {error}", file=sys.stderr)
            print(
                f"\n❌ {len(errors)} invalid docs_url value(s) in "
                f"{_EXTERNAL_DOCS_PATH.relative_to(_REPO_ROOT)}",
                file=sys.stderr,
            )
            return 1
        print(
            f"✅ All docs_url values in "
            f"{_EXTERNAL_DOCS_PATH.relative_to(_REPO_ROOT)} are safe"
        )
        return 0

    languages = list(LANGUAGES) if args.language == "all" else [args.language]
    components = (
        list(COMPONENTS.keys()) if args.component == "all" else [args.component]
    )

    package_cache: dict[tuple[str, str], int] = {}

    for language in languages:
        for component in components:
            if not (_integrations_dir(language) / component).is_dir():
                continue
            rows = _collect_rows(language, component, package_cache)
            if not rows:
                print(
                    f"skip: no rows for {language}/{component}",
                    file=sys.stderr,
                )
                continue

            featured = [row for row in rows if row.featured]
            tables: dict[str, str] = {"all": _render_table(language, component, rows)}
            # Always write featured for chat; for others only when any are featured
            if component == "chat" or featured:
                tables["featured"] = _render_table(language, component, featured)

            for kind, table in tables.items():
                if args.write:
                    path = _snippet_path(language, component, kind)
                    path.parent.mkdir(parents=True, exist_ok=True)
                    path.write_text(table, encoding="utf-8")
                    count = len(featured) if kind == "featured" else len(rows)
                    print(f"wrote {path.relative_to(_REPO_ROOT)} ({count} rows)")
                else:
                    print(f"======= {language} / {component} / {kind} =======")
                    print(table)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())