def download_kernels()

in extras/kernels/download_kernels.py [0:0]

• 65 lines of code
• 12 McCabe index (conditional complexity)

def download_kernels(competition):
    """Downloads kernels for a given competition based on the kernel references."""

    kernel_refs_path = registry.get_competitions_dir() / competition.id / "kernels.txt"
    kernels_dir = registry.get_competitions_dir() / competition.id / "kernels"
    kernels_dir.mkdir(parents=True, exist_ok=True)

    kernel_refs = kernel_refs_path.read_text().splitlines()

    existing_kernels_py = []
    existing_kernels = []
    downloaded_kernels = []

    for kernel_ref in tqdm(
        kernel_refs,
        desc=f"Downloading kernels for {competition.id}",
        unit="kernel",
    ):
        author, kernel_slug = kernel_ref.split("/")
        py_script_path = kernels_dir / f"{author}_{kernel_slug}.py"
        notebook_path = kernels_dir / f"{author}_{kernel_slug}.ipynb"

        if notebook_path.exists() or py_script_path.exists():
            if notebook_path.exists():
                existing_kernels.append(kernel_ref)
            else:
                existing_kernels_py.append(kernel_ref)
            logger.debug(f"Kernel {kernel_ref} already exists. Skipping download.")
            continue
        else:
            try:
                kaggle.api.kernels_pull(kernel_ref, path=kernels_dir)
                downloaded_file = kernels_dir / f"{kernel_ref.split('/')[-1]}.ipynb"
                if downloaded_file.exists():
                    downloaded_kernels.append(kernel_ref)
                    new_filename = f"{author}_{downloaded_file.name}"
                    downloaded_file.rename(kernels_dir / new_filename)
            except kaggle.rest.ApiException as e:
                if e.status == 403:
                    logger.warning(
                        f"403 Forbidden error when downloading kernel {kernel_ref}. {e.body}. Skipping."
                    )
                else:
                    logger.error(f"Error downloading kernel {kernel_ref}: {str(e)}")

    total_kernels = len(kernel_refs)
    processed_kernels = len(downloaded_kernels + existing_kernels + existing_kernels_py)
    if processed_kernels != total_kernels:
        failed_kernels = set(kernel_refs) - set(
            downloaded_kernels + existing_kernels + existing_kernels_py
        )
        logger.warning(
            f"Failed to download {len(failed_kernels)} kernels for {competition.id}. "
            f"Failed kernels: {', '.join(failed_kernels)}"
        )

    for kernel_ref in tqdm(
        downloaded_kernels + existing_kernels,
        desc="Converting notebooks to Python scripts",
        unit="kernel",
    ):
        author, kernel_slug = kernel_ref.split("/")
        notebook_path = kernels_dir / f"{author}_{kernel_slug}.ipynb"
        py_script_path = kernels_dir / f"{author}_{kernel_slug}.py"
        try:
            ipynb_to_py(notebook_path, py_script_path)
            notebook_path.unlink()
        except Exception as e:
            logger.error(
                f"Error converting notebook {notebook_path} to Python script: {e}. Skipping."
            )
            if py_script_path.exists():
                py_script_path.unlink()  # Remove any partially converted file