Refactor composer mining pipeline for git deployment (#787)

Signed-off-by: ziad hany <ziadhany2016@gmail.com>

Author: ziadhany

minecode_pipelines/__init__.py

@@ -8,4 +8,4 @@
 #
 
 
-VERSION = "0.0.1b25"
+VERSION = "0.0.1b42"

minecode_pipelines/miners/composer.py

@@ -9,6 +9,7 @@
 
 import json
 from minecode_pipelines.utils import get_temp_file
+from aboutcode.hashid import get_core_purl
 import requests
 from packageurl import PackageURL
 
@@ -67,7 +68,7 @@ def get_composer_purl(vendor, package):
         response = requests.get(url, timeout=10)
         response.raise_for_status()
     except requests.RequestException:
-        return purls
+        return None, purls
 
     data = response.json()
     packages = data.get("packages", {})
@@ -84,7 +85,11 @@ def get_composer_purl(vendor, package):
             )
             purls.append(purl.to_string())
 
-    return purls
+    base_purl = None
+    if purls:
+        first_purl = purls[0]
+        base_purl = get_core_purl(first_purl)
+    return base_purl, purls
 
 
 def load_composer_packages(packages_file):

minecode_pipelines/pipelines/mine_composer.py

@@ -20,63 +20,85 @@
 # 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.
 
-import os
-from scanpipe.pipelines import Pipeline
+from datetime import datetime
 from scanpipe.pipes import federatedcode
 
 from minecode_pipelines import pipes
-from minecode_pipelines.pipes import MINECODE_PIPELINES_CONFIG_REPO
-from minecode_pipelines.pipes.composer import mine_composer_packages
-from minecode_pipelines.pipes.composer import mine_and_publish_composer_purls
+from minecode_pipelines.pipelines import MineCodeBasePipeline
+from minecode_pipelines.pipes import composer
+from minecode_pipelines.pipes.composer import mine_composer_packages, PACKAGE_BATCH_SIZE
+from minecode_pipelines.pipelines import _mine_and_publish_packageurls
 
-MINECODE_COMPOSER_GIT_URL = os.environ.get(
-    "MINECODE_COMPOSER_GIT_URL", "https://github.com/aboutcode-data/minecode-data-composer-test"
-)
 
-
-class MineComposer(Pipeline):
+class MineComposer(MineCodeBasePipeline):
     """
-    Mine all packageURLs from a composer index and publish them to a FederatedCode repo.
+    Pipeline to mine Composer PHP packages and publish them to FederatedCode.
     """
 
+    pipeline_config_repo = "https://github.com/aboutcode-data/minecode-pipelines-config/"
+    checkpoint_path = "composer/checkpoints.json"
+    checkpoint_freq = 200
+
     @classmethod
     def steps(cls):
         return (
             cls.check_federatedcode_eligibility,
-            cls.clone_composer_repo,
-            cls.mine_and_publish_composer_purls,
+            cls.create_federatedcode_working_dir,
+            cls.fetch_checkpoint_and_start_index,
+            cls.fetch_federation_config,
+            cls.mine_and_publish_packageurls,
+            cls.delete_working_dir,
+        )
+
+    def fetch_checkpoint_and_start_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,
         )
 
-    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)
+        self.start_index = checkpoint.get("start_index", 0)
+        self.log(f"start_index: {self.start_index}")
 
-    def clone_composer_repo(self):
-        """
-        Clone the federatedcode composer url and return the Repo object
-        """
-        self.cloned_data_repo = federatedcode.clone_repository(MINECODE_COMPOSER_GIT_URL)
-        self.cloned_config_repo = federatedcode.clone_repository(MINECODE_PIPELINES_CONFIG_REPO)
+    def packages_count(self):
+        return len(self.composer_packages) if self.composer_packages else None
 
-    def mine_and_publish_composer_purls(self):
-        """
-        Mine Composer package names from Composer indexes and generate
-        package URLs (pURLs) for all mined Composer packages.
-        """
+    def mine_packageurls(self):
+        self.composer_packages = mine_composer_packages()
+        return composer.mine_composer_packageurls(
+            packages=self.composer_packages,
+            start_index=self.start_index,
+        )
 
-        composer_packages = mine_composer_packages()
-        mine_and_publish_composer_purls(
-            packages=composer_packages,
-            cloned_data_repo=self.cloned_data_repo,
-            cloned_config_repo=self.cloned_config_repo,
+    def mine_and_publish_packageurls(self):
+        """Mine and publish PackageURLs."""
+        _mine_and_publish_packageurls(
+            packageurls=self.mine_packageurls(),
+            total_package_count=self.packages_count(),
+            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,
+            checkpoint_freq=self.checkpoint_freq,
         )
 
-    def delete_cloned_repos(self):
-        pipes.delete_cloned_repos(
-            repos=[self.cloned_data_repo, self.cloned_config_repo],
+    def save_check_point(self):
+        checkpoint = {
+            "date": str(datetime.now()),
+            "start_index": self.start_index + self.checkpoint_freq * PACKAGE_BATCH_SIZE,
+        }
+
+        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/composer.py

@@ -20,22 +20,13 @@
 # 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 datetime import datetime
-from pathlib import Path
-from aboutcode import hashid
 from minecode_pipelines.miners.composer import get_composer_packages
 from minecode_pipelines.miners.composer import load_composer_packages
 from minecode_pipelines.miners.composer import get_composer_purl
-from minecode_pipelines.pipes import (
-    write_data_to_yaml_file,
-    get_checkpoint_from_file,
-    update_checkpoints_in_github,
-)
-from scanpipe.pipes.federatedcode import commit_and_push_changes
+
 from minecode_pipelines.utils import cycle_from_index, grouper
 
-PACKAGE_BATCH_SIZE = 1000
-COMPOSER_CHECKPOINT_PATH = "composer/checkpoints.json"
+PACKAGE_BATCH_SIZE = 100
 
 
 def mine_composer_packages():
@@ -44,55 +35,15 @@ def mine_composer_packages():
     return load_composer_packages(packages_file)
 
 
-def mine_and_publish_composer_purls(packages, cloned_data_repo, cloned_config_repo, logger):
-    """Mine Composer packages and publish their PURLs to a FederatedCode repository."""
-    composer_checkpoint = get_checkpoint_from_file(
-        cloned_repo=cloned_config_repo, path=COMPOSER_CHECKPOINT_PATH
-    )
-
-    start_index = composer_checkpoint.get("start_index", 0)
-
+def mine_composer_packageurls(packages, start_index):
+    """Mine Composer packages from Packagist"""
     packages_iter = cycle_from_index(packages, start_index)
-
     for batch_index, package_batch in enumerate(
         grouper(n=PACKAGE_BATCH_SIZE, iterable=packages_iter)
     ):
-        purl_files = []
-        purls = []
-
         for item in package_batch:
             if not item:
                 continue
 
             vendor, package = item
-
-            updated_purls = get_composer_purl(vendor=vendor, package=package)
-            if not updated_purls:
-                continue
-
-            base_purl = updated_purls[0]
-
-            purl_file_full_path = Path(
-                cloned_data_repo.working_dir
-            ) / hashid.get_package_purls_yml_file_path(base_purl)
-
-            write_data_to_yaml_file(path=purl_file_full_path, data=updated_purls)
-
-            purl_files.append(purl_file_full_path)
-            purls.append(str(base_purl))
-
-        if purls and purl_files:
-            logger(f"Committing packageURLs: {', '.join(purls)}")
-            commit_and_push_changes(
-                repo=cloned_data_repo, files_to_commit=purl_files, purls=purls, logger=logger
-            )
-
-        settings_data = {
-            "date": str(datetime.now()),
-            "start_index": start_index + (batch_index + 1) * PACKAGE_BATCH_SIZE,
-        }
-        update_checkpoints_in_github(
-            checkpoint=settings_data,
-            cloned_repo=cloned_config_repo,
-            path=COMPOSER_CHECKPOINT_PATH,
-        )
+            yield get_composer_purl(vendor=vendor, package=package)

minecode_pipelines/tests/pipes/test_composer.py

@@ -51,7 +51,8 @@ def test_generate_purls_from_composer(self, mock_get):
 
         all_purls = []
         for vendor, package in packages:
-            purls = get_composer_purl(vendor, package)
+            base_purl, purls = get_composer_purl(vendor, package)
+            assert str(base_purl) == "pkg:composer/monolog/monolog"
             all_purls.extend(purls)
 
         assert len(all_purls) == 85