Add utility script to detect orphaned PRCs in structured content

[noissue]

pulp_deb/app/management/commands/check_for_orphan_content.py

@@ -0,0 +1,62 @@
+from gettext import gettext as _
+from django.core.management import BaseCommand
+from pulpcore.plugin.models import RepositoryVersion
+
+from pulp_deb.app.models.content.content import Package
+from pulp_deb.app.models.content.structure_content import PackageReleaseComponent
+from pulp_deb.app.models.repository import AptRepository
+
+
+class Command(BaseCommand):
+    """
+    TODO DESCRIPTION
+    """
+
+    help = _(__doc__)
+
+    def handle(self, *args, **options):
+        """Implement the command."""
+        repo_qs = AptRepository.objects.all()
+        repo_version_qs = RepositoryVersion.objects.filter(repository__in=repo_qs)
+
+        orphans_found = False
+        orphan_details = []
+
+        for repo_version in repo_version_qs:
+            content = repo_version.get_content()
+            package_content_qs = content.filter(pulp_type=Package.get_pulp_type()).only("pk")
+            package_qs = Package.objects.filter(pk__in=package_content_qs)
+            prc_content_qs = content.filter(pulp_type=PackageReleaseComponent.get_pulp_type())
+            prc_qs = PackageReleaseComponent.objects.filter(pk__in=prc_content_qs.only("pk"))
+
+            if prc_qs:
+                prc_ids = {prc.package.pulp_id for prc in prc_qs}
+                package_ids = {package.pulp_id for package in package_qs}
+
+                if prc_ids != package_ids:
+                    orphans_found = True
+                    orphan_packages = prc_ids - package_ids
+                    orphan_prcs = prc_qs.filter(package__pulp_id__in=orphan_packages)
+                    orphan_details.append(
+                        {
+                            "repo_version": repo_version,
+                            "orphan_packages": list(orphan_prcs),
+                        }
+                    )
+
+        if orphans_found:
+            for detail in orphan_details:
+                orphan_prcs_output = "\n".join(str(prc) for prc in detail["orphan_packages"])
+                self.stdout.write(
+                    f"\n{'-' * 40}\n"
+                    f"Orphans found in {detail['repo_version']}:\n"
+                    f"{'-' * 40}\n"
+                    f"Orphaned PRCs:\n{orphan_prcs_output}"
+                )
+            self.stdout.write(f"{'=' * 40}\n")
+            self.stdout.write(f"\nTotal affected repository versions: {len(orphan_details)}\n")
+            self.stdout.write(f"{'=' * 40}\n")
+        else:
+            self.stdout.write(f"{'=' * 40}\n")
+            self.stdout.write("No orphans found.\n")
+            self.stdout.write(f"{'=' * 40}\n")