Function check_duplicates_files

constructor/fcp.py:152–245

Check for duplicate files across packages and compute size/path metrics. Iterates through all files in the provided package cache records to: 1. Detect duplicate files (same path in multiple packages) 2. Compute approximate tarball and extracted sizes 3. Track the longest relat

(
    pc_recs: Iterable[PackageCacheRecord],
    platform: str,
    duplicate_files: Literal["error", "warn", "skip"] = "error",
    env_prefixes: dict[PackageCacheRecord, str] | None = None,
)

def check_duplicates_files(
    pc_recs: Iterable[PackageCacheRecord],
    platform: str,
    duplicate_files: Literal["error", "warn", "skip"] = "error",
    env_prefixes: dict[PackageCacheRecord, str] | None = None,
) -> tuple[int, int, int]:
    """
    Check for duplicate files across packages and compute size/path metrics.

    Iterates through all files in the provided package cache records to:
    1. Detect duplicate files (same path in multiple packages)
    2. Compute approximate tarball and extracted sizes
    3. Track the longest relative file path (for MAX_PATH validation on Windows)

    Args:
        pc_recs: Package cache records to check.
        platform: Target platform string (e.g., "win-64", "linux-64").
        duplicate_files: How to handle duplicates - "error", "warn", or "skip".
        env_prefixes: Optional dict mapping PackageCacheRecord -> path prefix string.
            Used to account for extra_envs paths which are installed under
            "envs/<name>/" rather than the base install directory. Records not
            in this dict are assumed to be in the base environment (no prefix).
            A trailing separator is added automatically if missing.

    Returns:
        Tuple of (approx_tarball_size, approx_extracted_size, max_relative_path_length)
    """
    assert duplicate_files in ("warn", "skip", "error")
    if env_prefixes is None:
        env_prefixes = {}
    for env, prefix in env_prefixes.items():
        if prefix and not prefix.endswith("/"):
            env_prefixes[env] += "/"

    map_members_scase = defaultdict(set)
    map_members_icase = defaultdict(lambda: {"files": set(), "fns": set()})

    # Keep a min, 50MB buffer size
    total_tarball_size = 52428800
    total_extracted_pkgs_size = 52428800
    max_relative_path_length = 0

    for pc_rec in pc_recs:
        fn = pc_rec.fn
        extracted_package_dir = pc_rec.extracted_package_dir

        total_tarball_size += int(pc_rec.get("size", 0))

        paths_data = read_paths_json(extracted_package_dir).paths
        env_prefix_len = len(env_prefixes.get(pc_rec, ""))
        for path_data in paths_data:
            short_path = path_data.path
            max_relative_path_length = max(
                max_relative_path_length, env_prefix_len + len(short_path)
            )
            try:
                size = path_data.size_in_bytes or getsize(join(extracted_package_dir, short_path))
            except AttributeError: