# Source code for debusine.artifacts.local_artifact

    # Copyright © The Debusine Developers
    # See the AUTHORS file at the top-level directory of this distribution
    #
    # This file is part of Debusine. It is subject to the license terms
    # in the LICENSE file found in the top-level directory of this
    # distribution. No part of Debusine, including this file, may be copied,
    # modified, propagated, or distributed except according to the terms
    # contained in the LICENSE file.

    """
    Local artifact's representation.

    See docs/design/ontology.html for the semantics of the Artifacts.
    """

    import abc
    import json
    import re
    from collections.abc import Iterable, Sequence
    from json import JSONDecodeError
    from pathlib import Path, PurePath
    from typing import Any, ClassVar, Self, assert_never, overload

    import pydantic

    # https://github.com/pydantic/pydantic/issues/12720
    from pydantic._internal._generics import get_origin

    import debusine.artifacts.models as data_models
    from debian import deb822, debfile
    from debusine import utils
    from debusine.artifacts.artifact_utils import files_in_meta_file_match_files
    from debusine.assets import KeyPurpose
    from debusine.client.models import FileRequest
    from debusine.utils import extract_generic_type_arguments

    class NewLocalFile(pydantic.BaseModel):
        """A new local file to be added to the artifact."""

        #: File in the local file system to be added to the artifact.
        file: Path
        #: Base directory of the artifact.  Must be an absolute path.  If it's
        #: None, the file is added in the root of the artifact; otherwise, it is
        #: added with the path of the file relative to `artifact_base_dir`.
        artifact_base_dir: Path | None = None
        #: Name to use for the file in the artifact instead of `file.name`.
        override_name: str | None = None
        #: Content-Type to use when serving the file instead of guessing.
        content_type: str | None = None

    class NewRemoteFile(pydantic.BaseModel):
        """
        A new remote file to be added to the artifact.

        This may be used when a file does not exist locally, but can be expected
        to already exist on the server when uploading the rest of the artifact.
        """

        #: Path of the file in the artifact.
        path_in_artifact: str
        #: Description of the remote file.
        file_request: FileRequest

    class NascentLocalArtifact(pydantic.BaseModel):
        """Information about files that will be added to a new local artifact."""

        local_files: dict[str, Path] = pydantic.Field(default_factory=dict)
        remote_files: dict[str, FileRequest] = pydantic.Field(default_factory=dict)
        content_types: dict[str, str] = pydantic.Field(default_factory=dict)

        def _check_existing_file(
            self, path_in_artifact: str, new_file: Path | FileRequest
        ) -> None:
            """Check whether the nascent artifact already contains a given file."""
            all_files = self.local_files | self.remote_files
            if path_in_artifact in all_files:
                raise ValueError(
                    f"File with the same path ({path_in_artifact}) "
                    f"is already in the artifact "
                    f'("{all_files[path_in_artifact]}" and "{new_file}")'
                )

        def add_local_file(self, new_file: NewLocalFile) -> None:
            """
            Prepare a local file for addition to the artifact.

            :param new_file: information about the new local file.
            :raises ValueError: artifact_base_dir is not absolute or is not
              a directory; file does not exist; the path in the artifact already
              had a file.
            """
            file = new_file.file
            if new_file.artifact_base_dir is not None:
                if not new_file.artifact_base_dir.is_absolute():
                    raise ValueError(
                        f'"{new_file.artifact_base_dir}" must be absolute'
                    )
                if not new_file.artifact_base_dir.is_dir():
                    raise ValueError(
                        f'"{new_file.artifact_base_dir}" does not exist or '
                        f'is not a directory'
                    )

                if not file.is_absolute():
                    file = new_file.artifact_base_dir.joinpath(file)

                path_in_artifact = file.relative_to(
                    new_file.artifact_base_dir
                ).as_posix()
            else:
                path_in_artifact = file.name

            if new_file.override_name is not None:
                path_in_artifact = new_file.override_name

            if not file.exists():
                raise ValueError(f'"{file}" does not exist')

            if not file.is_file():
                raise ValueError(f'"{file}" is not a file')

            file_absolute = file.absolute()

            self._check_existing_file(path_in_artifact, file_absolute)
            self.local_files[path_in_artifact] = file_absolute
            if new_file.content_type is not None:
                self.content_types[path_in_artifact] = new_file.content_type

        def add_remote_file(self, new_file: NewRemoteFile) -> None:
            """
            Prepare a remote file for addition to the artifact.

            :param new_file: information about the new remote file.
            """
            self._check_existing_file(
                new_file.path_in_artifact, new_file.file_request
            )
            self.remote_files[new_file.path_in_artifact] = new_file.file_request

    class LocalArtifact[AD: data_models.ArtifactData](pydantic.BaseModel, abc.ABC):
        """Represent an artifact locally."""

        model_config = pydantic.ConfigDict(validate_assignment=True, extra="forbid")

        #: Artifact type
        category: data_models.ArtifactCategory
        # Keys are paths in the artifact. Values the paths in the local system
        files: dict[str, Path] = pydantic.Field(default_factory=dict)
        # Keys are paths in the artifact. Values are FileRequests for files that do
        # not exist locally but can be expected to already exist on the server.
        remote_files: dict[str, FileRequest] = pydantic.Field(default_factory=dict)
        # Keys are paths in the artifact. Values are MIME types, suitable for
        # Content-Type headers.
        content_types: dict[str, str] = pydantic.Field(default_factory=dict)
        #: Artifact data
        data: AD
        # TODO: it would be great to not have to redefine data in subclasses, but
        # it needs pydantic's Generics support, which might require work to work
        # with extract_generic_type_arguments

        #: Default value for category
        _category: ClassVar[data_models.ArtifactCategory]

        #: Class used as the in-memory representation of artifact data.
        _data_type: type[AD] = pydantic.PrivateAttr()
        # data_type is marked as PrivateAttr to make mypy happy. Setting
        # underscore_attrs_are_private to True does not seem to be enough

        _local_artifacts_category_to_class: ClassVar[
            dict[str, type["LocalArtifact['Any']"]]
        ] = {}

        def __init_subclass__(cls, **kwargs: Any) -> None:
            """
            Register subclass into LocalArtifact._local_artifacts_category_to_class.

            Allow to list possible valid options (in the client or server).
            """
            super().__init_subclass__(**kwargs)

            if get_origin(cls) is not None:
                # The task data type, computed by introspecting the type argument
                # used to specialize this generic class.
                [cls._data_type] = extract_generic_type_arguments(
                    cls, LocalArtifact
                )

                LocalArtifact._local_artifacts_category_to_class[
                    # This should just be cls._category, but if we try to use
                    # that in mypy we get:
                    # TypeError: unhashable type: 'ModelPrivateAttr'
                    cls._category.default  # type: ignore[attr-defined]
                ] = cls

        @classmethod
        def create_data(cls, data_dict: dict[str, Any]) -> AD:
            """Instantiate a data model from a dict."""
            return cls._data_type(**data_dict)

        @classmethod
        def _create(
            cls,
            data: dict[str, Any] | AD,
            files: Sequence[NewLocalFile | NewRemoteFile] | None = None,
        ) -> Self:
            """
            Create a new instance of this model.

            If possible, use the higher-level ``create`` method of the
            appropriate subclass instead.
            """
            if isinstance(data, dict):
                data = cls.create_data(data)
            nascent = NascentLocalArtifact()
            for file in files or ():
                match file:
                    case NewLocalFile():
                        nascent.add_local_file(file)
                    case NewRemoteFile():
                        nascent.add_remote_file(file)
                    case _ as unreachable:
                        assert_never(unreachable)
            return cls(
                category=cls._category,
                data=data,
                files=nascent.local_files,
                remote_files=nascent.remote_files,
                content_types=nascent.content_types,
            )

        @staticmethod
        def artifact_categories() -> list[str]:
            """Return list of artifact categories."""
            return sorted(LocalArtifact._local_artifacts_category_to_class.keys())

        @pydantic.field_validator("category", mode="before")
        @classmethod
        def _validate_category(cls, category: Any) -> Any:
            """Validate that the category is known."""
            if category not in cls._local_artifacts_category_to_class:
                raise ValueError(
                    f"Invalid category: '{category}'. Expected one of "
                    f"{', '.join(sorted(cls._local_artifacts_category_to_class))}."
                )
            return category

        @staticmethod
        def class_from_category(category: str) -> type["LocalArtifact['Any']"]:
            """Return class sub_local_artifact."""
            category = LocalArtifact._validate_category(category)
            return LocalArtifact._local_artifacts_category_to_class[category]

        @classmethod
        def _validate_files_length(
            cls, files: dict[str, Path], number_of_files: int
        ) -> dict[str, Path]:
            """Raise ValueError if number of files is not number_of_files."""
            if (actual_number_files := len(files)) != number_of_files:
                raise ValueError(
                    f"Expected number of files: {number_of_files} "
                    f"Actual: {actual_number_files}"
                )
            return files

        @classmethod
        def _validate_files_end_in(
            cls, files: dict[str, Path], suffixes: Sequence[str]
        ) -> dict[str, Path]:
            """Raise ValueError if any file does not end in one of suffixes."""
            for file_name in files.keys():
                if not file_name.endswith(tuple(suffixes)):
                    raise ValueError(
                        f'Valid file suffixes: {suffixes}. '
                        f'Invalid filename: "{file_name}"'
                    )
            return files

        @classmethod
        def _validate_exactly_one_file_ends_in[T: Any](
            cls, files: dict[str, T], suffix: str
        ) -> dict[str, T]:
            """Raise ValueError if files doesn't have exactly 1 file with suffix."""
            changes_files = sum(1 for file in files if file.endswith(suffix))
            if changes_files != 1:
                raise ValueError(
                    f"Expecting 1 {suffix} file in {sorted(files.keys())}"
                )
            return files

    class WorkRequestDebugLogs(LocalArtifact[data_models.EmptyArtifactData]):
        """
        WorkRequestDebugLogs: help debugging issues executing the task.

        Log files for debusine users in order to debug possible problems in their
        WorkRequests.
        """

        _category = data_models.ArtifactCategory.WORK_REQUEST_DEBUG_LOGS
        data: data_models.EmptyArtifactData = pydantic.Field(
            default_factory=data_models.EmptyArtifactData
        )

        @classmethod
        def create(cls, *, files: Iterable[Path]) -> Self:
            """Return a WorkRequestDebugLogs."""
            return cls._create(
                data={},
                files=[
                    NewLocalFile(
                        file=file, content_type="text/plain; charset=utf-8"
                    )
                    for file in files
                ],
            )

    class DebusineTest(LocalArtifact[data_models.EmptyArtifactData]):
        """DebusineTest: noop artifact used in Debusine tests."""

        _category = data_models.ArtifactCategory.TEST
        data: data_models.EmptyArtifactData = pydantic.Field(
            default_factory=data_models.EmptyArtifactData
        )

    class PackageBuildLog(LocalArtifact[data_models.DebianPackageBuildLog]):
        """PackageBuildLog: represents a build log file."""

        _category = data_models.ArtifactCategory.PACKAGE_BUILD_LOG

        @classmethod
        def create(
            cls,
            *,
            file: Path,
            source: str,
            version: str,
            architecture: str | None = None,
            bd_uninstallable: data_models.DoseDistCheck | None = None,
        ) -> Self:
            """Return a PackageBuildLog."""
            return cls._create(
                data=data_models.DebianPackageBuildLog(
                    source=source,
                    version=version,
                    architecture=architecture,
                    filename=file.name,
                    bd_uninstallable=bd_uninstallable,
                ),
                files=[
                    NewLocalFile(
                        file=file, content_type="text/plain; charset=utf-8"
                    )
                ],
            )

        @pydantic.field_validator("files", mode="after")
        @classmethod
        def validate_files_length_is_one(
            cls, files: dict[str, Path]
        ) -> dict[str, Path]:
            """Validate that artifact has only one file."""
            return super()._validate_files_length(files, 1)

        @pydantic.field_validator("files", mode="after")
        @classmethod
        def file_must_end_in_build(cls, files: dict[str, Path]) -> dict[str, Path]:
            """Raise ValueError if the file does not end in .build."""
            return super()._validate_files_end_in(files, [".build"])

    @overload
    def deb822dict_to_dict(
        element: dict[Any, Any] | deb822.Deb822Dict,
    ) -> dict[Any, Any]: ...

    @overload
    def deb822dict_to_dict(element: Any) -> Any: ...

    def deb822dict_to_dict(element: Any) -> Any:
        """
        Traverse recursively element converting Deb822Dict to dict.

        deb822.Changes() return a Deb822Dict with some other inner-elements
        being Deb822Dict. This function traverse it and converts any elements
        being a Deb822Dict to a Python dict.

        Reason is to simplify is that json module cannot encode Deb822Dict. To
        simplify let's convert Deb822Dict as soon as possible to dict.
        """
        result: Any
        if isinstance(element, (dict, deb822.Deb822Dict)):
            result = {}
            for key, value in element.items():
                result[key] = deb822dict_to_dict(value)
        elif isinstance(element, list):
            result = []
            for item in element:
                result.append(deb822dict_to_dict(item))
        else:
            result = element

        return result

    class Upload(LocalArtifact[data_models.DebianUpload]):
        """Upload: encapsulate a .changes and files listed in it."""

        _category = data_models.ArtifactCategory.UPLOAD

        @classmethod
        def create(cls, *, changes_file: Path, allow_remote: bool = False) -> Self:
            """
            Return a Upload. Add the changes_files and files listed in it.

            :param changes_file: a .changes file. Parsed by deb822.Changes.
            :param allow_remote: if True, allow files referenced by the .changes
              file to be missing locally; uploading this artifact will require
              them to already exist on the server.
            """
            with changes_file.open() as changes_obj:
                data = data_models.DebianUpload(
                    type="dpkg",
                    changes_fields=deb822dict_to_dict(deb822.Changes(changes_obj)),
                )

            files: list[NewLocalFile | NewRemoteFile] = []
            files.append(
                NewLocalFile(
                    file=changes_file, content_type="text/plain; charset=utf-8"
                )
            )

            # Add any files referenced by .changes.
            base_directory = changes_file.parent
            for file in data.changes_fields.get("Checksums-Sha256", []):
                path = base_directory / file["name"]
                if allow_remote and not path.exists():
                    files.append(
                        NewRemoteFile(
                            path_in_artifact=file["name"],
                            file_request=FileRequest(
                                size=file["size"],
                                checksums={"sha256": file["sha256"]},
                                type="file",
                            ),
                        )
                    )
                else:
                    files.append(NewLocalFile(file=path))

            return cls._create(data=data, files=files)

        @pydantic.model_validator(mode="after")
        def files_contain_changes(self) -> Self:
            """Raise ValueError when files does not have exactly 1 .changes file."""
            self._validate_exactly_one_file_ends_in(
                {**self.files, **self.remote_files}, ".changes"
            )
            return self

        @pydantic.model_validator(mode="after")
        def files_contains_files_in_changes(self) -> Self:
            """
            Validate that set(files) == set(files_in_changes_file).

            Exception: The .changes file must be in files but not in the .changes
            file.
            """
            files_in_meta_file_match_files(
                ".changes", deb822.Changes, {**self.files, **self.remote_files}
            )
            return self

    class SourcePackage(LocalArtifact[data_models.DebianSourcePackage]):
        """SourcePackage: contains source code to be built into BinaryPackages."""

        _category = data_models.ArtifactCategory.SOURCE_PACKAGE

        @classmethod
        def create(cls, *, name: str, version: str, files: list[Path]) -> Self:
            """Return a SourcePackage setting files and data."""
            dsc_fields = {}
            for file in files:
                if file.suffix == ".dsc":
                    dsc = utils.read_dsc(file)
                    if dsc is None:
                        raise ValueError(f"{file} is not a valid .dsc file")
                    dsc_fields = deb822dict_to_dict(dsc)

            data = data_models.DebianSourcePackage(
                name=name,
                version=version,
                type="dpkg",
                dsc_fields=dsc_fields,
            )

            return cls._create(
                data=data, files=[NewLocalFile(file=file) for file in files]
            )

        @pydantic.field_validator("files", mode="after")
        @classmethod
        def files_contain_one_dsc(cls, files: dict[str, Path]) -> dict[str, Path]:
            """Raise ValueError when files does not have exactly 1 .dsc file."""
            return cls._validate_exactly_one_file_ends_in(files, ".dsc")

        @pydantic.field_validator("files", mode="after")
        @classmethod
        def files_contains_files_in_dsc(
            cls, files: dict[str, Path]
        ) -> dict[str, Path]:
            """
            Validate that set(files) == set(files_in_dsc_file).

            Exception: The .dsc file must be in files but not in the .dsc file.
            """
            files_in_meta_file_match_files(".dsc", deb822.Dsc, files)
            return files

    class BinaryPackage(LocalArtifact[data_models.DebianBinaryPackage]):
        r"""BinaryPackage: encapsulates a single \*.deb / \*.udeb."""

        _category = data_models.ArtifactCategory.BINARY_PACKAGE

        @classmethod
        def create(cls, *, file: Path) -> Self:
            """Return a BinaryPackage setting file and data."""
            pkg = debfile.DebFile(file)
            try:
                control = pkg.control.debcontrol()
                control_files = sorted(
                    name.lstrip("./")
                    for name in pkg.control
                    if name.startswith("./")
                )
            finally:
                pkg.close()
            # Work around Debian #1082838
            srcpkg_name = control.get("Source", control["Package"])
            if (m := re.match(r"^(.*) \((.*)\)$", srcpkg_name)) is not None:
                srcpkg_name, srcpkg_version = m.groups()
            else:
                srcpkg_version = control["Version"]
            data = data_models.DebianBinaryPackage(
                srcpkg_name=srcpkg_name,
                srcpkg_version=srcpkg_version,
                deb_fields=deb822dict_to_dict(control),
                deb_control_files=control_files,
            )

            return cls._create(data=data, files=[NewLocalFile(file=file)])

        @pydantic.field_validator("files", mode="after")
        @classmethod
        def files_must_end_in_deb_or_udeb(
            cls, files: dict[str, Path]
        ) -> dict[str, Path]:
            """Raise ValueError if a file does not end in .deb or .udeb."""
            return super()._validate_files_end_in(files, [".deb", ".udeb"])

        @pydantic.field_validator("files", mode="after")
        @classmethod
        def files_exactly_one(cls, files: dict[str, Path]) -> dict[str, Path]:
            """Raise ValueError if len(files) != 1."""
            if len(files) != 1:
                raise ValueError("Must have exactly one file")

            return files

    class BinaryPackages(LocalArtifact[data_models.DebianBinaryPackages]):
        r"""BinaryPackages: encapsulates a group of \*.deb / \*.udeb."""

        _category = data_models.ArtifactCategory.BINARY_PACKAGES

        @classmethod
        def create(
            cls,
            *,
            srcpkg_name: str,
            srcpkg_version: str,
            version: str,
            architecture: str,
            files: list[Path],
        ) -> Self:
            """Return a BinaryPackages setting files and data."""
            data = data_models.DebianBinaryPackages(
                srcpkg_name=srcpkg_name,
                srcpkg_version=srcpkg_version,
                version=version,
                architecture=architecture,
                # It might be better to get this from the metadata in the
                # package, but that would be rather more effort and isn't
                # currently vital.
                packages=[path.name.split("_", 1)[0] for path in files],
            )

            return cls._create(
                data=data, files=[NewLocalFile(file=file) for file in files]
            )

        @pydantic.field_validator("files", mode="after")
        @classmethod
        def files_must_end_in_deb_or_udeb(
            cls, files: dict[str, Path]
        ) -> dict[str, Path]:
            """Raise ValueError if a file does not end in .deb or .udeb."""
            return super()._validate_files_end_in(files, [".deb", ".udeb"])

        @pydantic.field_validator("files", mode="after")
        @classmethod
        def files_more_than_zero(cls, files: dict[str, Path]) -> dict[str, Path]:
            """Raise ValueError if len(files) == 0."""
            if len(files) == 0:
                raise ValueError("Must have at least one file")

            return files

    class LintianArtifact(LocalArtifact[data_models.DebianLintian]):
        """LintianArtifact: encapsulate result of the Lintian run."""

        _category = data_models.ArtifactCategory.LINTIAN

        @classmethod
        def create(
            cls,
            analysis: Path,
            lintian_output: Path,
            architecture: str,
            summary: data_models.DebianLintianSummary,
        ) -> Self:
            """Return a LintianArtifact with the files set."""
            data = data_models.DebianLintian(
                architecture=architecture, summary=summary
            )

            return cls._create(
                data=data,
                files=[
                    NewLocalFile(file=analysis, override_name="analysis.json"),
                    NewLocalFile(file=lintian_output, override_name="lintian.txt"),
                ],
            )

        @pydantic.field_validator("files", mode="after")
        @classmethod
        def _validate_required_files(
            cls, files: dict[str, Path]
        ) -> dict[str, Path]:
            """Artifact contain "analysis.json, "lintian.txt"."""
            # The .create() method already enforces having the correct files
            # But the artifact can be created using debusine.client or a web form,
            # which do not use the .create() of LintianArtifact but the
            # LocalArtifact. That's the reason that is needed to validate
            # that the required files are attached
            required_files = {"analysis.json", "lintian.txt"}
            if files.keys() != required_files:
                raise ValueError(f"Files required: {sorted(required_files)}")

            return files

        @staticmethod
        def _file_is_json_or_raise_value_error(
            file_name: str, files: dict[str, Path]
        ) -> None:
            """Raise ValueError() if file_name in files is not valid JSON."""
            with files[file_name].open() as file:
                try:
                    json.load(file)
                except JSONDecodeError as exc:
                    raise ValueError(f"{file_name} is not valid JSON: {exc}")

        @pydantic.field_validator("files", mode="after")
        @classmethod
        def _validate_file_analysis_is_json(
            cls, files: dict[str, Path]
        ) -> dict[str, Path]:
            """Validate that "analysis.json" is valid JSON."""
            cls._file_is_json_or_raise_value_error("analysis.json", files)
            return files

    class AutopkgtestArtifact(LocalArtifact[data_models.DebianAutopkgtest]):
        """Autopkgtest: encapsulate result of the Autopkgtest run."""

        _category = data_models.ArtifactCategory.AUTOPKGTEST

        @classmethod
        def create(
            cls, artifact_directory: Path, data: data_models.DebianAutopkgtest
        ) -> Self:
            """Return AutopkgtestArtifact with the files and data set."""
            files = []

            for file in artifact_directory.rglob("*"):
                if not file.is_file():
                    # Only add files
                    continue

                if file.is_relative_to(artifact_directory / "binaries"):
                    # Skip binaries/
                    continue

                files.append(
                    NewLocalFile(file=file, artifact_base_dir=artifact_directory)
                )

            return cls._create(data=data, files=files)

    class PiupartsArtifact(LocalArtifact[data_models.DebianPiuparts]):
        """Piuparts: encapsulate result of a ``piuparts`` run."""

        _category = data_models.ArtifactCategory.PIUPARTS

        @classmethod
        def create(
            cls, *, piuparts_output: Path, data: data_models.DebianPiuparts
        ) -> Self:
            """Return a ``PiupartsArtifact`` with the files and data set."""
            return cls._create(
                data=data,
                files=[
                    NewLocalFile(file=piuparts_output, override_name="piuparts.txt")
                ],
            )

    class DebianSystemTarballArtifact(
        LocalArtifact[data_models.DebianSystemTarball]
    ):
        """
        Contain system.tar.xz file with a Debian.

        Can be used by a chroot, container, etc.
        """

        _category = data_models.ArtifactCategory.SYSTEM_TARBALL

        @classmethod
        def create(cls, tarball: Path, data: dict[str, Any]) -> Self:
            """Return a DebianSystemTarballArtifact with the tarball file."""
            data = data.copy()
            data["filename"] = tarball.name
            return cls._create(
                data=data_models.DebianSystemTarball(**data),
                files=[NewLocalFile(file=tarball)],
            )

        @pydantic.field_validator("files", mode="after")
        @classmethod
        def _validate_file_name_ends_in_tar_xz(
            cls, files: dict[str, Path]
        ) -> dict[str, Path]:
            """Check if the artifact contains only one file and it is a .tar.xz."""
            if not len(files) == 1:
                raise ValueError(
                    "DebianSystemTarballArtifact does not contain exactly one file"
                )

            if not (name := next(iter(files.keys()))).endswith(".tar.xz"):
                raise ValueError(f"Invalid file name: '{name}'. Expected .tar.xz")

            return files

    class BlhcArtifact(LocalArtifact[data_models.DebianBlhc]):
        """BlhcArtifact: encapsulate result of the blhc run."""

        _category = data_models.ArtifactCategory.BLHC

        @classmethod
        def create(
            cls,
            blhc_output: Path,
            build_log: data_models.DebianPackageBuildLog | None,
        ) -> Self:
            """Return a BlhcArtifact with the files set."""
            return cls._create(
                data=data_models.DebianBlhc(
                    source=None if build_log is None else build_log.source,
                    version=None if build_log is None else build_log.version,
                    architecture=(
                        None if build_log is None else build_log.architecture
                    ),
                ),
                files=[NewLocalFile(file=blhc_output, override_name="blhc.txt")],
            )

    class DebDiffArtifact(LocalArtifact[data_models.DebDiff]):
        """DebDiffArtifact: encapsulate result of the debdiff run."""

        _category = data_models.ArtifactCategory.DEBDIFF

        @classmethod
        def create(
            cls,
            debdiff_output: Path,
            original: str,
            new: str,
            *,
            label: str | None = None,
        ) -> Self:
            """Return a DebDiffArtifact with the files set."""
            return cls._create(
                data=data_models.DebDiff(original=original, new=new, label=label),
                files=[
                    NewLocalFile(file=debdiff_output, override_name="debdiff.txt")
                ],
            )

    class DebianSystemImageArtifact(LocalArtifact[data_models.DebianSystemImage]):
        """
        Contains a image.tar.xz file with a bootable Debian system.

        Can be used by a VM.
        """

        _category = data_models.ArtifactCategory.SYSTEM_IMAGE

        @classmethod
        def create(cls, image: Path, data: dict[str, Any]) -> Self:
            """Return a DebianSystemImageArtifact with the image file."""
            data = data.copy()
            data["filename"] = image.name
            return cls._create(
                data=data_models.DebianSystemImage(**data),
                files=[NewLocalFile(file=image)],
            )

        @pydantic.field_validator("files", mode="after")
        @classmethod
        def _validate_files(cls, files: dict[str, Path]) -> dict[str, Path]:
            """Check if the artifact contains only one file and it's ending."""
            if not len(files) == 1:
                raise ValueError(
                    "DebianSystemImageArtifact does not contain exactly one file"
                )

            name = next(iter(files.keys()))
            if not name.endswith(".tar.xz") and not name.endswith(".qcow2"):
                raise ValueError(
                    f"Invalid file name: '{name}'. Expected .tar.xz or qcow2"
                )

            return files

    class SigningInputArtifact(LocalArtifact[data_models.DebusineSigningInput]):
        """Input to a Sign task."""

        _category = data_models.ArtifactCategory.SIGNING_INPUT

        @classmethod
        def create(
            cls,
            files: Iterable[Path],
            base_dir: Path,
            trusted_certs: list[str] | None = None,
            binary_package_name: str | None = None,
        ) -> Self:
            """Return a new SigningInputArtifact."""
            data = data_models.DebusineSigningInput(
                trusted_certs=(
                    None if trusted_certs is None else list(trusted_certs)
                ),
                binary_package_name=binary_package_name,
            )
            return cls._create(
                data=data,
                files=[
                    NewLocalFile(
                        file=file,
                        artifact_base_dir=base_dir,
                        # The content-type here doesn't matter, but we don't
                        # want to involve libmagic in a potentially-sensitive
                        # signing workflow.
                        content_type="application/octet-stream",
                    )
                    for file in files
                ],
            )

        @pydantic.field_validator("files", mode="after")
        @classmethod
        def validate_at_least_one_file(
            cls, files: dict[str, Path]
        ) -> dict[str, Path]:
            """Validate that artifact has at least one file."""
            if not files:
                raise ValueError("Expected at least one file")
            return files

    class SigningOutputArtifact(LocalArtifact[data_models.DebusineSigningOutput]):
        """Output of a Sign task."""

        _category = data_models.ArtifactCategory.SIGNING_OUTPUT

        @classmethod
        def create(
            cls,
            purpose: KeyPurpose,
            fingerprint: str,
            results: Iterable[data_models.SigningResult],
            files: Iterable[Path],
            base_dir: Path,
            binary_package_name: str | None = None,
        ) -> Self:
            """Return a new SigningOutputArtifact."""
            data = data_models.DebusineSigningOutput(
                purpose=purpose,
                fingerprint=fingerprint,
                results=list(results),
                binary_package_name=binary_package_name,
            )
            return cls._create(
                data=data,
                files=[
                    NewLocalFile(
                        file=file,
                        artifact_base_dir=base_dir,
                        # The content-type here doesn't matter, but we don't
                        # want to involve libmagic in a potentially-sensitive
                        # signing workflow.
                        content_type="application/octet-stream",
                    )
                    for file in files
                ],
            )

    class RepositoryIndex(LocalArtifact[data_models.DebianRepositoryIndex]):
        """An index file in a repository."""

        _category = data_models.ArtifactCategory.REPOSITORY_INDEX

        @classmethod
        def create(
            cls, *, file: Path, path: str, content_type: str | None = None
        ) -> Self:
            """Return a RepositoryIndex."""
            data = data_models.DebianRepositoryIndex(path=path)
            return cls._create(
                data=data,
                files=[
                    NewLocalFile(
                        file=file,
                        override_name=PurePath(path).name,
                        content_type=content_type,
                    )
                ],
            )