create a new pipe for compliancing

Signed-off-by: NucleonGodX <[email protected]>

Author: NucleonGodX

scanpipe/pipelines/fetch_scores.py

@@ -25,7 +25,7 @@
 
 from scanpipe.models import DiscoveredPackageScore
 from scanpipe.pipelines import Pipeline
-from scanpipe.pipes.compliance_thresholds import get_project_scorecard_thresholds
+from scanpipe.pipes import scorecard_compliance
 
 
 class FetchScores(Pipeline):
@@ -50,6 +50,7 @@ def steps(cls):
         return (
             cls.check_scorecode_service_availability,
             cls.fetch_packages_scorecode_info,
+            cls.evaluate_compliance_alerts,
         )
 
     def check_scorecode_service_availability(self):
@@ -59,26 +60,13 @@ def check_scorecode_service_availability(self):
 
     def fetch_packages_scorecode_info(self):
         """Fetch ScoreCode information for each of the project's discovered packages."""
-        scorecard_policy = get_project_scorecard_thresholds(self.project)
-        worst_alert = None
-
         for package in self.project.discoveredpackages.all():
             if scorecard_data := ossf_scorecard.fetch_scorecard_info(package=package):
                 DiscoveredPackageScore.create_from_package_and_scorecard(
                     scorecard_data=scorecard_data,
                     package=package,
                 )
 
-                if scorecard_policy and scorecard_data.score is not None:
-                    try:
-                        score = float(scorecard_data.score)
-                        alert = scorecard_policy.get_alert_for_score(score)
-                    except Exception:
-                        alert = "error"
-
-                    order = {"ok": 0, "warning": 1, "error": 2}
-                    if worst_alert is None or order[alert] > order.get(worst_alert, -1):
-                        worst_alert = alert
-
-        if worst_alert is not None:
-            self.project.update_extra_data({"scorecard_compliance_alert": worst_alert})
+    def evaluate_compliance_alerts(self):
+        """Evaluate scorecard compliance alerts for the project."""
+        scorecard_compliance.evaluate_scorecard_compliance(self.project)

scanpipe/pipes/scorecard_compliance.py

@@ -0,0 +1,62 @@
+# 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.
+#
+# Data Generated with ScanCode.io is provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+# ScanCode.io should be considered or used as legal advice. Consult an Attorney
+# for any legal advice.
+#
+# 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.pipes.compliance_thresholds import get_project_scorecard_thresholds
+
+
+def evaluate_scorecard_compliance(project):
+    """
+    Evaluate scorecard compliance for all discovered packages in the project.
+    
+    This function checks OpenSSF Scorecard scores against project-defined
+    thresholds and determines the worst compliance alert level across all packages.
+    Updates the project's extra_data with the overall compliance status.
+    """
+    scorecard_policy = get_project_scorecard_thresholds(project)
+    if not scorecard_policy:
+        return 
+
+    worst_alert = None
+    packages_with_scores = project.discoveredpackages.filter(
+        scores__scoring_tool="ossf-scorecard"
+    ).distinct()
+
+    for package in packages_with_scores:
+        latest_score = package.scores.filter(
+            scoring_tool="ossf-scorecard"
+        ).order_by("-score_date").first()
+
+        if not latest_score or latest_score.score is None:
+            continue
+
+        try:
+            score = float(latest_score.score)
+            alert = scorecard_policy.get_alert_for_score(score)
+        except Exception:
+            alert = "error"
+
+        order = {"ok": 0, "warning": 1, "error": 2}
+        if worst_alert is None or order[alert] > order.get(worst_alert, -1):
+            worst_alert = alert
+
+    if worst_alert is not None:
+        project.update_extra_data({"scorecard_compliance_alert": worst_alert})