[primer] Extract comparison logic into PylintComparator class

Move cross-referencing and JSON loading from CompareCommand into a
dedicated PylintComparator class with an abstract Comparator base,
enabling future comparator implementations (e.g. AstroidComparator).

Closes #6984

Author: Pierre-Sassoulas

pylint/testutils/_primer/comparator.py

@@ -0,0 +1,70 @@
+# Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
+# For details: https://github.com/pylint-dev/pylint/blob/main/LICENSE
+# Copyright (c) https://github.com/pylint-dev/pylint/blob/main/CONTRIBUTORS.txt
+
+from __future__ import annotations
+
+import abc
+import json
+from collections.abc import Iterator
+from pathlib import Path
+
+from pylint.reporters.json_reporter import OldJsonExport
+from pylint.testutils._primer.primer_command import (
+    PackageData,
+    PackageMessages,
+)
+
+
+class Comparator(abc.ABC):
+    """Abstract base class for primer comparators."""
+
+    missing_messages: PackageMessages
+    new_messages: PackageMessages
+
+    @abc.abstractmethod
+    def __iter__(
+        self,
+    ) -> Iterator[tuple[str, PackageData, PackageData]]:
+        """Yield (package, missing_messages, new_messages) for packages with
+        changes.
+        """
+
+
+class PylintComparator(Comparator):
+    """Compare two primer runs and compute missing/new messages per package."""
+
+    def __init__(
+        self,
+        main_data: PackageMessages,
+        pr_data: PackageMessages,
+    ) -> None:
+        self.missing_messages: PackageMessages = {}
+        for package, data in main_data.items():
+            package_missing_messages: list[OldJsonExport] = []
+            for message in data["messages"]:
+                try:
+                    pr_data[package]["messages"].remove(message)
+                except ValueError:
+                    package_missing_messages.append(message)
+            self.missing_messages[package] = PackageData(
+                commit=pr_data[package]["commit"],
+                messages=package_missing_messages,
+            )
+        self.new_messages = pr_data
+
+    def __iter__(
+        self,
+    ) -> Iterator[tuple[str, PackageData, PackageData]]:
+        for package, missing in self.missing_messages.items():
+            new = self.new_messages[package]
+            if not missing["messages"] and not new["messages"]:
+                print(f"PRIMER: No changes in {package}.")
+                continue
+            yield package, missing, new
+
+    @staticmethod
+    def load_json(file_path: Path | str) -> PackageMessages:
+        with open(file_path, encoding="utf-8") as f:
+            result: PackageMessages = json.load(f)
+        return result

pylint/testutils/_primer/primer_compare_command.py

@@ -3,13 +3,11 @@
 # Copyright (c) https://github.com/pylint-dev/pylint/blob/main/CONTRIBUTORS.txt
 from __future__ import annotations
 
-import json
-from pathlib import Path, PurePosixPath
+from pathlib import PurePosixPath
 
-from pylint.reporters.json_reporter import OldJsonExport
+from pylint.testutils._primer.comparator import PylintComparator
 from pylint.testutils._primer.primer_command import (
     PackageData,
-    PackageMessages,
     PrimerCommand,
 )
 
@@ -19,63 +17,33 @@
 class CompareCommand(PrimerCommand):
     def run(self) -> None:
         if self.config.batches is None:
-            main_data = self._load_json(self.config.base_file)
-            pr_data = self._load_json(self.config.new_file)
+            main_data = PylintComparator.load_json(self.config.base_file)
+            pr_data = PylintComparator.load_json(self.config.new_file)
         else:
             main_data = {}
             pr_data = {}
             for idx in range(self.config.batches):
                 main_data.update(
-                    self._load_json(
+                    PylintComparator.load_json(
                         self.config.base_file.replace("BATCHIDX", "batch" + str(idx))
                     )
                 )
                 pr_data.update(
-                    self._load_json(
+                    PylintComparator.load_json(
                         self.config.new_file.replace("BATCHIDX", "batch" + str(idx))
                     )
                 )
 
-        missing_messages_data, new_messages_data = self._cross_reference(
-            main_data, pr_data
-        )
-        comment = self._create_comment(missing_messages_data, new_messages_data)
+        comparator = PylintComparator(main_data, pr_data)
+        comment = self._create_comment(comparator)
         with open(self.primer_directory / "comment.txt", "w", encoding="utf-8") as f:
             f.write(comment)
 
-    @staticmethod
-    def _cross_reference(
-        main_data: PackageMessages, pr_data: PackageMessages
-    ) -> tuple[PackageMessages, PackageMessages]:
-        missing_messages_data: PackageMessages = {}
-        for package, data in main_data.items():
-            package_missing_messages: list[OldJsonExport] = []
-            for message in data["messages"]:
-                try:
-                    pr_data[package]["messages"].remove(message)
-                except ValueError:
-                    package_missing_messages.append(message)
-            missing_messages_data[package] = PackageData(
-                commit=pr_data[package]["commit"], messages=package_missing_messages
-            )
-        return missing_messages_data, pr_data
-
-    @staticmethod
-    def _load_json(file_path: Path | str) -> PackageMessages:
-        with open(file_path, encoding="utf-8") as f:
-            result: PackageMessages = json.load(f)
-        return result
-
-    def _create_comment(
-        self, all_missing_messages: PackageMessages, all_new_messages: PackageMessages
-    ) -> str:
+    def _create_comment(self, comparator: PylintComparator) -> str:
         comment = ""
-        for package, missing_messages in all_missing_messages.items():
+        for package, missing_messages, new_messages in comparator:
             if len(comment) >= MAX_GITHUB_COMMENT_LENGTH:
                 break
-            new_messages = all_new_messages[package]
-            if not missing_messages["messages"] and not new_messages["messages"]:
-                continue
             comment += self._create_comment_for_package(
                 package, new_messages, missing_messages
             )