cdm_data_loaders.ncbi_ftp.assembly ¶

def parse_md5_checksums_file(text: str) -> dict[PurePosixPath, str]:
    """Parse an NCBI ``md5checksums.txt`` file into a filename-to-hash mapping.

    Each line has the format ``<md5>  ./<filename>`` (two-space separator).

    :param text: raw text of the md5checksums.txt file
    :type text: str
    :return: dict mapping filename to MD5 hex digest
    """
    checksums: dict[PurePosixPath, str] = {}
    for raw_line in text.strip().splitlines():
        parts = raw_line.strip().split("  ", maxsplit=1)
        if len(parts) == 2:  # noqa: PLR2004
            md5_hash, filename = parts
            checksums[PurePosixPath(filename.removeprefix("./"))] = md5_hash.strip()
    return checksums

def filter_assembly_checksums(checksums: dict[PurePosixPath, str]) -> dict[PurePosixPath, str]:
    """Filter a parsed ``md5checksums.txt`` mapping to files matching ``FILE_FILTERS``."""
    return {fname: md5 for fname, md5 in checksums.items() if any(fname.match(f"**{s}") for s in FILE_FILTERS)}

def build_accession_path(assembly_dir: PurePosixPath) -> PurePosixPath:
    """Build the relative output path for an assembly directory.

    Produces ``raw_data/{GCF|GCA}/{000}/{001}/{215}/{assembly_dir}/``.

    :param assembly_dir: full assembly directory name (e.g. ``GCF_000001215.4_Release_6...``)
    :type assembly_dir: PurePosixPath
    :return: relative path
    :raises ValueError: if the assembly directory name cannot be parsed
    """
    if m := ACCESSION_PARTS_REGEX.match(f"{assembly_dir}"):  # returns e.g., "GCF", "000", "001", "215" captured groups
        return PurePosixPath("raw_data", *m.groups(), assembly_dir)
    msg = f"Cannot parse accession: {assembly_dir}"
    raise ValueError(msg)

def parse_assembly_path(assembly_path: PurePosixPath) -> tuple[str, PurePosixPath, str]:
    """Extract database, assembly_dir, and accession from an FTP assembly path.

    :param assembly_path: FTP directory path (e.g. ``/genomes/all/GCF/000/.../GCF_000001215.4_Rel.../``)
    :type assembly_path: PurePosixPath
    :return: tuple of ``(database, assembly_dir, accession)``
    :raises ValueError: if the path cannot be parsed
    """
    if m := ASSEMBLY_PATH_REGEX.search(str(assembly_path)):
        return m.group(3), PurePosixPath(m.group(1)), m.group(2)
    msg = f"Cannot parse assembly path: {assembly_path}"
    raise ValueError(msg)

def download_assembly_to_local(
    assembly_path: PurePosixPath,
    output_dir: Path,
    ftp_host: str = FTP_HOST,
    ftp: FTP | None = None,
) -> dict[str, Any]:
    """Download one assembly from NCBI FTP to a local directory.

    Creates a directory structure under *output_dir* matching the S3 layout,
    downloads filtered files, verifies MD5 checksums, and writes ``.md5``
    sidecar files for downstream metadata.

    :param assembly_path: FTP directory path for the assembly
    :type assembly_path: PurePosixPath
    :param output_dir: base output directory
    :type output_dir: Path
    :param ftp_host: FTP hostname
    :type ftp_host: str
    :param ftp: optional existing FTP connection (caller manages lifecycle)
    :type ftp: FTP | None
    :return: json parsable dict with download statistics
    """
    _database, assembly_dir, accession = parse_assembly_path(assembly_path)
    rel_path = build_accession_path(assembly_dir)
    dest_dir: Path = output_dir / rel_path
    dest_dir.mkdir(parents=True, exist_ok=True)

    logger.debug("Downloading %s -> %s", accession, dest_dir)

    stats: dict[str, Any] = {
        "accession": accession,
        "assembly_dir": str(assembly_dir),
        "files_downloaded": 0,
        "files_skipped_checksum_mismatch": 0,
        "files_without_checksum": 0,
    }

    with ftp_session(ftp_host, ftp) as active_ftp:
        files = ftp_list_dir(active_ftp, assembly_path)

        # Download and parse md5checksums.txt
        md5_checksums: dict[PurePosixPath, str] = {}
        if PurePosixPath("md5checksums.txt") in files:
            md5_text = ftp_retrieve_text(active_ftp, PurePosixPath("md5checksums.txt"))
            md5_checksums = parse_md5_checksums_file(md5_text)
            (dest_dir / "md5checksums.txt").write_text(md5_text)
            stats["files_downloaded"] += 1

        target_files = [f for f in files if any(f.match(f"**{s}") for s in FILE_FILTERS)]
        last_activity = time.monotonic()

        for filename in target_files:
            last_activity = _download_and_verify(active_ftp, filename, dest_dir, md5_checksums, stats, last_activity)

        logger.debug("  %s: %d files downloaded", accession, stats["files_downloaded"])

    return stats