# 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.

"""
Task to build Debian packages with sbuild.

This task implements the PackageBuild generic task for its task_data:
https://freexian-team.pages.debian.net/debusine/reference/tasks/ontology-generic-tasks.html#task-packagebuild
"""

import email.utils
import os
import re
import shlex
import shutil
import tempfile
from pathlib import Path
from typing import cast, override

import pydantic
import yaml
from debian import debfile

import debusine.tasks.tags as ttags
import debusine.utils
import debusine.worker.tags as wtags
from debusine.artifacts import (
    BinaryPackage,
    DebugSymbols,
    PackageBuildLog,
    SigningInputArtifact,
    Upload,
)
from debusine.artifacts.models import (
    ArtifactCategory,
    DebianSourcePackage,
    DebianSystemImage,
    DebianSystemTarball,
    DoseDistCheck,
    WorkRequestResults,
)
from debusine.client.models import RelationType, RemoteArtifact
from debusine.tasks import (
    BaseTaskWithExecutor,
    ExtraRepositoryMixin,
    RunCommandTask,
)
from debusine.tasks.inputs import (
    DebusineFQDNInput,
    EnvironmentInput,
    ExtraRepositoriesInput,
    MultiInput,
    SingleInput,
    Stage,
)
from debusine.tasks.models import (
    BackendType,
    SbuildBuildComponent,
    SbuildData,
    SbuildDynamicData,
)
from debusine.tasks.sbuild_validator_mixin import SbuildValidatorMixin
from debusine.utils import read_dsc

_BUILD_ID_DEBUG_PREFIX = "./usr/lib/debug/.build-id/"

_BUILD_ID_DEBUG_PATH_RE = re.compile(
    r"^\./usr/lib/debug/\.build-id/"
    r"(?P<prefix>[0-9a-f]{2})/(?P<suffix>[0-9a-f]{38})\.debug$"
)


class DebugSymbolPathError(Exception):
    """Raised when a -dbgsym .deb contains a malformed .build-id path."""


class Sbuild(
    SbuildValidatorMixin,
    ExtraRepositoryMixin[SbuildData, SbuildDynamicData],
    RunCommandTask[SbuildData, SbuildDynamicData],
    BaseTaskWithExecutor[SbuildData, SbuildDynamicData],
):
    """Task implementing a Debian package build with sbuild."""

    TASK_VERSION = 1

    input_source_artifact = SingleInput(
        field="input.source_artifact",
        categories=[ArtifactCategory.SOURCE_PACKAGE],
        stage=Stage.PENDING,
        verbose_name="Source package",
        help_text="Source artifact to build",
    )
    input_extra_binary_artifacts = MultiInput(
        field="input.extra_binary_artifacts",
        verbose_name="Extra binary artifacts",
        help_text="Extra packages to pass to sbuild --extra-package option",
    )
    environment = EnvironmentInput(
        categories=[
            ArtifactCategory.SYSTEM_TARBALL,
            ArtifactCategory.SYSTEM_IMAGE,
        ],
        stage=Stage.PENDING,
        verbose_name="Build environment",
        help_text="Tarball or system image used for the build",
    )
    extra_repositories = ExtraRepositoriesInput(
        help_text="Extra repositories used for the build"
    )
    debusine_fqdn = DebusineFQDNInput(
        verbose_name="Debusine FQDN",
        help_text="Domain of this Debusine instance",
        hidden=True,
    )

    @override
    def post_init(self) -> None:
        super().post_init()

        self.chroots: list[str] | None = None
        self.builder = "sbuild"

        # dsc_file Path. Set by self.fetch_input()
        self._dsc_file: Path | None = None
        self._extra_packages: list[Path] = []

        self._uploaded_binary_packages: list[
            tuple[BinaryPackage, RemoteArtifact]
        ] = []

    @override
    def compute_scheduler_tags(
        self,
    ) -> tuple[ttags.TaskProvidedTags, wtags.TaskRequiredTags]:
        provided, required = super().compute_scheduler_tags()
        assert isinstance(self.input_source_artifact.data, DebianSourcePackage)

        provided.add(
            ttags.ProvenanceProvided.SYSTEM,
            {
                ttags.SOURCE_PACKAGE_PREFIX
                + self.input_source_artifact.data.name
            },
        )

        required.add(
            wtags.ProvenanceRequired.SYSTEM, {wtags.CAP_PREFIX + "sbuild"}
        )

        return provided, required

    @override
    def get_subject(self) -> str | None:
        assert isinstance(self.input_source_artifact.data, DebianSourcePackage)
        return self.input_source_artifact.data.name

    @override
    def get_configuration_context(self) -> str | None:
        assert isinstance(
            self.environment.data, (DebianSystemTarball, DebianSystemImage)
        )
        return self.environment.data.codename

    @override
    def compute_dynamic_data(self) -> SbuildDynamicData:
        r"""
        Resolve artifact lookups for this task.

        :subject: package name of ``input.source_artifact``
        :parameter_summary: package name of
          ``input.source_artifact``\_"version" ("architecture")
        :runtime_context:
          ``$build_components:$build_architecture:$host_architecture`` or
          None if ``build_options`` is provided
        :configuration_context: codename of ``environment``
        """
        input_source_artifact = self.input_source_artifact
        assert isinstance(input_source_artifact.data, DebianSourcePackage)

        environment = self.environment
        assert isinstance(
            environment.data, (DebianSystemTarball, DebianSystemImage)
        )

        if self._build_options():
            # Do not record stats for non-standard builds
            runtime_context = None
        else:
            build_components = "+".join(sorted(self.data.build_components))

            runtime_context = (
                f"{build_components}:"
                f"{self.data.build_architecture}:"
                f"{self._host_architecture}"
            )

        source_package_name = input_source_artifact.data.name
        source_version = input_source_artifact.data.version

        return SbuildDynamicData(
            environment_id=environment.artifact_id,
            extra_repositories=self.extra_repositories,
            input_source_artifact_id=input_source_artifact.artifact_id,
            input_extra_binary_artifacts_ids=(
                self.input_extra_binary_artifacts.resolved_ids()
            ),
            binnmu_maintainer=f"Debusine <noreply@{self.debusine_fqdn}>",
            subject=self.get_subject(),
            parameter_summary=f"{source_package_name}_{source_version} "
            f"({self._host_architecture})",
            runtime_context=runtime_context,
            configuration_context=self.get_configuration_context(),
        )

    @property
    def _host_architecture(self) -> str:
        # Currently defining host_architecture for Sbuild is not
        # supported. It uses "build_architecture" (or "all" or "source")
        match self.data.build_components:
            case [SbuildBuildComponent.ALL]:
                return "all"
            case [SbuildBuildComponent.SOURCE]:
                return "source"
            case _:
                return self.data.build_architecture

    @override
    @classmethod
    def get_provided_worker_tags(cls) -> set[str]:
        tags = super().get_provided_worker_tags()
        if debusine.utils.is_command_available("sbuild"):
            tags.add(wtags.CAP_PREFIX + "sbuild")
        return tags

    @override
    def fetch_input(self, destination: Path) -> bool:
        """Download the source artifact."""
        if not self.debusine:
            raise AssertionError("self.debusine not set")
        assert self.dynamic_data

        source_artifact_id = self.dynamic_data.input_source_artifact_id
        artifact = self.debusine.artifact_get(
            source_artifact_id, workspace=self.workspace_name
        )
        if artifact.category != ArtifactCategory.SOURCE_PACKAGE:
            self.append_to_log_file(
                "fetch_input.log",
                [
                    f"input.source_artifact points to a "
                    f"{artifact.category}, not the expected "
                    f"{ArtifactCategory.SOURCE_PACKAGE}."
                ],
            )
            return False
        for file in artifact.files:
            if file.endswith(".dsc"):
                self._dsc_file = destination / file
                break
        else:
            raise AssertionError("No .dsc file found in source package.")
        self.fetch_artifact(source_artifact_id, destination)

        for artifact_id in self.dynamic_data.input_extra_binary_artifacts_ids:
            artifact = self.debusine.artifact_get(
                artifact_id, workspace=self.workspace_name
            )
            if artifact.category in (
                ArtifactCategory.BINARY_PACKAGE,
                ArtifactCategory.UPLOAD,
            ):
                for file in artifact.files:
                    if file.endswith(".deb"):
                        self._extra_packages.append(destination / file)
            else:
                self.append_to_log_file(
                    "fetch_input.log",
                    [
                        f"input.extra_binary_artifacts includes artifact "
                        f"{artifact_id}, a {artifact.category}, not the "
                        f"expected {ArtifactCategory.BINARY_PACKAGE}, or "
                        f"{ArtifactCategory.UPLOAD}."
                    ],
                )
                return False
            self.fetch_artifact(artifact_id, destination)

        self._prepare_executor()

        self.write_extra_repository_config(
            codename=self._environment_codename(), destination=destination
        )

        return True

    def _environment_codename(self) -> str:
        """Return the codename of the environment we're building in."""
        if self.executor is None:
            raise ValueError(
                "self.executor not set, call _prepare_executor() first"
            )

        return cast(str, self.executor.system_image.data["codename"])

    @override
    def _cmdline(self) -> list[str]:
        """
        Build the sbuild command line.

        Use self.data and self._dsc_file.
        """
        cmd = [
            self.builder,
            "--purge-deps=never",
            # Use a higher-level workflow instead to run both sbuild and
            # lintian, such as debian_pipeline.
            "--no-run-lintian",
        ]
        if "any" in self.data.build_components:
            cmd.append("--arch-any")
        else:
            cmd.append("--no-arch-any")
        if "all" in self.data.build_components:
            cmd.append("--arch-all")
        else:
            cmd.append("--no-arch-all")
        if "source" in self.data.build_components:
            cmd.append("--source")
        else:
            cmd.append("--no-source")

        # TODO: This should probably be host_architecture, if it isn't "all"
        # or "source".
        cmd.append("--arch=" + self.data.build_architecture)
        cmd += self._cmdline_backend()
        cmd += self._cmdline_extra_repository()
        cmd += self._cmdline_binnmu()

        if self.data.build_profiles:
            cmd.append(f"--profiles={','.join(self.data.build_profiles)}")

        if self.data.build_dep_resolver:
            cmd.append(f"--build-dep-resolver={self.data.build_dep_resolver}")
        if self.data.aspcud_criteria:
            cmd.append(f"--aspcud-criteria={self.data.aspcud_criteria}")
        if self.data.resolve_alternatives:
            cmd.append("--resolve-alternatives")

        # BD-Uninstallable
        cmd.append("--bd-uninstallable-explainer=dose3")

        for package in self._extra_packages:
            cmd.append(f"--extra-package={package}")

        cmd.append(str(self._dsc_file))

        return cmd

    def _cmdline_backend(self) -> list[str]:
        """Generate command line arguments for the backend and environment."""
        distribution = self._environment_codename()
        args: list[str] = [f"--dist={distribution}"]

        assert self.executor
        if self.backend == BackendType.UNSHARE:
            args += [
                "--chroot-mode=unshare",
                f"--chroot={self.executor.image_name()}",
                # Remove any dangling resolv.conf symlink (from
                # systemd-resolved installed in the environment, #1071736)
                "--chroot-setup-commands=rm -f /etc/resolv.conf",
            ]
        else:
            virt_server = self.executor.autopkgtest_virt_server()
            args += [
                "--chroot-mode=autopkgtest",
                f"--autopkgtest-virt-server=autopkgtest-virt-{virt_server}",
            ] + [
                f"--autopkgtest-virt-server-opt={opt.replace('%', '%%')}"
                for opt in self.executor.autopkgtest_virt_args()
            ]
        return args

    def _cmdline_extra_repository(self) -> list[str]:
        """Generate the command line arguments for extra_repositories."""
        args: list[str] = []

        # sbuild has no native deb822 support, yet: #1089735
        for path in self.extra_repository_sources:
            inner_path = f"/etc/apt/sources.list.d/{path.name}"
            inner_command = f"cat > {shlex.quote(inner_path)}"
            args.append(
                f"--pre-build-commands=cat {shlex.quote(str(path))} | "
                f"%SBUILD_CHROOT_EXEC sh -c {shlex.quote(inner_command)}"
            )

        # apt < 2.3.10 has no support for keys embedded in Signed-By
        for path in self.extra_repository_keys:
            inner_path = f"/etc/apt/keyrings/{path.name}"
            inner_command = (
                f"mkdir -p /etc/apt/keyrings && cat > {shlex.quote(inner_path)}"
            )
            args.append(
                f"--pre-build-commands=cat {shlex.quote(str(path))} | "
                f"%SBUILD_CHROOT_EXEC sh -c {shlex.quote(inner_command)}"
            )

        return args

    def _cmdline_binnmu(self) -> list[str]:
        """Generate the command line arguments for binnmu."""
        if self.data.binnmu is None:
            return []

        args: list[str] = [
            f"--make-binNMU={self.data.binnmu.changelog}",
            f"--append-to-version={self.data.binnmu.suffix}",
            "--binNMU=0",
        ]
        if self.data.binnmu.timestamp is not None:
            rfc5322date = email.utils.format_datetime(
                self.data.binnmu.timestamp
            )
            args.append(f"--binNMU-timestamp={rfc5322date}")
        maintainer: str | None
        if self.data.binnmu.maintainer is not None:
            maintainer = str(self.data.binnmu.maintainer)
        else:
            assert self.dynamic_data
            maintainer = self.dynamic_data.binnmu_maintainer
        args.append(f"--maintainer={maintainer}")
        return args

    def _build_options(self) -> list[str]:
        return [
            profile
            for profile in self.data.build_profiles or []
            if profile in ("nocheck", "nodoc")
        ]

    @override
    def _cmd_env(self) -> dict[str, str]:
        """Set DEB_BUILD_OPTIONS for build profiles."""
        env = super()._cmd_env()
        if build_options := self._build_options():
            env["DEB_BUILD_OPTIONS"] = " ".join(
                os.environ.get("DEB_BUILD_OPTIONS", "").split() + build_options
            )
        return env

    @override
    def task_result(
        self, returncode: int | None, execute_directory: Path
    ) -> WorkRequestResults:
        """
        Check whether the task succeeded.

        `sbuild` returns 0 for "Status: skipped" (i.e. host architecture not
        supported by source package); we consider that a failure.
        """
        super_result = super().task_result(returncode, execute_directory)
        if super_result != WorkRequestResults.SUCCESS:
            return super_result

        build_log_path = debusine.utils.find_file_suffixes(
            execute_directory, [".build"]
        )
        if build_log_path is None:
            return WorkRequestResults.ERROR

        status = ""
        # Reading the whole build log is inefficient, but shouldn't usually
        # be too bad in practice.
        with open(build_log_path, errors="replace") as f:
            for line in f:
                if line.startswith("Status: "):
                    status = line[len("Status: ") :].rstrip("\n")
        if status == "successful":
            return WorkRequestResults.SUCCESS
        return WorkRequestResults.FAILURE

    @staticmethod
    def _extract_dose3_explanation(
        build_log_path: Path,
    ) -> DoseDistCheck | None:
        """Isolate and parse dose3 output in BD-Uninstallable scenario."""
        output = ''
        with open(build_log_path, errors='replace') as f:
            # find start of dose-debcheck output
            while (line := f.readline()) != '':
                if line == '(I)Dose_applications: Solving...\n':
                    break

            # exit if no/invalid output
            if not (line := f.readline()).startswith('output-version'):
                return None
            output += line

            # grab until next section
            while (line := f.readline()) != '':
                if line.startswith('+----'):
                    break
                output += line

            # convert & validate to Pydantic structure
            # Use yaml.CBaseLoad to work-around malformed yaml
            # https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=834059#22
            # https://gitlab.com/irill/dose3/-/issues/18
            parsed_dict = yaml.load(output, yaml.CBaseLoader)
            try:
                return DoseDistCheck.model_validate(parsed_dict)
            except pydantic.ValidationError:
                return None

    def _upload_package_build_log(
        self,
        build_directory: Path,
        source: str,
        version: str,
        architecture: str,
        execution_result: WorkRequestResults,
    ) -> RemoteArtifact | None:
        if not self.debusine:
            raise AssertionError("self.debusine not set")

        build_log_path = debusine.utils.find_file_suffixes(
            build_directory, [".build"]
        )
        if build_log_path is None:
            return None

        explanation = None
        if execution_result != WorkRequestResults.SUCCESS:
            # BD-Uninstallable: look for the dose3 output
            explanation = self._extract_dose3_explanation(build_log_path)

        package_build_log = PackageBuildLog.create(
            source=source,
            version=version,
            architecture=architecture,
            file=build_log_path,
            bd_uninstallable=explanation,
        )

        return self.debusine.upload_artifact(
            package_build_log,
            workspace=self.workspace_name,
            work_request=self.work_request_id,
        )

    def _upload_binary_upload(
        self, build_directory: Path
    ) -> RemoteArtifact | None:
        if not self.debusine:
            raise AssertionError("self.debusine not set")

        changes_path = debusine.utils.find_file_suffixes(
            build_directory, [".changes"]
        )
        if changes_path is None:
            return None

        artifact_binary_upload = Upload.create(
            changes_file=changes_path,
        )
        return self.debusine.upload_artifact(
            artifact_binary_upload,
            workspace=self.workspace_name,
            work_request=self.work_request_id,
        )

    def _upload_signing_input(
        self, build_directory: Path
    ) -> RemoteArtifact | None:
        if not self.debusine:
            raise AssertionError("self.debusine not set")

        changes_path = debusine.utils.find_file_suffixes(
            build_directory, [".changes"]
        )
        if changes_path is None:
            return None

        artifact_signing_input = SigningInputArtifact.create(
            [changes_path], build_directory
        )
        return self.debusine.upload_artifact(
            artifact_signing_input,
            workspace=self.workspace_name,
            work_request=self.work_request_id,
        )

    def _create_binary_package_local_artifacts(
        self,
        build_directory: Path,
        suffixes: list[str],
    ) -> list[BinaryPackage]:
        deb_paths = debusine.utils.find_files_suffixes(
            build_directory, suffixes
        )

        artifacts: list[BinaryPackage] = []
        for deb_path in deb_paths:
            artifacts.append(BinaryPackage.create(file=deb_path))  # noqa: PERF401
        return artifacts

    def _upload_binary_packages(
        self,
        build_directory: Path,
    ) -> list[RemoteArtifact]:
        r"""Upload \*.deb and \*.udeb files."""
        if not self.debusine:
            raise AssertionError("self.debusine not set")

        host_arch = self._host_architecture

        packages = []

        if "any" in self.data.build_components:
            prefix = "_" + host_arch
            packages.extend(
                self._create_binary_package_local_artifacts(
                    build_directory,
                    [prefix + ".deb", prefix + ".udeb"],
                )
            )

        if "all" in self.data.build_components:
            prefix = "_all"
            packages.extend(
                self._create_binary_package_local_artifacts(
                    build_directory,
                    [prefix + ".deb", prefix + ".udeb"],
                )
            )

        remote_artifacts: list[RemoteArtifact] = []
        for package in packages:
            remote = self.debusine.upload_artifact(
                package,
                workspace=self.workspace_name,
                work_request=self.work_request_id,
            )
            remote_artifacts.append(remote)
            self._uploaded_binary_packages.append((package, remote))

        return remote_artifacts

    @staticmethod
    def _find_matching_dbgsym(directory: Path, binary_name: str) -> Path | None:
        if not binary_name.endswith(".deb"):
            return None

        stem = binary_name[: -len(".deb")]
        if "_" not in stem:
            return None

        package, rest = stem.split("_", 1)
        for ext in (".deb", ".ddeb"):
            candidate = directory / f"{package}-dbgsym_{rest}{ext}"
            if candidate.is_file():
                return candidate
        return None

    @classmethod
    def _extract_build_id_files(
        cls, *, dbgsym_path: Path, destination: Path
    ) -> dict[str, Path]:
        build_id_files: dict[str, Path] = {}
        with debfile.DebFile(dbgsym_path) as pkg:
            tar = pkg.data.tgz()
            for member in tar.getmembers():
                if not member.isreg():
                    continue
                if not member.name.startswith(_BUILD_ID_DEBUG_PREFIX):
                    continue
                if (
                    match := _BUILD_ID_DEBUG_PATH_RE.match(member.name)
                ) is None:
                    raise DebugSymbolPathError(
                        f"{dbgsym_path.name}: malformed .build-id "
                        f"path {member.name!r}; expected "
                        f"./usr/lib/debug/.build-id/<XX>/<YYYY>.debug"
                    )
                build_id = match.group("prefix") + match.group("suffix")
                out_path = destination / f"{build_id}.debug"
                src = tar.extractfile(member)
                if src is None:  # pragma: no cover - defensive
                    continue
                try:
                    with out_path.open("wb") as dst:
                        shutil.copyfileobj(src, dst)
                finally:
                    src.close()
                build_id_files[build_id] = out_path
        return build_id_files

    def _upload_debug_symbols(
        self,
        build_directory: Path,
    ) -> list[tuple[RemoteArtifact, RemoteArtifact]]:
        """
        Upload one ``debian:debug-symbols`` artifact per -dbgsym .deb.

        Iterates over binaries already uploaded by
        :meth:`_upload_binary_packages`.

        :return: ``(debug_remote, parent_binary_remote)`` pairs for
            :meth:`_create_debug_symbols_relations` — debug-symbols artifact
            first, parent binary package second.
        """
        if not self.debusine:
            raise AssertionError("self.debusine not set")

        pairs: list[tuple[RemoteArtifact, RemoteArtifact]] = []

        for binary_local, binary_remote in self._uploaded_binary_packages:
            for binary_name in binary_local.files:
                dbgsym_path = self._find_matching_dbgsym(
                    build_directory, binary_name
                )
                if dbgsym_path is None:
                    continue

                with tempfile.TemporaryDirectory(
                    prefix="debusine-dbgsym-"
                ) as raw_tmp:
                    build_id_files = self._extract_build_id_files(
                        dbgsym_path=dbgsym_path,
                        destination=Path(raw_tmp),
                    )
                    if not build_id_files:
                        continue

                    local_debug = DebugSymbols.create(
                        build_id_files=build_id_files
                    )
                    remote_debug = self.debusine.upload_artifact(
                        local_debug,
                        workspace=self.workspace_name,
                        work_request=self.work_request_id,
                    )

                pairs.append((remote_debug, binary_remote))

        return pairs

    def _create_debug_symbols_relations(
        self,
        debug_symbol_pairs: list[tuple[RemoteArtifact, RemoteArtifact]],
    ) -> None:
        """Create RELATES_TO from each debug-symbols artifact to its binary."""
        if not self.debusine:
            raise AssertionError("self.debusine not set")

        for debug_remote, binary_remote in debug_symbol_pairs:
            self.debusine.relation_create(
                debug_remote.id,
                binary_remote.id,
                RelationType.RELATES_TO,
            )

    def _create_remote_binary_packages_relations(
        self,
        remote_build_log: RemoteArtifact | None,
        remote_binary_upload: RemoteArtifact | None,
        remote_binary_packages: list[RemoteArtifact],
        remote_signing_input: RemoteArtifact | None,
    ) -> None:
        if not self.debusine:
            raise AssertionError("self.debusine not set")

        for remote_binary_package in remote_binary_packages:
            for source_artifact_id in self._source_artifacts_ids:
                self.debusine.relation_create(
                    remote_binary_package.id,
                    source_artifact_id,
                    RelationType.BUILT_USING,
                )

            if remote_build_log is not None:
                self.debusine.relation_create(
                    remote_build_log.id,
                    remote_binary_package.id,
                    RelationType.RELATES_TO,
                )

            if remote_binary_upload is not None:
                self.debusine.relation_create(
                    remote_binary_upload.id,
                    remote_binary_package.id,
                    RelationType.EXTENDS,
                )
                self.debusine.relation_create(
                    remote_binary_upload.id,
                    remote_binary_package.id,
                    RelationType.RELATES_TO,
                )
                if remote_build_log is not None:
                    self.debusine.relation_create(
                        remote_build_log.id,
                        remote_binary_upload.id,
                        RelationType.RELATES_TO,
                    )

        if (
            remote_binary_upload is not None
            and remote_signing_input is not None
        ):
            self.debusine.relation_create(
                remote_signing_input.id,
                remote_binary_upload.id,
                RelationType.RELATES_TO,
            )

    @override
    def configure_for_execution(
        self,
        download_directory: Path,  # noqa: U100
    ) -> bool:
        """
        Configure Task: set variables needed for the build() step.

        Return True if configuration worked, False, if there was a problem.
        """
        if self._dsc_file is None or not self._dsc_file.exists():
            self.append_to_log_file(
                "configure_for_execution.log",
                ["Input source package not found."],
            )
            return False

        return True

    @override
    def upload_artifacts(
        self, directory: Path, *, execution_result: WorkRequestResults
    ) -> None:
        """Upload artifacts for the task.

        :param directory: directory containing the files that
          will be uploaded.
        :param execution_result: if this is not ``WorkRequestResults.SUCCESS``,
          skip uploading ``.changes`` and ``*.deb``/``*.udeb``.
        """
        if not self.debusine:
            raise AssertionError("self.debusine not set")

        if self._dsc_file is None:
            return
        # ValueError should be impossible here, since it would already have
        # been caught by
        # SbuildValidatorMixin.check_directory_for_consistency_errors.  If
        # it somehow happens anyway, just let it propagate up.
        dsc = read_dsc(self._dsc_file)

        # Upload the .build file (PackageBuildLog)
        remote_build_log = self._upload_package_build_log(
            directory,
            dsc["source"],
            dsc["version"],
            self._host_architecture,
            execution_result,
        )

        if remote_build_log is not None:
            for source_artifact_id in self._source_artifacts_ids:
                self.debusine.relation_create(
                    remote_build_log.id,
                    source_artifact_id,
                    RelationType.RELATES_TO,
                )

        if execution_result == WorkRequestResults.SUCCESS:
            # Upload the *.deb/*.udeb files (BinaryPackages)
            remote_binary_packages = self._upload_binary_packages(directory)

            # Upload debian:debug-symbols artifacts extracted from
            # -dbgsym .debs produced by sbuild
            debug_symbol_pairs = self._upload_debug_symbols(directory)

            # Upload the .changes and the rest of the files
            remote_binary_changes = self._upload_binary_upload(directory)

            # Upload the .changes on its own as signing input
            remote_signing_input = self._upload_signing_input(directory)

            # Create the relations
            self._create_remote_binary_packages_relations(
                remote_build_log,
                remote_binary_changes,
                remote_binary_packages,
                remote_signing_input,
            )
            self._create_debug_symbols_relations(debug_symbol_pairs)
