Merge pull request #785 from aboutcode-org/refactor-maven-pipeline

Refactor maven mining pipeline for git deployment

Author: keshav-space

minecode_pipelines/pipelines/mine_maven.py

@@ -20,42 +20,69 @@
 # ScanCode.io is a free software code scanning tool from nexB Inc. and others.
 # Visit https://github.com/aboutcode-org/scancode.io for support and download.
 
-from scanpipe.pipelines import Pipeline
 from scanpipe.pipes import federatedcode
 
 from minecode_pipelines import pipes
+from minecode_pipelines.pipelines import MineCodeBasePipeline
+from minecode_pipelines.pipelines import _mine_and_publish_packageurls
 from minecode_pipelines.pipes import maven
 
 
-class MineMaven(Pipeline):
-    """
-    Create DiscoveredPackages for packages found on maven:
-    - input: url of maven repo
-    - process index
-      - collect purls, grouped by package
-      - write to files
-      - publish to fetchcode
-      - loop
+class MineMaven(MineCodeBasePipeline):
+    """Mine PackageURLs from maven index and publish them to FederatedCode."""
 
-    """
+    pipeline_config_repo = "https://github.com/aboutcode-data/minecode-pipelines-config/"
+    checkpoint_path = "maven/checkpoints.json"
+    append_purls = True
 
     @classmethod
     def steps(cls):
         return (
             cls.check_federatedcode_eligibility,
-            cls.collect_packages_from_maven,
-            cls.delete_cloned_repos,
+            cls.create_federatedcode_working_dir,
+            cls.fetch_federation_config,
+            cls.fetch_checkpoint_and_maven_index,
+            cls.mine_and_publish_alpine_packageurls,
+            cls.delete_working_dir,
         )
 
-    def check_federatedcode_eligibility(self):
-        """
-        Check if the project fulfills the following criteria for
-        pushing the project result to FederatedCode.
-        """
-        federatedcode.check_federatedcode_configured_and_available(logger=self.log)
+    def fetch_checkpoint_and_maven_index(self):
+        self.checkpoint_config_repo = federatedcode.clone_repository(
+            repo_url=self.pipeline_config_repo,
+            clone_path=self.working_path / "minecode-pipelines-config",
+            logger=self.log,
+        )
+        checkpoint = pipes.get_checkpoint_from_file(
+            cloned_repo=self.checkpoint_config_repo,
+            path=self.checkpoint_path,
+        )
+
+        last_incremental = checkpoint.get("last_incremental")
+        self.log(f"last_incremental: {last_incremental}")
+        self.maven_nexus_collector = maven.MavenNexusCollector(last_incremental=last_incremental)
 
-    def collect_packages_from_maven(self):
-        self.repos = maven.collect_packages_from_maven(logger=self.log)
+    def mine_and_publish_alpine_packageurls(self):
+        _mine_and_publish_packageurls(
+            packageurls=self.maven_nexus_collector.get_packages(),
+            total_package_count=None,
+            data_cluster=self.data_cluster,
+            checked_out_repos=self.checked_out_repos,
+            working_path=self.working_path,
+            append_purls=self.append_purls,
+            commit_msg_func=self.commit_message,
+            logger=self.log,
+            checkpoint_func=self.save_check_point,
+        )
 
-    def delete_cloned_repos(self):
-        pipes.delete_cloned_repos(repos=self.repos, logger=self.log)
+    def save_check_point(self):
+        last_incremental = self.maven_nexus_collector.index_properties.get(
+            "nexus.index.last-incremental"
+        )
+        checkpoint = {"last_incremental": last_incremental}
+        self.log(f"Saving checkpoint: {checkpoint}")
+        pipes.update_checkpoints_in_github(
+            checkpoint=checkpoint,
+            cloned_repo=self.checkpoint_config_repo,
+            path=self.checkpoint_path,
+            logger=self.log,
+        )

minecode_pipelines/pipes/__init__.py

@@ -43,7 +43,7 @@ def get_checkpoint_from_file(cloned_repo, path):
     return checkpoint_data or {}
 
 
-def update_checkpoints_in_github(checkpoint, cloned_repo, path):
+def update_checkpoints_in_github(checkpoint, cloned_repo, path, logger):
     from scanpipe.pipes.federatedcode import commit_and_push_changes
 
     checkpoint_path = os.path.join(cloned_repo.working_dir, path)
@@ -53,6 +53,7 @@ def update_checkpoints_in_github(checkpoint, cloned_repo, path):
         repo=cloned_repo,
         files_to_commit=[checkpoint_path],
         commit_message=commit_message,
+        logger=logger,
     )
 
 

minecode_pipelines/pipes/maven.py

@@ -7,29 +7,24 @@
 # See https://aboutcode.org for more information about nexB OSS projects.
 #
 
+import gzip
+import io
+import os
 from collections import namedtuple
 from itertools import chain
 from shutil import rmtree
-import os
-import gzip
-import io
 
-from dateutil import tz
-from jawa.util.utf import decode_modified_utf8
 import arrow
 import javaproperties
-
-from aboutcode import hashid
+from dateutil import tz
+from jawa.util.utf import decode_modified_utf8
 from packagedcode.maven import build_filename
 from packagedcode.maven import build_url
 from packagedcode.maven import get_urls
 from packagedcode.models import PackageData
 from packageurl import PackageURL
-from scanpipe.pipes.fetch import fetch_http
-from scanpipe.pipes import federatedcode
 
-from minecode_pipelines import pipes
-from minecode_pipelines import VERSION
+
 from minecode_pipelines.pipes import java_stream
 
 TRACE = False
@@ -43,12 +38,6 @@
 MAVEN_INDEX_PROPERTIES_URL = (
     "https://repo1.maven.org/maven2/.index/nexus-maven-repository-index.properties"
 )
-MAVEN_CHECKPOINT_PATH = "maven/checkpoints.json"
-
-# We are testing and storing mined packageURLs in one single repo per ecosystem for now
-MINECODE_DATA_MAVEN_REPO = "https://github.com/aboutcode-data/minecode-data-maven-test"
-
-PACKAGE_BATCH_SIZE = 1000
 
 
 def is_worthy_artifact(artifact):
@@ -624,6 +613,8 @@ def __del__(self):
                 rmtree(download.directory)
 
     def _fetch_http(self, uri):
+        from scanpipe.pipes.fetch import fetch_http
+
         fetched = fetch_http(uri)
         self.downloads.append(fetched)
         return fetched
@@ -724,7 +715,7 @@ def _get_packages(self, content=None):
                 name=artifact_id,
                 version=version,
             )
-            yield current_purl, package
+            yield current_purl, [package.purl]
 
     def _get_packages_from_index_increments(self):
         for index_increment in self.index_increment_locations:
@@ -737,87 +728,3 @@ def get_packages(self):
         else:
             packages = self._get_packages(content=self.index_location)
         return packages
-
-
-def commit_message(commit_batch, total_commit_batch="many"):
-    from django.conf import settings
-
-    author_name = settings.FEDERATEDCODE_GIT_SERVICE_NAME
-    author_email = settings.FEDERATEDCODE_GIT_SERVICE_EMAIL
-    tool_name = "pkg:github/aboutcode-org/scancode.io"
-
-    return f"""\
-        Collect PackageURLs from Maven ({commit_batch}/{total_commit_batch})
-
-        Tool: {tool_name}@v{VERSION}
-        Reference: https://{settings.ALLOWED_HOSTS[0]}
-
-        Signed-off-by: {author_name} <{author_email}>
-        """
-
-
-def collect_packages_from_maven(files_per_commit=PACKAGE_BATCH_SIZE, logger=None):
-    # Clone data and config repo
-    data_repo = federatedcode.clone_repository(
-        repo_url=MINECODE_DATA_MAVEN_REPO,
-        logger=logger,
-    )
-    config_repo = federatedcode.clone_repository(
-        repo_url=pipes.MINECODE_PIPELINES_CONFIG_REPO,
-        logger=logger,
-    )
-    if logger:
-        logger(f"{MINECODE_DATA_MAVEN_REPO} repo cloned at: {data_repo.working_dir}")
-        logger(f"{pipes.MINECODE_PIPELINES_CONFIG_REPO} repo cloned at: {config_repo.working_dir}")
-
-    # get last_incremental to see if we can start from incrementals
-    checkpoint = pipes.get_checkpoint_from_file(cloned_repo=config_repo, path=MAVEN_CHECKPOINT_PATH)
-    last_incremental = checkpoint.get("last_incremental")
-    if logger:
-        logger(f"last_incremental: {last_incremental}")
-
-    # download and iterate through maven nexus index
-    maven_nexus_collector = MavenNexusCollector(last_incremental=last_incremental)
-    files_to_commit = []
-    commit_batch = 1
-    for current_purl, package in maven_nexus_collector.get_packages():
-        # write packageURL to file
-        package_base_dir = hashid.get_package_base_dir(purl=current_purl)
-        purl_file = pipes.write_packageurls_to_file(
-            repo=data_repo,
-            base_dir=package_base_dir,
-            packageurls=[package.purl],
-            append=True,
-        )
-        if purl_file not in files_to_commit:
-            files_to_commit.append(purl_file)
-
-        if len(files_to_commit) == files_per_commit:
-            federatedcode.commit_and_push_changes(
-                commit_message=commit_message(commit_batch),
-                repo=data_repo,
-                files_to_commit=files_to_commit,
-                logger=logger,
-            )
-            files_to_commit.clear()
-            commit_batch += 1
-
-    if files_to_commit:
-        federatedcode.commit_and_push_changes(
-            commit_message=commit_message(commit_batch),
-            repo=data_repo,
-            files_to_commit=files_to_commit,
-            logger=logger,
-        )
-
-    # update last_incremental so we can pick up from the proper place next time
-    last_incremental = maven_nexus_collector.index_properties.get("nexus.index.last-incremental")
-    checkpoint = {"last_incremental": last_incremental}
-    if logger:
-        logger(f"checkpoint: {checkpoint}")
-    pipes.update_checkpoints_in_github(
-        checkpoint=checkpoint, cloned_repo=config_repo, path=MAVEN_CHECKPOINT_PATH
-    )
-
-    repos_to_clean = [data_repo, config_repo]
-    return repos_to_clean

pyproject-minecode_pipelines.toml

@@ -4,7 +4,7 @@ build-backend = "flot.buildapi"
 
 [project]
 name = "minecode_pipelines"
-version = "0.0.1b30"
+version = "0.0.1b32"
 description = "A library for mining packageURLs and package metadata from ecosystem repositories."
 readme = "minecode_pipelines/README.rst"
 license = { text = "Apache-2.0" }