Conda Forge Dependency Forensics

get_artifact_host_dep_versions.py

#!/usr/bin/env python

## Written w/ Perplexity

import csv
import os
import shutil
import sys
import tempfile
from pathlib import Path

import requests
import yaml
# https://conda.github.io/conda-package-streaming/url.html
from conda_package_streaming.url import extract_conda_info


BASE_URL = "https://conda.anaconda.org/conda-forge"

SUBDIRS = [
    "linux-64",
    "linux-aarch64",
    "linux-ppc64le",
    "osx-64",
    "osx-arm64",
    "win-32",
    "win-64",
    "noarch",
]


def fetch_repodata(subdir):
    url = f"{BASE_URL}/{subdir}/repodata.json"
    r = requests.get(url, timeout=60)
    r.raise_for_status()
    return r.json()


def iter_target_packages(repodata, package_name):
    for section in ("packages", "packages.conda"):
        for filename, meta in repodata.get(section, {}).items():
            if meta.get("name") == package_name:
                yield filename, meta


def stream_meta_yaml_via_url(subdir, filename, workdir, session):
    """
    Use conda-package-streaming.url.extract_conda_info to fetch only
    info/index.json and info/recipe/meta.yaml from the remote package,
    then read meta.yaml from disk. [page:0]
    """
    url = f"{BASE_URL}/{subdir}/{filename}"
    destdir = Path(workdir) / "info"
    if destdir.exists():
        shutil.rmtree(destdir)
    destdir.mkdir(parents=True, exist_ok=True)

    # checklist defaults to {'info/index.json', 'info/recipe/meta.yaml'}, so
    # this stops as soon as those files are extracted.
    extract_conda_info(url, destdir=str(destdir), session=session)

    meta_yaml = destdir / "info" / "recipe" / "meta.yaml"
    if not meta_yaml.exists():
        return None
    return meta_yaml.read_text(encoding="utf-8")


def parse_dependency_from_host(meta_text, dependency_name):
    """
    Parse requirements.host from evaluated meta.yaml.
    Entries of the form:
      - imagemagick 7.1.2_15 imagemagick_hf4cd0b3_0
    are split into name, version, build.
    """
    data = yaml.safe_load(meta_text)
    if not isinstance(data, dict):
        return {
            "dependency_version": None,
            "dependency_build": None,
        }

    requirements = data.get("requirements", {})
    host = requirements.get("host", []) or []

    for entry in host:
        if isinstance(entry, str):
            parts = entry.split()
            if not parts:
                continue
            name = parts[0]
            if name != dependency_name:
                continue

            if len(parts) == 2:
                version = parts[1]
                build = None
            elif len(parts) >= 3:
                version = parts[1]
                build = parts[2]
            else:
                version = None
                build = None

            return {
                "dependency_version": version,
                "dependency_build": build,
            }

        elif isinstance(entry, dict):
            name = entry.get("name")
            if name != dependency_name:
                continue
            version = entry.get("version")
            build = entry.get("build")
            return {
                "dependency_version": version,
                "dependency_build": build,
            }

    return {
        "dependency_version": None,
        "dependency_build": None,
    }


def main():
    if len(sys.argv) != 3:
        print(
            f"Usage: {Path(sys.argv[0]).name} <package-name> <dependency-name>",
            file=sys.stderr,
        )
        sys.exit(1)

    package_name = sys.argv[1]
    dependency_name = sys.argv[2]
    output_csv = f"{package_name}__{dependency_name}__host_requirements.csv"

    rows = []

    # Reuse a single requests.Session for all URL operations.
    session = requests.Session()

    with tempfile.TemporaryDirectory() as tmpdir:
        tmpdir = Path(tmpdir)

        for subdir in SUBDIRS:
            try:
                repodata = fetch_repodata(subdir)
            except Exception as e:
                print(f"[WARN] Failed to fetch repodata for {subdir}: {e}", file=sys.stderr)
                continue

            for filename, meta in iter_target_packages(repodata, package_name):
                artifact = f"{subdir}/{filename}"
                print(f"[INFO] Processing {artifact}", file=sys.stderr)

                # Fresh work dir per artifact
                workdir = tmpdir / "pkginfo"
                if workdir.exists():
                    shutil.rmtree(workdir)
                workdir.mkdir(parents=True, exist_ok=True)

                try:
                    meta_text = stream_meta_yaml_via_url(subdir, filename, workdir, session)
                except Exception as e:
                    print(f"[WARN] Failed to fetch info for {artifact}: {e}", file=sys.stderr)
                    rows.append({
                        "artifact": artifact,
                        "dependency_name": dependency_name,
                        "dependency_version": None,
                        "dependency_build": None,
                    })
                    continue

                if meta_text is None:
                    rows.append({
                        "artifact": artifact,
                        "dependency_name": dependency_name,
                        "dependency_version": None,
                        "dependency_build": None,
                    })
                    continue

                try:
                    dep_info = parse_dependency_from_host(meta_text, dependency_name)
                except Exception as e:
                    print(f"[WARN] Failed to parse meta.yaml for {artifact}: {e}", file=sys.stderr)
                    dep_info = {
                        "dependency_version": None,
                        "dependency_build": None,
                    }

                rows.append({
                    "artifact": artifact,
                    "dependency_name": dependency_name,
                    "dependency_version": dep_info["dependency_version"],
                    "dependency_build": dep_info["dependency_build"],
                })

    rows.sort(key=lambda x: x["artifact"])

    with open(output_csv, "w", newline="", encoding="utf-8") as fh:
        writer = csv.DictWriter(
            fh,
            fieldnames=[
                "artifact",
                "dependency_name",
                "dependency_version",
                "dependency_build",
            ],
        )
        writer.writeheader()
        writer.writerows(rows)

    print(output_csv)


if __name__ == "__main__":
    main()