airflow/dev/registry/extract_metadata.py

#!/usr/bin/env python3
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
"""
Airflow Registry Metadata Extractor

Extracts provider metadata from:
1. provider.yaml files - Rich metadata (integrations, logos, categories)
2. pyproject.toml files - Dependencies, Python version constraints
3. PyPI API - Download statistics and release dates (optional, requires network)

Output: providers.json for the Astro static site generator

Module discovery (modules.json) is handled by extract_parameters.py, which uses
runtime inspection inside breeze for accurate class discovery.
"""

from __future__ import annotations

import concurrent.futures
import datetime
import json
import re
import shutil
import subprocess
import tomllib
import urllib.request
import zlib
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any

import yaml
from registry_contract_models import validate_providers_catalog
from registry_tools.uri_schemes import ReadModuleSource, collect_uri_schemes

# External endpoints used by metadata extraction.
PYPISTATS_RECENT_URL = "https://pypistats.org/api/packages/{package_name}/recent"
PYPI_PACKAGE_JSON_URL = "https://pypi.org/pypi/{package_name}/json"
# ClickHouse's public PyPI dataset (the data behind clickpy.clickhouse.com), sourced
# from the same PyPI download logs as pypistats.org but queryable for every package
# in a single SQL request -- no per-package rate limiting.
CLICKHOUSE_PYPI_URL = "https://sql-clickhouse.clickhouse.com/?user=demo"
S3_DOC_URL = "http://apache-airflow-docs.s3-website.eu-central-1.amazonaws.com"
AIRFLOW_PROVIDER_DOCS_URL = "https://airflow.apache.org/docs/{package_name}/stable/"
AIRFLOW_PROVIDER_SOURCE_URL = (
    "https://github.com/apache/airflow/tree/providers-{provider_id}/{version}/providers/{provider_path}"
)
PYPI_PACKAGE_URL = "https://pypi.org/project/{package_name}/"


def fetch_pypi_downloads(package_name: str) -> dict[str, int]:
    """Fetch download statistics from pypistats.org API."""
    try:
        url = PYPISTATS_RECENT_URL.format(package_name=package_name)
        with urllib.request.urlopen(url, timeout=5) as response:
            data = json.loads(response.read().decode())
            return {
                "weekly": data["data"].get("last_week", 0),
                "monthly": data["data"].get("last_month", 0),
                "total": 0,  # Total not available in recent endpoint
            }
    except Exception as e:
        print(f"    Warning: Could not fetch PyPI stats for {package_name}: {e}")
        return {"weekly": 0, "monthly": 0, "total": 0}


def fetch_pypi_dates(package_name: str) -> dict[str, str]:
    """Fetch first release and latest release dates from PyPI JSON API."""
    try:
        url = PYPI_PACKAGE_JSON_URL.format(package_name=package_name)
        with urllib.request.urlopen(url, timeout=10) as response:
            data = json.loads(response.read().decode())

        earliest = None
        latest = None
        for version_files in data.get("releases", {}).values():
            for file_info in version_files:
                upload = file_info.get("upload_time_iso_8601") or file_info.get("upload_time")
                if not upload:
                    continue
                ts = upload[:10]
                if earliest is None or ts < earliest:
                    earliest = ts
                if latest is None or ts > latest:
                    latest = ts

        return {
            "first_released": earliest or "",
            "last_updated": latest or "",
        }
    except Exception as e:
        print(f"    Warning: Could not fetch PyPI dates for {package_name}: {e}")
        return {"first_released": "", "last_updated": ""}


def fetch_pypi_downloads_clickhouse(package_names: list[str]) -> dict[str, dict[str, int]]:
    """Fetch weekly/monthly downloads for many packages in ONE query.

    Replaces the previous ~N parallel ``pypistats.org/api/recent`` calls. That burst
    tripped pypistats' per-IP rate limit, and ``fetch_pypi_downloads`` silently turned
    each 429 into ``{weekly: 0, monthly: 0}`` -- zeroing roughly a third of providers
    on the live registry. ClickHouse exposes the same underlying PyPI download data
    via a single SQL endpoint, so one request covers every package with nothing to
    rate-limit.

    Returns ``{package_name: {"weekly": int, "monthly": int, "total": 0}}``. On any
    failure returns ``{}`` so callers fall back to pypistats per-package. ``total`` is
    left at 0 to match the existing schema (the registry has never populated it).
    """
    if not package_names:
        return {}
    # Package names are ``apache-airflow-providers-*`` (only [a-z0-9-]); strip any
    # stray quote defensively before interpolating into the IN-list.
    in_list = ", ".join("'" + name.replace("'", "") + "'" for name in package_names)
    # Anchor the rolling windows on the dataset's latest loaded date, not today(): the
    # public dataset lags a few days, so a today()-relative window would be truncated to
    # the loaded days and undercount. max(date) gives a true last-7/30-day rolling sum,
    # matching pypistats' last_week/last_month semantics.
    query = (
        "WITH (SELECT max(date) FROM pypi.pypi_downloads_per_day) AS max_date "
        "SELECT project, "
        "toUInt64(sumIf(count, date > max_date - 7)) AS weekly, "
        "toUInt64(sumIf(count, date > max_date - 30)) AS monthly "
        "FROM pypi.pypi_downloads_per_day "
        f"WHERE project IN ({in_list}) "
        "GROUP BY project FORMAT TSV"
    )
    try:
        request = urllib.request.Request(CLICKHOUSE_PYPI_URL, data=query.encode())
        with urllib.request.urlopen(request, timeout=30) as response:
            body = response.read().decode()
    except Exception as e:
        print(f"    Warning: ClickHouse download query failed ({e}); falling back to pypistats")
        return {}
    result: dict[str, dict[str, int]] = {}
    for line in body.splitlines():
        parts = line.split("\t")
        if len(parts) != 3:
            continue
        project, weekly, monthly = parts
        try:
            result[project] = {"weekly": int(weekly), "monthly": int(monthly), "total": 0}
        except ValueError:
            continue
    return result


def fetch_pypi_data_parallel(
    package_names: list[str], max_workers: int = 16
) -> dict[str, tuple[dict[str, int], dict[str, str]]]:
    """Fetch downloads + release dates for many packages.

    Downloads come from a single ClickHouse query (``fetch_pypi_downloads_clickhouse``).
    Any package ClickHouse doesn't return -- or returns zero for, e.g. a just-published
    provider not yet in its dataset -- falls back to a per-package pypistats.org call;
    that path is now rare, so it no longer produces a rate-limiting burst. Release dates
    still come from pypi.org/json, fetched in parallel (a per-package-only endpoint that
    was never the source of the rate-limiting that motivated the ClickHouse switch).

    Returns ``{package_name: (downloads_dict, dates_dict)}``. Per-package failures are
    isolated -- the fallbacks return zero-value defaults -- so one flaky response only
    affects that package.
    """
    clickhouse_downloads = fetch_pypi_downloads_clickhouse(package_names)
    results: dict[str, tuple[dict[str, int], dict[str, str]]] = {}

    def _fetch_one(pkg: str) -> tuple[str, dict[str, int], dict[str, str]]:
        downloads = clickhouse_downloads.get(pkg)
        if not downloads or downloads.get("monthly", 0) == 0:
            # Missing from ClickHouse (or zero) -- fall back to pypistats for this one
            # package only. No burst because this is the exception, not every package.
            downloads = fetch_pypi_downloads(pkg)
        return pkg, downloads, fetch_pypi_dates(pkg)

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(_fetch_one, pkg) for pkg in package_names]
        for future in concurrent.futures.as_completed(futures):
            pkg, downloads, dates = future.result()
            results[pkg] = (downloads, dates)

    return results


def preserve_nonzero_downloads(new_providers: list[dict], existing_providers: list[dict]) -> int:
    """Keep a provider's previous download counts when this build fetched zero.

    Defense-in-depth for the case where BOTH ClickHouse and the pypistats fallback fail
    for a package: rather than overwrite a known-good number on the live registry with a
    spurious zero, retain the previous ``pypi_downloads`` from the existing catalog.
    Mutates ``new_providers`` in place; returns the number of providers preserved.
    """
    existing_by_id = {p["id"]: p for p in existing_providers}
    preserved = 0
    for provider in new_providers:
        if (provider.get("pypi_downloads") or {}).get("monthly", 0):
            continue  # this build got a real number; nothing to preserve
        previous = existing_by_id.get(provider["id"], {}).get("pypi_downloads") or {}
        if previous.get("monthly", 0) > 0:
            provider["pypi_downloads"] = previous
            preserved += 1
            print(f"    Preserved previous downloads for {provider['id']} (this build fetched 0)")
    return preserved


def _parse_inventory_lines(inv_path: Path) -> list[str]:
    """Read and decompress the body of a Sphinx objects.inv file."""
    with inv_path.open("rb") as f:
        for _ in range(4):
            f.readline()
        return zlib.decompress(f.read()).decode("utf-8").splitlines()


def read_inventory(inv_path: Path) -> dict[str, str]:
    """Parse a Sphinx objects.inv file and return {qualified_name: url_path} for py:class entries."""
    result: dict[str, str] = {}
    for line in _parse_inventory_lines(inv_path):
        parts = line.split(None, 4)
        if len(parts) != 5:
            continue
        name, domain_role, _prio, location, _dispname = parts
        if domain_role == "py:class":
            # "$" in location means "use the name as anchor"
            result[name] = location.replace("$", name)
    return result


def read_connection_urls(inv_path: Path) -> dict[str, str]:
    """Parse a Sphinx objects.inv and return {conn_type: relative_url} for connection pages.

    Uses two inventory entry types:
    - ``std:label howto/connection:{conn_type}`` — maps conn_type directly to a page
    - ``std:doc connections/{name}`` — fallback by matching conn_type to doc name
    """
    label_map: dict[str, str] = {}  # conn_type -> page URL (from std:label)
    doc_map: dict[str, str] = {}  # doc_name -> page URL (from std:doc)
    for line in _parse_inventory_lines(inv_path):
        parts = line.split(None, 4)
        if len(parts) != 5:
            continue
        name, domain_role, _prio, location, _dispname = parts
        if domain_role == "std:label" and name.startswith("howto/connection:"):
            label_key = name[len("howto/connection:") :]
            # Skip sub-section labels like "gcp:configuring_the_connection"
            if ":" not in label_key:
                label_map[label_key] = location.split("#")[0]
        elif domain_role == "std:doc" and name.startswith("connections/"):
            doc_name = name[len("connections/") :]
            if doc_name != "index":
                doc_map[doc_name] = location

    # Merge: label_map takes precedence, doc_map fills gaps
    result: dict[str, str] = {}
    result.update(label_map)
    for doc_name, url in doc_map.items():
        if doc_name not in result:
            result[doc_name] = url
    return result


INVENTORY_CACHE_DIR = Path(__file__).parent / ".inventory_cache"
INVENTORY_TTL = datetime.timedelta(hours=12)


def fetch_provider_inventory(package_name: str, cache_dir: Path = INVENTORY_CACHE_DIR) -> Path | None:
    """Download a provider's objects.inv from S3, caching locally with a 12-hour TTL.

    Returns the local cache path on success, or None if the fetch fails
    (e.g. provider not yet published).
    """
    cache_path = cache_dir / package_name / "objects.inv"
    if cache_path.exists():
        age = datetime.datetime.now(tz=datetime.UTC) - datetime.datetime.fromtimestamp(
            cache_path.stat().st_mtime, tz=datetime.UTC
        )
        if age < INVENTORY_TTL:
            return cache_path

    url = f"{S3_DOC_URL}/docs/{package_name}/stable/objects.inv"
    try:
        with urllib.request.urlopen(url, timeout=10) as response:
            content = response.read()
        # Validate it's a Sphinx inventory
        if not content.startswith(b"# Sphinx inventory version"):
            print(f"    Warning: Invalid inventory header for {package_name}")
            return None
        cache_path.parent.mkdir(parents=True, exist_ok=True)
        cache_path.write_bytes(content)
        return cache_path
    except Exception as e:
        print(f"    Warning: Could not fetch inventory for {package_name}: {e}")
        # On refetch failure, serve stale cache rather than nothing
        if cache_path.exists():
            print(f"    Using stale cache for {package_name}")
            return cache_path
        return None


def resolve_connection_docs_url(conn_type: str, conn_url_map: dict[str, str], base_docs_url: str) -> str:
    """Resolve the docs URL for a connection type using the inventory map.

    Lookup order:
    1. Exact match on conn_type in the inventory map
    2. Fallback to connections/ directory listing
    """
    if conn_type in conn_url_map:
        return f"{base_docs_url}/{conn_url_map[conn_type]}"
    return f"{base_docs_url}/connections/"


# Base paths
AIRFLOW_ROOT = Path(__file__).parent.parent.parent
SCRIPT_DIR = Path(__file__).parent
PROVIDERS_DIR = AIRFLOW_ROOT / "providers"
REGISTRY_DIR = AIRFLOW_ROOT / "registry"
OUTPUT_DIR = REGISTRY_DIR / "src" / "_data"


@dataclass
class Category:
    """Category within a provider."""

    id: str
    name: str
    module_count: int = 0


@dataclass
class Provider:
    """Provider metadata."""

    id: str
    name: str
    package_name: str
    description: str
    lifecycle: str = "production"  # AIP-95: incubation, production, mature, deprecated
    logo: str | None = None
    version: str = ""
    versions: list[str] = field(default_factory=list)
    airflow_versions: list[str] = field(default_factory=list)
    pypi_downloads: dict[str, int] = field(default_factory=lambda: {"weekly": 0, "monthly": 0, "total": 0})
    module_counts: dict[str, int] = field(
        default_factory=lambda: {
            "operator": 0,
            "hook": 0,
            "sensor": 0,
            "trigger": 0,
            "transfer": 0,
            "notifier": 0,
            "secret": 0,
            "logging": 0,
            "executor": 0,
            "bundle": 0,
            "decorator": 0,
        }
    )
    categories: list[dict] = field(default_factory=list)
    connection_types: list[dict] = field(
        default_factory=list
    )  # {conn_type, hook_class, docs_url, external_services}
    uri_schemes: list[dict] = field(default_factory=list)  # {scheme, filesystem?, asset?, remote_logging?}
    requires_python: str = ""  # e.g., ">=3.11"
    dependencies: list[str] = field(default_factory=list)  # from pyproject.toml
    optional_extras: dict[str, list[str]] = field(default_factory=dict)  # {extra_name: [deps]}
    dependents: list[str] = field(default_factory=list)
    related_providers: list[str] = field(default_factory=list)
    docs_url: str = ""
    source_url: str = ""
    pypi_url: str = ""
    first_released: str = ""
    last_updated: str = ""


def parse_provider_yaml(yaml_path: Path) -> dict[str, Any]:
    """Parse a provider.yaml file."""
    with open(yaml_path) as f:
        return yaml.safe_load(f)


def parse_pyproject_toml(pyproject_path: Path) -> dict[str, Any]:
    """Parse pyproject.toml and extract requires-python, dependencies, and optional extras."""
    result: dict[str, Any] = {"requires_python": "", "dependencies": [], "optional_extras": {}}

    if not pyproject_path.exists():
        return result

    try:
        with open(pyproject_path, "rb") as f:
            data = tomllib.load(f)

        project = data.get("project", {})

        result["requires_python"] = project.get("requires-python", "")
        result["dependencies"] = [d.strip() for d in project.get("dependencies", [])][:20]

        optional_deps = project.get("optional-dependencies", {})
        for extra_name, extra_deps in optional_deps.items():
            clean = [d.strip() for d in extra_deps if d.strip()]
            if clean:
                result["optional_extras"][extra_name] = clean[:5]

    except Exception as e:
        print(f"    Warning: Could not parse {pyproject_path}: {e}")

    return result


def extract_integrations_as_categories(provider_yaml: dict[str, Any]) -> list[Category]:
    """Extract integrations from provider.yaml as categories."""
    categories: dict[str, Category] = {}

    for integration in provider_yaml.get("integrations", []):
        name = integration.get("integration-name", "")
        if not name:
            continue

        # Create a slug for the category ID
        cat_id = name.lower().replace(" ", "-").replace("(", "").replace(")", "")
        cat_id = re.sub(r"[^a-z0-9-]", "", cat_id)

        if cat_id not in categories:
            categories[cat_id] = Category(id=cat_id, name=name, module_count=0)

    return list(categories.values())


def module_path_to_file_path(module_path: str, provider_path: Path) -> Path:
    """Convert a Python module path to an actual file path.

    provider_path is the actual filesystem directory (e.g., providers/microsoft/azure/),
    not the dash-joined provider_id.
    """
    # e.g., airflow.providers.amazon.operators.s3 -> providers/amazon/src/airflow/providers/amazon/operators/s3.py
    parts = module_path.split(".")
    file_path = provider_path / "src" / "/".join(parts)
    return file_path.with_suffix(".py")


def module_source_reader(provider_path: Path) -> ReadModuleSource:
    """Return a reader for module sources in the provider's working-tree ``src/`` directory."""

    def read(module_path: str) -> str | None:
        try:
            return module_path_to_file_path(module_path, provider_path).read_text()
        except FileNotFoundError:
            return None

    return read


def determine_airflow_versions(dependencies: list[str]) -> list[str]:
    """Determine minimum Airflow version from pyproject.toml dependencies."""
    for dep in dependencies:
        if dep.startswith("apache-airflow>="):
            version_str = dep.split(">=")[1].split(",")[0].strip()
            parts = version_str.split(".")
            if len(parts) >= 2:
                return [f"{parts[0]}.{parts[1]}+"]
    return ["3.0+"]


def find_related_providers(provider_id: str, all_provider_yamls: dict[str, dict]) -> list[str]:
    """Find related providers based on shared integrations or dependencies."""
    current = all_provider_yamls.get(provider_id, {})
    current_integrations = {i.get("integration-name") for i in current.get("integrations", [])}

    related = []
    for other_id, other_yaml in all_provider_yamls.items():
        if other_id == provider_id:
            continue
        other_integrations = {i.get("integration-name") for i in other_yaml.get("integrations", [])}
        overlap = current_integrations & other_integrations
        if overlap:
            related.append(other_id)

    return related[:5]  # Limit to 5 related providers


def load_release_tags() -> set[str]:
    """Return all ``providers-<id>/<version>`` git tags as a set for fast lookup.

    Used to filter ``provider.yaml`` ``versions:`` lists to only entries that
    correspond to a real release (excludes phantom version bumps where the
    next-version entry was prepended to ``versions:`` before the tag landed,
    or pre-release-only versions like ``providers-celery/3.19.0rc1`` where the
    ``rc1`` exists but the final does not).

    Returns an empty set if the ``git`` command fails (e.g., outside a checkout);
    callers can decide whether to fall back to the unfiltered top entry.
    """
    try:
        result = subprocess.run(
            ["git", "tag", "--list", "providers-*"],
            capture_output=True,
            text=True,
            cwd=AIRFLOW_ROOT,
            check=True,
        )
    except (subprocess.CalledProcessError, FileNotFoundError):
        return set()
    return {line.strip() for line in result.stdout.splitlines() if line.strip()}


def find_latest_released_version(
    provider_id: str,
    versions_list: list[str],
    release_tags: set[str],
) -> str | None:
    """Walk ``versions_list`` newest-first, return the first version with a real release tag.

    Returns ``None`` when no entry in ``versions_list`` has a corresponding
    ``providers-<id>/<version>`` tag, indicating the provider is unreleased
    (brand-new in-tree, no tags yet) or in an inconsistent state.
    """
    for version in versions_list:
        if f"providers-{provider_id}/{version}" in release_tags:
            return version
    return None


def main():
    """Main extraction function."""
    import argparse

    parser = argparse.ArgumentParser(description="Airflow Registry Metadata Extractor")
    parser.add_argument(
        "--provider",
        default=None,
        help="Extract only this provider ID (e.g. 'amazon'). Omit for full build.",
    )
    parser.add_argument(
        "--allow-unreleased",
        action="store_true",
        help=(
            "Include providers and versions that don't have a matching "
            "providers-<id>/<ver> git tag. Use for staging builds and local dev "
            "where maintainers want to preview unreleased provider pages before "
            "the tag lands. Default is to filter unreleased entries so live "
            "builds don't ship phantom pointers."
        ),
    )
    args = parser.parse_args()

    print("Airflow Registry Metadata Extractor")
    print("=" * 50)
    if args.provider:
        requested_providers = set(args.provider.split())
        print(f"Incremental mode: extracting provider(s) {requested_providers}")
    else:
        requested_providers = None
    if args.allow_unreleased:
        print("Unreleased providers: INCLUDED (--allow-unreleased)")

    # Ensure output directory exists
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

    all_providers: list[Provider] = []
    all_provider_yamls: dict[str, dict] = {}

    # First pass: Load all provider.yaml files (including nested ones like dbt/cloud, microsoft/azure)
    provider_yaml_paths: dict[str, Path] = {}
    for yaml_path in PROVIDERS_DIR.rglob("provider.yaml"):
        # Calculate provider_id from path relative to PROVIDERS_DIR
        # e.g., providers/amazon/provider.yaml -> amazon
        # e.g., providers/microsoft/azure/provider.yaml -> microsoft-azure
        relative_path = yaml_path.relative_to(PROVIDERS_DIR)
        parts = relative_path.parts[:-1]  # Remove 'provider.yaml'
        provider_id = "-".join(parts)

        try:
            provider_yaml = parse_provider_yaml(yaml_path)
            all_provider_yamls[provider_id] = provider_yaml
            provider_yaml_paths[provider_id] = yaml_path.parent
        except Exception as e:
            print(f"  Error parsing {yaml_path}: {e}")

    print(f"Found {len(all_provider_yamls)} providers with provider.yaml")

    # Filter to requested providers if --provider was given
    if requested_providers:
        filtered_ids = {pid for pid in all_provider_yamls if pid in requested_providers}
        missing = requested_providers - filtered_ids
        if missing:
            print(f"  Warning: provider(s) not found: {missing}")
        # Keep only the requested providers for extraction
        extraction_ids = filtered_ids
    else:
        extraction_ids = set(all_provider_yamls.keys())

    # Load all release tags once. Used below to filter `provider.yaml`'s
    # `versions:` to only entries that have a real `providers-<id>/<ver>`
    # git tag, avoiding phantom-version leaks (next-release bumps prepended
    # to `versions:` before the tag lands, RC-only releases, brand-new
    # providers with no tags yet).
    #
    # Skipped entirely when --allow-unreleased is set: staging builds and
    # local dev want to preview unreleased provider pages so maintainers
    # can verify them before tagging.
    skipped_unreleased: list[str] = []
    if args.allow_unreleased:
        release_tags: set[str] = set()
    else:
        release_tags = load_release_tags()
        if not release_tags:
            # registry-build.yml refuses a live build without tags; locally this is a
            # clone without the providers-* tags, such as a fork.
            print(
                "  Warning: no providers-* git tags found; "
                "phantom version filter is disabled (falling back to versions[0])."
            )
        # A not-ready provider is not released yet (breeze release tooling skips it too).
        not_ready = {pid for pid in extraction_ids if all_provider_yamls[pid].get("state") == "not-ready"}
        for provider_id in sorted(not_ready):
            print(f"  Skipping {provider_id}: state is not-ready")
        skipped_unreleased.extend(not_ready)
        extraction_ids -= not_ready

    # Pre-fetch PyPI download stats and release dates for every provider in
    # parallel. This used to be ~2N sequential urlopen calls inside the loop
    # below (5-10s timeouts each), which dominated wall-clock and serialised
    # any pypistats slowdowns into a multi-minute build delay. Doing it once
    # up-front turns ~86 sequential calls into ~6 batches of 16.
    pypi_package_names = sorted(
        all_provider_yamls[pid].get("package-name", f"apache-airflow-providers-{pid}")
        for pid in extraction_ids
    )
    print(f"Fetching PyPI metadata for {len(pypi_package_names)} packages (parallel)...")
    pypi_data = fetch_pypi_data_parallel(pypi_package_names)

    # Second pass: Extract full metadata (only for providers in extraction_ids)
    for provider_id in extraction_ids:
        provider_yaml = all_provider_yamls[provider_id]
        provider_path = provider_yaml_paths[provider_id]

        package_name = provider_yaml.get("package-name", f"apache-airflow-providers-{provider_id}")
        name = provider_yaml.get("name", provider_id.replace("-", " ").title())
        description = provider_yaml.get("description", "")

        # Clean up RST formatting in description
        # Convert RST links like `Text <url>`__ to just "Text"
        description = re.sub(r"`([^<]+)\s*<[^>]+>`__", r"\1", description)
        # Remove any remaining backticks
        description = re.sub(r"`", "", description)
        # Remove bullet points (- at start of lines)
        description = re.sub(r"\n\s*-\s*", ", ", description)
        # Clean up extra whitespace and normalize
        description = re.sub(r"\s+", " ", description).strip()
        # Remove trailing commas
        description = re.sub(r",\s*$", "", description)
        # Fix extra spaces before punctuation
        description = re.sub(r"\s+([),.])", r"\1", description)
        # Remove "including:" followed by just commas
        description = re.sub(r"including:\s*,", "including", description)
        # Fix "(including X )" -> "(including X)"
        description = re.sub(r"\s+\)", ")", description)
        # Remove empty parentheses
        description = re.sub(r"\(\s*\)", "", description)
        # Truncate long descriptions
        if len(description) > 200:
            description = description[:197] + "..."

        # Get versions, filtering to entries that have a real release tag.
        # Provider release prep prepends the next version to `versions:` BEFORE
        # the tag lands, and pre-release-only versions match `versions:` but
        # have no final tag. Without filtering, `version` (the latest pointer)
        # AND the `versions` list both leak phantoms downstream -- the latter
        # is consumed by extract_versions.py's backfill, which would try to
        # `git show` from a non-existent tag.
        raw_versions = provider_yaml.get("versions", [])
        if release_tags:
            versions = [v for v in raw_versions if f"providers-{provider_id}/{v}" in release_tags]
            version = find_latest_released_version(provider_id, raw_versions, release_tags)
            if version is None:
                skipped_unreleased.append(provider_id)
                print(
                    f"  Skipping {provider_id}: no released version found in "
                    f"versions list {raw_versions} "
                    f"(no matching providers-{provider_id}/<ver> tag)"
                )
                continue
        else:
            # --allow-unreleased, or no tags found: keep every version listed in provider.yaml.
            versions = list(raw_versions)
            version = versions[0] if versions else "0.0.0"

        # Extract categories from integrations
        categories = extract_integrations_as_categories(provider_yaml)

        # Find logo from provider's docs/integration-logos/ directory and copy
        # to dev/registry/logos/ (mounted in breeze).  The CI workflow copies
        # these into registry/public/logos/ before the 11ty build.
        logo = None
        logo_path = None

        # Try to find the most representative logo for the provider
        # Priority: Look for "main" provider logos that match the provider name
        integration_logos_dir = provider_path / "docs" / "integration-logos"

        # Common patterns for main provider logos
        main_logo_patterns = [
            f"{name.replace(' ', '-')}.png",  # e.g., "Google-Cloud.png"
            f"{name.replace(' ', '-')}.svg",
            f"{name}.png",
            f"{provider_id.replace('-', '_')}.png",
        ]

        # Special case mappings for well-known providers
        logo_priority_map = {
            "amazon": ["AWS-Cloud-alt_light-bg@4x.png", "Amazon-Web-Services.png"],
            "google": ["Google-Cloud.png", "Google.png"],
            "microsoft-azure": ["Microsoft-Azure.png"],
            "snowflake": ["Snowflake.png"],
            "databricks": ["Databricks.png"],
        }

        # Write logos to dev/registry/logos/ — this directory is mounted in
        # breeze (unlike registry/public/) so copies survive the container.
        # Also copy to registry/public/logos/ for local dev convenience.
        # Directories are created lazily (only when a logo is found) to avoid
        # empty dirs that trip up glob-based cp in the CI workflow.
        logos_dest_dir = SCRIPT_DIR / "logos"
        registry_logos_dir = SCRIPT_DIR.parent.parent / "registry" / "public" / "logos"

        if integration_logos_dir.exists():
            logos_dest_dir.mkdir(parents=True, exist_ok=True)

            # First, check for priority logos for known providers
            if provider_id in logo_priority_map:
                for priority_logo in logo_priority_map[provider_id]:
                    potential_logo = integration_logos_dir / priority_logo
                    if potential_logo.exists():
                        logo_dest = logos_dest_dir / f"{provider_id}-{potential_logo.name}"
                        shutil.copy2(potential_logo, logo_dest)
                        logo = f"/logos/{provider_id}-{potential_logo.name}"
                        break

            # If no priority logo found, try general patterns
            if not logo:
                for pattern in main_logo_patterns:
                    potential_logo = integration_logos_dir / pattern
                    if potential_logo.exists():
                        logo_dest = logos_dest_dir / f"{provider_id}-{potential_logo.name}"
                        shutil.copy2(potential_logo, logo_dest)
                        logo = f"/logos/{provider_id}-{potential_logo.name}"
                        break

            # Still no logo? Use the one from provider.yaml integrations
            if not logo:
                for integration in provider_yaml.get("integrations", []):
                    if integration.get("logo"):
                        logo_path = integration["logo"]
                        logo_filename = logo_path.split("/")[-1]
                        logo_source = integration_logos_dir / logo_filename
                        if logo_source.exists():
                            logo_dest = logos_dest_dir / f"{provider_id}-{logo_filename}"
                            shutil.copy2(logo_source, logo_dest)
                            logo = f"/logos/{provider_id}-{logo_filename}"
                            break

            # Last resort: use the first available logo
            if not logo:
                logos = list(integration_logos_dir.glob("*.png")) + list(integration_logos_dir.glob("*.svg"))
                if logos:
                    logo_source = logos[0]
                    logo_dest = logos_dest_dir / f"{provider_id}-{logo_source.name}"
                    shutil.copy2(logo_source, logo_dest)
                    logo = f"/logos/{provider_id}-{logo_source.name}"

        # Also copy to registry/public/logos/ so local `pnpm dev` works without
        # the extra CI copy step.
        if logo:
            logo_filename = logo.split("/")[-1]
            src = logos_dest_dir / logo_filename
            if src.exists():
                registry_logos_dir.mkdir(parents=True, exist_ok=True)
                shutil.copy2(src, registry_logos_dir / logo_filename)

        # Extract connection types from provider.yaml
        # Resolve per-connection docs URLs from Sphinx inventory when available
        connection_types = []
        base_docs_url = AIRFLOW_PROVIDER_DOCS_URL.format(package_name=package_name).rstrip("/")
        conn_url_map: dict[str, str] = {}
        inv_path = fetch_provider_inventory(package_name)
        if inv_path:
            conn_url_map = read_connection_urls(inv_path)
        for conn in provider_yaml.get("connection-types", []):
            conn_type = conn.get("connection-type", "")
            hook_class = conn.get("hook-class-name", "")
            external_services = conn.get("external-services", [])
            if conn_type:
                connection_types.append(
                    {
                        "conn_type": conn_type,
                        "hook_class": hook_class,
                        "docs_url": resolve_connection_docs_url(conn_type, conn_url_map, base_docs_url),
                        "external_services": external_services,
                    }
                )

        # Pre-fetched in parallel before this loop; missing entries fall back
        # to zero-value defaults so a never-published provider doesn't crash.
        pypi_downloads, pypi_dates = pypi_data.get(
            package_name,
            ({"weekly": 0, "monthly": 0, "total": 0}, {"first_released": "", "last_updated": ""}),
        )

        # Parse pyproject.toml for requires-python and dependencies
        pyproject_path = provider_path / "pyproject.toml"
        pyproject_data = parse_pyproject_toml(pyproject_path)

        # Airflow version compatibility (from pyproject.toml dependencies)
        airflow_versions = determine_airflow_versions(pyproject_data["dependencies"])

        provider_source_path = provider_path.relative_to(PROVIDERS_DIR).as_posix()
        provider = Provider(
            id=provider_id,
            name=name,
            package_name=package_name,
            description=description,
            lifecycle=provider_yaml.get("lifecycle", "production"),
            logo=logo,
            version=version,
            versions=versions,
            airflow_versions=airflow_versions,
            pypi_downloads=pypi_downloads,
            categories=[asdict(c) for c in categories],
            connection_types=connection_types,
            uri_schemes=collect_uri_schemes(provider_yaml, module_source_reader(provider_path)),
            requires_python=pyproject_data["requires_python"],
            dependencies=pyproject_data["dependencies"],
            optional_extras=pyproject_data.get("optional_extras", {}),
            docs_url=AIRFLOW_PROVIDER_DOCS_URL.format(package_name=package_name),
            source_url=AIRFLOW_PROVIDER_SOURCE_URL.format(
                provider_id=provider_id, version=version, provider_path=provider_source_path
            ),
            pypi_url=PYPI_PACKAGE_URL.format(package_name=package_name),
            first_released=pypi_dates["first_released"],
            last_updated=pypi_dates["last_updated"],
        )

        all_providers.append(provider)
        print(f"  {provider_id}: {len(categories)} categories")

    if skipped_unreleased:
        print(
            f"\nSkipped {len(skipped_unreleased)} unreleased provider(s) "
            f"(not-ready, or no matching git tag): {sorted(skipped_unreleased)}"
        )

    # Find related providers
    for provider in all_providers:
        provider.related_providers = find_related_providers(provider.id, all_provider_yamls)

    # Convert to JSON-serializable format
    new_providers = [asdict(p) for p in all_providers]

    # Defense-in-depth: if this build fetched 0 downloads for a provider that had a real
    # number on the previous (S3) providers.json, keep the previous number rather than
    # publishing a spurious zero (#1309). No-op when there's no existing catalog on disk.
    for candidate_dir in (SCRIPT_DIR, OUTPUT_DIR):
        existing_catalog_path = candidate_dir / "providers.json"
        if existing_catalog_path.exists():
            try:
                previous_catalog = json.loads(existing_catalog_path.read_text())
            except json.JSONDecodeError:
                break
            preserved = preserve_nonzero_downloads(new_providers, previous_catalog.get("providers", []))
            if preserved:
                print(f"Preserved previous download counts for {preserved} provider(s) that fetched 0")
            break

    # In incremental mode, merge new providers into existing providers.json
    # so parallel runs for different providers don't clobber each other.
    if requested_providers:
        new_by_id = {p["id"]: p for p in new_providers}
        for out_dir in [SCRIPT_DIR, OUTPUT_DIR]:
            existing_path = out_dir / "providers.json"
            if existing_path.exists():
                try:
                    existing = json.loads(existing_path.read_text())
                    merged = [new_by_id.pop(p["id"], p) for p in existing["providers"]]
                    merged.extend(new_by_id.values())
                    new_providers = merged
                    print(
                        f"Merged {len(all_providers)} updated + {len(merged) - len(all_providers)} existing providers"
                    )
                except (json.JSONDecodeError, KeyError):
                    pass
                break

    new_providers.sort(key=lambda p: p["name"].lower())
    providers_json = validate_providers_catalog({"providers": new_providers})

    # Write output files to all output directories.
    # The SCRIPT_DIR (dev/registry/) copy is the first place the other extraction
    # scripts, which run inside breeze, look for providers.json.
    output_dirs = [OUTPUT_DIR, SCRIPT_DIR]

    for out_dir in output_dirs:
        if not out_dir.parent.exists():
            continue
        out_dir.mkdir(parents=True, exist_ok=True)
        with open(out_dir / "providers.json", "w") as f:
            json.dump(providers_json, f, indent=2)
        print(f"\nWrote {len(new_providers)} providers to {out_dir}")

    print("\nDone!")


if __name__ == "__main__":
    main()