refactor: robust stat tracking system

Author: asajjad2

courses/management/commands/sync_external_course_runs.py

@@ -9,6 +9,7 @@
     update_external_course_runs,
 )
 from ecommerce.mail_api import send_external_data_sync_email
+from courses.management.utils import StatsCollector
 
 
 class Command(BaseCommand):
@@ -56,80 +57,24 @@ def handle(self, *args, **options):  # noqa: ARG002
             )
             return
 
+        stats_collector = StatsCollector()
+
         self.stdout.write(f"Starting course sync for {vendor_name}.")
         keymap = keymap()
         external_course_runs = fetch_external_courses(keymap)
-        stats = update_external_course_runs(external_course_runs, keymap)
+        update_external_course_runs(external_course_runs, keymap, stats_collector)
+
+        email_stats = stats_collector.email_stats()
+
         send_external_data_sync_email(
             vendor_name=vendor_name,
-            stats=stats,
+            stats=email_stats,
         )
-        self.log_stats(stats)
+        stats_collector.log_stats(self)
         self.stdout.write(
             self.style.SUCCESS(f"External course sync successful for {vendor_name}.")
         )
 
-    def log_stats(self, stats):
-        """
-        Logs the stats for the external course sync.
-
-        Args:
-            stats(dict): Dict containing results for the objects created/updated.
-        """
-
-        def extract_first_item(data_set):
-            return {item[0] for item in data_set} if data_set else set()
-
-        def log_stat(category, key, label):
-            items = extract_first_item(stats.get(key, set()))
-            self.log_style_success(f"Number of {category}: {len(items)}.")
-            self.log_style_success(f"{label}: {items or 0}\n")
-
-        log_stat("Courses Created", "courses_created", "External Course Codes")
-        log_stat("Existing Courses", "existing_courses", "External Course Codes")
-        log_stat(
-            "Course Runs Created", "course_runs_created", "External Course Run Codes"
-        )
-        log_stat(
-            "Course Runs Updated", "course_runs_updated", "External Course Run Codes"
-        )
-        log_stat("Products Created", "products_created", "Course Run courseware_ids")
-        log_stat(
-            "Product Versions Created",
-            "product_versions_created",
-            "Course Run courseware_ids",
-        )
-        log_stat(
-            "Course Runs without prices",
-            "course_runs_without_prices",
-            "External Course Codes",
-        )
-        log_stat(
-            "Course Pages Created", "course_pages_created", "External Course Codes"
-        )
-        log_stat(
-            "Course Pages Updated", "course_pages_updated", "External Course Codes"
-        )
-        log_stat(
-            "Certificate Pages Created", "certificates_created", "Course Readable IDs"
-        )
-        log_stat(
-            "Certificate Pages Updated", "certificates_updated", "Course Readable IDs"
-        )
-        log_stat(
-            "Course Runs Skipped due to bad data",
-            "course_runs_skipped",
-            "External Course Run Codes",
-        )
-        log_stat(
-            "Expired Course Runs", "course_runs_expired", "External Course Run Codes"
-        )
-        log_stat(
-            "Course Runs Deactivated",
-            "course_runs_deactivated",
-            "External Course Run Codes",
-        )
-
     def log_style_success(self, log_msg):
         """
         Logs success styled message.

courses/management/utils.py

@@ -1,5 +1,8 @@
 """Utility functions/classes for course management commands"""
 
+from dataclasses import dataclass
+from typing import Optional
+
 from django.core.management.base import BaseCommand, CommandError
 
 from courses.models import CourseRun, CourseRunEnrollment, Program, ProgramEnrollment
@@ -243,3 +246,144 @@ def enroll_in_edx(self, user, course_runs):
         ) as exc:
             self.stdout.write(self.style.WARNING(str(exc)))
         return False
+
+
+@dataclass
+class CourseInfo:
+    """
+    Data class for course information with named fields
+    """
+
+    code: str
+    title: Optional[str] = None
+    msg: Optional[str] = None
+
+
+class StatCategory:
+    """Represents a category of statistics with its display information"""
+
+    def __init__(self, key, display_name=None, label=None):
+        """
+        Initialize a stat category
+
+        Args:
+            key: The dictionary key for this stat
+            display_name: Human-readable category name
+            label: Description of the items
+        """
+        self.key = key
+        self.display_name = display_name or self._generate_display_name(key)
+        self.label = label or self._generate_label(key)
+        self.items = []
+
+    def _generate_display_name(self, key):
+        """Generate a display name from the key"""
+        return " ".join(word.capitalize() for word in key.split("_"))
+
+    def _generate_label(self, key):
+        """Generate a label based on the key"""
+        if "course" in key and "run" not in key:
+            return "External Course Codes"
+        elif "run" in key:
+            return "External Course Run Codes"
+        elif "product" in key:
+            return "Course Run courseware_ids"
+        elif "certificate" in key:
+            return "Course Readable IDs"
+        else:
+            return "Items"
+
+    def add(self, code, title=None, msg=None):
+        """Add an item to this stat category"""
+        self.items.append(CourseInfo(code=code, title=title, msg=msg))
+
+    def get_codes(self):
+        """Get the set of unique codes in this category"""
+        return {item.code for item in self.items if item.code is not None}
+
+    def __len__(self):
+        """Return the number of items in this category"""
+        return len(self.items)
+
+
+class StatsCollector:
+    """Collector for external course sync statistics with named properties"""
+
+    def __init__(self):
+        self.categories = {
+            "courses_created": StatCategory("courses_created"),
+            "existing_courses": StatCategory("existing_courses"),
+            "course_runs_created": StatCategory("course_runs_created"),
+            "course_runs_updated": StatCategory("course_runs_updated"),
+            "course_runs_without_prices": StatCategory("course_runs_without_prices"),
+            "course_runs_skipped": StatCategory(
+                "course_runs_skipped",
+                display_name="Course Runs Skipped due to bad data",
+            ),
+            "course_runs_expired": StatCategory(
+                "course_runs_deactivated", display_name="Expired Course Runs"
+            ),
+            "course_runs_deactivated": StatCategory("course_runs_deactivated"),
+            "course_pages_created": StatCategory("course_pages_created"),
+            "course_pages_updated": StatCategory("course_pages_updated"),
+            "products_created": StatCategory("products_created"),
+            "product_versions_created": StatCategory("product_versions_created"),
+            "certificates_created": StatCategory(
+                "certificates_created", display_name="Certificate Pages Created"
+            ),
+            "certificates_updated": StatCategory(
+                "certificates_updated", display_name="Certificate Pages Updated"
+            ),
+        }
+
+    def add_stat(self, key, code, title=None, msg=None):
+        """
+        Add an item to a specific stat category
+        """
+        if key in self.categories:
+            self.categories[key].add(code, title, msg)
+
+    def add_bulk(self, key, codes):
+        """
+        Add multiple items within the same category
+        """
+        if key in self.categories:
+            for code in codes:
+                self.add_stat(key, code)
+
+    def remove_duplicates(self, source_key, items_to_remove_key):
+        """
+        Remove items from one category that exist in another category
+        """
+        if (
+            source_key not in self.categories
+            or items_to_remove_key not in self.categories
+        ):
+            return
+
+        codes_to_remove = {
+            item.code for item in self.categories[items_to_remove_key].items
+        }
+
+        self.categories[source_key].items = [
+            item
+            for item in self.categories[source_key].items
+            if item.code not in codes_to_remove
+        ]
+
+    def log_stats(self, logger):
+        """
+        Log all collected statistics
+        """
+        for category in self.categories.values():
+            codes = category.get_codes()
+            logger.log_style_success(
+                f"Number of {category.display_name}: {len(codes)}."
+            )
+            logger.log_style_success(f"{category.label}: {codes or 0}\n")
+
+    def email_stats(self):
+        """
+        Return statistics formatted for email template"
+        """
+        return {key: category.items for key, category in self.categories.items()}