diff --git a/twitch/management/commands/better_import_drops.py b/twitch/management/commands/better_import_drops.py index e402310..4a58d93 100644 --- a/twitch/management/commands/better_import_drops.py +++ b/twitch/management/commands/better_import_drops.py @@ -1021,14 +1021,16 @@ class Command(BaseCommand): campaign_obj.save(update_fields=["allow_is_enabled"]) # Get or create all channels and collect them + # Only update the M2M relationship if we have channel data from the API. + # This prevents clearing existing channel associations when the API returns + # no channels (which can happen for disabled campaigns or incomplete responses). channel_objects: list[Channel] = [] if allow_schema.channels: for channel_schema in allow_schema.channels: channel_obj: Channel = self._get_or_create_channel(channel_info=channel_schema) channel_objects.append(channel_obj) - - # Update the M2M relationship with the allowed channels - campaign_obj.allow_channels.set(channel_objects) + # Only update the M2M relationship if we have channels + campaign_obj.allow_channels.set(channel_objects) def _process_reward_campaigns( self, diff --git a/twitch/management/commands/cleanup_orphaned_channels.py b/twitch/management/commands/cleanup_orphaned_channels.py new file mode 100644 index 0000000..672b97e --- /dev/null +++ b/twitch/management/commands/cleanup_orphaned_channels.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from django.core.management.base import BaseCommand +from django.db.models import Count + +from twitch.models import Channel + +if TYPE_CHECKING: + from argparse import ArgumentParser + + from debug_toolbar.panels.templates.panel import QuerySet + +SAMPLE_PREVIEW_COUNT = 10 + + +class Command(BaseCommand): + """Management command to clean up orphaned channels with no campaigns. + + Orphaned channels are channels that exist in the database but are not + associated with any drop campaigns. This can happen when campaign ACLs + are updated with empty channel lists, clearing previous associations. + """ + + help = "Remove channels that have no associated drop campaigns" + + def add_arguments(self, parser: ArgumentParser) -> None: + """Add command arguments.""" + parser.add_argument( + "--dry-run", + action="store_true", + help="Show what would be deleted without actually deleting", + ) + parser.add_argument( + "--force", + action="store_true", + help="Delete channels without confirmation prompt", + ) + + def handle(self, **options: str | bool) -> None: + """Execute the command.""" + dry_run: str | bool = options["dry_run"] + force: str | bool = options["force"] + + # Find channels with no campaigns + orphaned_channels: QuerySet[Channel, Channel] = Channel.objects.annotate( + campaign_count=Count("allowed_campaigns"), + ).filter(campaign_count=0) + + count: int = orphaned_channels.count() + + if count == 0: + self.stdout.write(self.style.SUCCESS("No orphaned channels found.")) + return + + self.stdout.write(f"Found {count} orphaned channels with no associated campaigns:") + + # Show sample of channels to be deleted + for channel in orphaned_channels[:SAMPLE_PREVIEW_COUNT]: + self.stdout.write(f" - {channel.display_name} (Twitch ID: {channel.twitch_id})") + + if count > SAMPLE_PREVIEW_COUNT: + self.stdout.write(f" ... and {count - SAMPLE_PREVIEW_COUNT} more") + + if dry_run: + self.stdout.write(self.style.WARNING(f"\n[DRY RUN] Would delete {count} orphaned channels.")) + return + + if not force: + response: str = input(f"\nAre you sure you want to delete {count} orphaned channels? (yes/no): ") + if response.lower() != "yes": + self.stdout.write(self.style.WARNING("Cancelled.")) + return + + # Delete the orphaned channels + deleted_count, _ = orphaned_channels.delete() + + self.stdout.write(self.style.SUCCESS(f"\nSuccessfully deleted {deleted_count} orphaned channels."))