Add npm health metrics orchestration pipeline

docs/built-in-pipelines.rst

@@ -282,3 +282,36 @@ Fetch Scores (addon)
 .. autoclass:: scanpipe.pipelines.fetch_scores.FetchScores()
     :members:
     :member-order: bysource
+
+
+.. _pipeline_npm_health:
+
+NPM Health (addon)
+------------------
+
+The ``npm_health`` pipeline analyzes one exact versioned npm ``Project.purl``.
+It retrieves npm registry metadata, records repository and package tarball
+locations, normalizes health signals, computes a weighted score from 0 to 100,
+and stores a reusable snapshot in ``Project.extra_data["npm_health"]``.
+
+.. autoclass:: scanpipe.pipelines.npm_health.NpmHealth()
+    :members:
+    :member-order: bysource
+
+
+NPM Health configuration
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+Fresh snapshots are reused for 90 days by default. Override the age with
+``npm_health_cache_max_age_days`` in project pipeline settings.
+
+Optional GrimoireLab or other project-health metrics can be connected through
+``npm_health_metrics_command``. The command is parsed to an argument list and
+executed without a shell. It may use ``{purl}``, ``{repository_url}``,
+``{tarball_url}``, and ``{output}`` placeholders. The collector must write JSON
+metrics to ``{output}``, either directly or under a top-level ``metrics`` key.
+Metric values may be fractions from 0 to 1 or percentages from 0 to 100.
+
+The optional ``npm_health_metric_weights`` mapping customizes scoring weights.
+Only metrics present in the current analysis participate in the denominator.
+The pipeline also writes a timestamped ``npm-health-*.json`` project output.

pyproject.toml

@@ -158,6 +158,7 @@ collect_symbols_pygments = "scanpipe.pipelines.collect_symbols_pygments:CollectS
 collect_symbols_tree_sitter = "scanpipe.pipelines.collect_symbols_tree_sitter:CollectSymbolsTreeSitter"
 enrich_with_purldb = "scanpipe.pipelines.enrich_with_purldb:EnrichWithPurlDB"
 fetch_scores = "scanpipe.pipelines.fetch_scores:FetchScores"
+npm_health = "scanpipe.pipelines.npm_health:NpmHealth"
 find_vulnerabilities = "scanpipe.pipelines.find_vulnerabilities:FindVulnerabilities"
 inspect_elf_binaries = "scanpipe.pipelines.inspect_elf_binaries:InspectELFBinaries"
 inspect_packages = "scanpipe.pipelines.inspect_packages:InspectPackages"

scanpipe/pipelines/npm_health.py

@@ -0,0 +1,124 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# http://nexb.com and https://github.com/aboutcode-org/scancode.io
+# The ScanCode.io software is licensed under the Apache License version 2.0.
+# Data generated with ScanCode.io is provided as-is without warranties.
+# ScanCode is a trademark of nexB Inc.
+#
+# You may not use this software except in compliance with the License.
+# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+
+"""Collect and score npm package health information."""
+
+from scanpipe.pipelines import Pipeline
+from scanpipe.pipes import npm_health
+
+
+class NpmHealth(Pipeline):
+    """Collect reusable npm package health metrics for one project PURL."""
+
+    download_inputs = False
+    is_addon = True
+    results_url = "/project/{slug}/"
+
+
+    def validate_project_purl(self):
+        """Validate and parse the project's versioned npm PURL."""
+        self.package = npm_health.validate_npm_package_url(self.project.purl)
+
+
+    def load_cached_snapshot(self):
+        """Reuse a fresh project snapshot when one is already available."""
+        self.cached_snapshot = None
+        snapshot = npm_health.get_cached_snapshot(self.project)
+        max_age = int(
+            self.env.get(
+                "npm_health_cache_max_age_days",
+                npm_health.DEFAULT_CACHE_MAX_AGE_DAYS,
+            )
+        )
+        if snapshot and not npm_health.is_stale(snapshot, max_age_days=max_age):
+            self.cached_snapshot = snapshot
+            self.snapshot = snapshot
+            self.log("Using fresh cached npm-health analysis.")
+
+
+    def fetch_package_metadata(self):
+        """Fetch exact npm registry metadata unless a fresh cache is used."""
+        if self.cached_snapshot:
+            return
+        self.metadata = npm_health.fetch_registry_metadata(self.package)
+
+
+    def collect_package_metrics(self):
+        """Collect built-in registry signals and optional external metrics."""
+        if self.cached_snapshot:
+            return
+
+        baseline = npm_health.collect_registry_metrics(self.metadata)
+        external = {}
+        command_template = self.env.get("npm_health_metrics_command")
+        if command_template:
+            output = self.project.tmp_path / "npm-health-external-metrics.json"
+            external = npm_health.collect_external_metrics(
+                command_template=command_template,
+                purl=self.project.purl,
+                metadata=self.metadata,
+                output=output,
+                cwd=self.project.tmp_path,
+            )
+        self.metrics = npm_health.merge_metrics(baseline, external)
+
+
+    def compute_package_health_score(self):
+        """Compute the normalized package health score."""
+        if self.cached_snapshot:
+            return
+        weights = self.env.get("npm_health_metric_weights")
+        self.score = npm_health.compute_health_score(self.metrics, weights=weights)
+
+
+    def build_result_snapshot(self):
+        """Build the reusable snapshot stored on the project."""
+        if self.cached_snapshot:
+            return
+        self.snapshot = npm_health.build_snapshot(
+            purl=self.project.purl,
+            metadata=self.metadata,
+            metrics=self.metrics,
+            score=self.score,
+        )
+
+
+    def persist_results(self):
+        """Persist fresh results to extra_data and a portable JSON output."""
+        if self.cached_snapshot:
+            return
+
+        npm_health.cache_snapshot(self.project, self.snapshot)
+        output = npm_health.write_snapshot(self.project, self.snapshot)
+        self.project.add_info(
+            model="npm_health",
+            description=(
+                f"npm-health score: {self.snapshot['score']} "
+                f"({self.snapshot['classification']})"
+            ),
+            details={"output": output.name},
+        )
+
+
+    @classmethod
+    def steps(cls):
+        return (
+            cls.validate_project_purl,
+            cls.load_cached_snapshot,
+            cls.fetch_package_metadata,
+            cls.collect_package_metrics,
+            cls.compute_package_health_score,
+            cls.build_result_snapshot,
+            cls.persist_results,
+        )