Source code for craft_parts.executor.organize

# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
#
# Copyright 2015-2025 Canonical Ltd.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License version 3 as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""Handle part files organization.

Installed part files can be reorganized according to a mapping specified
under the `organize` entry in a part definition. In the key/value pair,
the key represents the path of a file inside the part and the value
represents how the file is going to be staged.
"""

from __future__ import annotations

import contextlib
import os
import shutil
from glob import iglob
from typing import TYPE_CHECKING

from craft_parts import errors
from craft_parts.utils import file_utils, path_utils
from craft_parts.utils.partition_utils import DEFAULT_PARTITION

if TYPE_CHECKING:
    from collections.abc import Mapping
    from pathlib import Path




def organize_files(
    *,
    part_name: str,
    file_map: dict[str, str],
    install_dir_map: Mapping[str | None, Path],
    overwrite: bool,
    default_partition: str,
) -> None:
    """Rearrange files for part staging.

    If partitions are enabled, source filepaths must be in the default partition.
    The default partition can be referenced by the provided default_partition name
    or by the DEFAULT_PARTITION value.

    :param part_name: The name of the part to organize files for.
    :param file_map: A mapping of source filepaths to destination filepaths.
    :param install_dir_map: A mapping of partition names to their install directories.
    :param overwrite: Whether existing files should be overwritten. This is
        only used in build updates, when a part may organize over files
        it previously organized.

    :raises FileOrganizeError: If the destination file already exists or multiple files
        are organized to the same destination.
    :raises FileOrganizeError: If partitions are enabled and the source file is not from
        the default partition.
    """
    for key in sorted(file_map, key=lambda x: ["*" in x, x]):
        src_partition, src_inner_path = path_utils.get_partition_and_path(
            key, default_partition
        )

        if src_partition and src_partition not in [
            default_partition,
            DEFAULT_PARTITION,
        ]:
            raise errors.FileOrganizeError(
                part_name=part_name,
                message=(
                    f"Cannot organize files from {src_partition!r} partition. "
                    f"Files can only be organized from the {default_partition!r} partition"
                ),
            )
        # Replace default partition default name with alias name to allow
        # using (default) in paths even with aliased default partition
        if src_partition == DEFAULT_PARTITION:
            src_partition = default_partition

        src = os.path.join(install_dir_map[src_partition], src_inner_path)

        # Remove the leading slash so the path actually joins
        # Also trailing slash is significant, be careful if using pathlib!
        dst_partition, dst_inner_path = path_utils.get_partition_and_path(
            file_map[key].lstrip("/"),
            default_partition,
        )

        # Replace default partition default name with alias name to allow
        # using (default) in paths even with aliased default partition
        if dst_partition == DEFAULT_PARTITION:
            dst_partition = default_partition

        dst = os.path.join(install_dir_map[dst_partition], dst_inner_path)

        # prefix the partition to the log-friendly version of the destination
        if dst_partition and dst_partition != default_partition:
            dst_string = f"({dst_partition})/{dst_inner_path}"
        else:
            dst_string = str(dst_inner_path)

        sources = iglob(src, recursive=True)

        # Keep track of the number of glob expansions so we can properly error if more
        # than one tries to organize to the same file
        src_count = 0
        for src in sources:
            src_count += 1

            if os.path.isdir(src) and "*" not in key:
                file_utils.link_or_copy_tree(src, dst)
                shutil.rmtree(src)
                continue

            if os.path.isfile(dst):
                if overwrite and src_count <= 1:
                    with contextlib.suppress(FileNotFoundError):
                        os.remove(dst)
                elif src_count > 1:
                    raise errors.FileOrganizeError(
                        part_name=part_name,
                        message=(
                            "multiple files to be organized into "
                            f"{dst_string!r}. If this is "
                            "supposed to be a directory, end it with a slash."
                        ),
                    )
                else:
                    raise errors.FileOrganizeError(
                        part_name=part_name,
                        message=(
                            f"trying to organize file {key!r} to "
                            f"{file_map[key]!r}, but "
                            f"{dst_string!r} already exists"
                        ),
                    )

            if os.path.isdir(dst) and overwrite:
                real_dst = os.path.join(dst, os.path.basename(src))
                if os.path.isdir(real_dst):
                    shutil.rmtree(real_dst)
                else:
                    with contextlib.suppress(FileNotFoundError):
                        os.remove(real_dst)

            os.makedirs(os.path.dirname(dst), exist_ok=True)
            shutil.move(src, dst)