Source code for swh.osv.luigi

# Copyright (C) 2026  The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information

# WARNING: do not import unnecessary things here to keep cli startup time under
# control

from typing import List

import luigi

from swh.graph.luigi.compressed_graph import LocalGraph

from .vuln_reports import DownloadVulnerabilities, UploadVulnerabilitiesSqliteToS3




class MapPackagesToVcs(luigi.Task):
    """Creates a directory of Parquet files that maps each package version
    to a release in a VCS repository"""

    local_graph_path = luigi.PathParameter()
    graph_name = luigi.StrParameter(default="graph")
    vulnerabilities_dir = luigi.PathParameter()



    def requires(self) -> dict[str, luigi.Task]:
        """Returns an instance of :class:`swh.graph.luigi.compressed_graph.LocalGraph`"""
        return {
            "graph": LocalGraph(local_graph_path=self.local_graph_path),
            "sqlite": DownloadVulnerabilities(
                local_graph_path=self.local_graph_path,
                graph_name=self.graph_name,
                vulnerabilities_dir=self.vulnerabilities_dir,
            ),
        }




    def output(self) -> luigi.LocalTarget:
        """Returns `vulnerabilities_dir / "all.sqlite"`"""
        return luigi.LocalTarget(self.vulnerabilities_dir / "packages_to_vcs")




    def run(self) -> None:
        """Downloads vulnerability reports to a temporary directory and imports
        them in a sqlite database"""
        from .shell import Rust

        # fmt: off
        Rust(
            "packages",
            "--graph",
            self.local_graph_path / self.graph_name,
            "--db",
            self.input()["sqlite"],
            "--out-dir",
            self.output(),
        ).run()


        # fmt: on




class UploadVulnerabilityDatasetToS3(luigi.WrapperTask):
    """Uploads all OSV artifacts to S3."""

    dataset_name = luigi.StrParameter()
    local_graph_path = luigi.PathParameter()
    graph_name = luigi.StrParameter(default="graph")
    vulnerabilities_dir = luigi.PathParameter()
    include_vcs_matches = luigi.BoolParameter(default=True)



    def requires(self) -> List[luigi.Task]:
        """Returns instances of all upload tasks."""
        from .cherrypicks import UploadCherrypicksToS3
        from .commit2vuln import UploadCommit2VulnBvToS3
        from .reachability import UploadSubgraphWccsToS3
        from .vuln2commit import UploadVuln2CommitParquetToS3

        kwargs = dict(
            dataset_name=self.dataset_name,
            local_graph_path=self.local_graph_path,
            graph_name=self.graph_name,
            vulnerabilities_dir=self.vulnerabilities_dir,
        )
        tasks: List[luigi.Task] = [
            UploadSubgraphWccsToS3(
                **kwargs, include_vcs_matches=self.include_vcs_matches
            ),
            UploadVuln2CommitParquetToS3(
                **kwargs,
                include_vcs_matches=self.include_vcs_matches,
                use_cherrypicks=False,
            ),
            UploadVuln2CommitParquetToS3(
                **kwargs,
                include_vcs_matches=self.include_vcs_matches,
                use_cherrypicks=True,
            ),
            UploadCommit2VulnBvToS3(
                **kwargs,
                include_vcs_matches=self.include_vcs_matches,
                use_cherrypicks=False,
            ),
            UploadCommit2VulnBvToS3(
                **kwargs,
                include_vcs_matches=self.include_vcs_matches,
                use_cherrypicks=True,
            ),
            UploadCherrypicksToS3(
                **kwargs, include_vcs_matches=self.include_vcs_matches
            ),
            UploadVulnerabilitiesSqliteToS3(**kwargs),
        ]
        return tasks