List all Conan packages v2 ready in Conan Center

conan_v2_ready.py

import json
import os
import yaml
import argparse
import tempfile
import asyncio

from conan.api.conan_api import ConanAPI
from conan.api.output import ConanOutput
from conan.cli.command import OnceArgument
from conan.errors import ConanException
from conan.api.model import RecipeReference


def output_json(results):
    print(json.dumps(results, indent=2))


def background(f):
    def wrapped(*args, **kwargs):
        return asyncio.get_event_loop().run_in_executor(None, f, *args, **kwargs)

    return wrapped


def chunks(lst, n):
    for i in range(0, len(lst), n):
        yield lst[i:i + n]


async def has_package_in_remote(conan_api, out, recipe_name, version, remote):
    try:
        out.info(f"Listing {recipe_name} from {remote} remote")
        recipe_ref = RecipeReference(name=recipe_name, version=version)
        latest_revision = await conan_api.list.latest_recipe_revision(ref=recipe_ref, remote=remote)
        out.success(f"conan list {recipe_name}/{version}#{latest_revision.revision} -r {remote}")
        configurations = await conan_api.list.packages_configurations(latest_revision, remote=remote)
        out.success(f"conan list {recipe_name}/{version}#{latest_revision.revision}:* -r {remote}")
        return configurations != {}
    except ConanException as error:
        # The rev does not exist in the remote
        out.error(f"Error while listing revs for {recipe_name} in {remote}: {error}") \
            .warning(f"{recipe_name} does not exist in the remote")
        return False


async def export_recipe(conan_api, recipe_path, recipe_name, version):
    rref, conanfile = conan_api.export.export(recipe_path, recipe_name, version, None, None)
    return rref


async def can_export_recipes(conan_api, out, recipe_name, recipes):
    try:
        for version, conanfile in recipes.items():
            recipe_path = os.path.abspath(conanfile)
            rref = await export_recipe(conan_api, recipe_path, recipe_name, version)
            rev = f"{rref}#{rref.revision}"
            out.success(f"Exported {rev}")
        return True
    except ConanException as e:
        out.error(f"Could not export conanfile.py {e}")
        return False


def get_recipes_paths(out, config, cci_folder, recipe_name):
    recipes = {}

    for version in config["versions"]:
        recipe_folder = os.path.join(cci_folder, "recipes", recipe_name)
        recipe_subfolder = config["versions"][version]["folder"]
        conanfile = os.path.join(recipe_folder, recipe_subfolder, "conanfile.py")
        if not os.path.exists(conanfile):
            out.error(f"The recipe path does not exist: {conanfile}")
            return {}
        recipes[version] = os.path.join(recipe_folder, recipe_subfolder, "conanfile.py")
    return recipes


def get_config_file(out, cci_folder, recipe_name):
    recipe_folder = os.path.join(cci_folder, "recipes", recipe_name)
    config_file = os.path.join(recipe_folder, "config.yml")
    if not os.path.exists(config_file):
        out.error(f"The file {config_file} does not exist")
        return None
    with open(config_file, "r") as fd:
        return yaml.safe_load(fd)


async def get_conan_v2_ready_recipes(out, recipes_list, cci_folder, remote_arg):
    recipes_ready = []
    cache_folder = tempfile.TemporaryDirectory()
    conan_api = ConanAPI(cache_folder=cache_folder.name)
    remote = conan_api.remotes.get(remote_arg)

    for recipe_name in recipes_list:

        # load conan-center-index/recipes/{recipe_name}/config.yml
        config = get_config_file(out, cci_folder, recipe_name)
        if not config:
            continue

        # list all conanfile.py according to versions in config.yml
        recipes = get_recipes_paths(out, config, cci_folder, recipe_name)
        if not recipes:
            continue

        # run conan export locally to check Conan v2 compatibility for all versions
        if not await can_export_recipes(conan_api, out, recipe_name, recipes):
            out.error(f"Error while exporting {recipe_name}")
            continue

        # check for remote binary packages for each reference version
        packages = []
        for version in recipes.keys():
            has_package = await has_package_in_remote(conan_api, out, recipe_name, version, remote)
            packages.append(has_package)
            if not has_package:
                break
        # only recipes with all versions listed in remote with binary packages are ready for Conan v2
        if False not in packages:
            recipes_ready.append(recipe_name)

    return recipes_ready


async def main():
    """
    Lists which recipes are present in a given remote, and whether they have binaries available or not
    For this to work in conan-center-index, hook_reduce_conandata must be used and enabled with, user.hooks:reduce_conandata=True
    """
    parser = argparse.ArgumentParser()
    parser.add_argument('--path', default='/Users/uilian/Development/conan/conan-center-index', help="Path to ConanCenterIndex folder")
    parser.add_argument('-r', '--remote', default='conancenter', action=OnceArgument, help="Remote to query against")
    args = parser.parse_args()
    out = ConanOutput()

    recipes_to_create = sorted(os.listdir(os.path.join(args.path, "recipes")))
    recipes_chunks = chunks(recipes_to_create, 10)
    recipes_ready_list = await asyncio.gather(*(get_conan_v2_ready_recipes(out, chunk, args.path, args.remote) for chunk in recipes_chunks))
    recipes_ready = [element for sublist in recipes_ready_list for element in sublist]

    recipes_ready = sorted(list(set(recipes_ready)))
    out.info(f"There are {len(recipes_ready)} recipes ready for Conan v2:")
    output_json(recipes_ready)
    temp_json = tempfile.NamedTemporaryFile(suffix="conan-v2-ready.json").name
    with open(temp_json, "w") as fd:
        fd.write(json.dumps(recipes_ready))
    out.info(f"Saved JSON file: {temp_json}")


if __name__ == "__main__":
    asyncio.run(main())