zulip/zerver/management/commands/unarchive_channel.py

from argparse import ArgumentParser
from typing import Any

from django.core.management.base import CommandError
from typing_extensions import override

from zerver.actions.streams import deactivated_streams_by_old_name, do_unarchive_stream
from zerver.lib.management import ZulipBaseCommand
from zerver.models import RealmAuditLog, Stream


class Command(ZulipBaseCommand):
    help = """Reactivate a channel that was deactivated."""

    @override
    def add_arguments(self, parser: ArgumentParser) -> None:
        specify_channel = parser.add_mutually_exclusive_group(required=True)
        specify_channel.add_argument(
            "-c",
            "--channel",
            help="Name of a deactivated channel in the realm.",
        )
        specify_channel.add_argument(
            "--channel-id",
            help="ID of a deactivated channel in the realm.",
        )
        parser.add_argument(
            "-n",
            "--new-name",
            required=False,
            help="Name to reactivate as; defaults to the old name.",
        )
        self.add_realm_args(parser, required=True)

    @override
    def handle(self, *args: Any, **options: str | None) -> None:
        realm = self.get_realm(options)
        assert realm is not None  # Should be ensured by parser

        # Looking up the channel is complicated, since they get renamed
        # when they are deactivated, in a transformation which may be
        # lossy.

        if options["channel_id"] is not None:
            channel = Stream.objects.get(id=options["channel_id"])
            if channel.realm_id != realm.id:
                raise CommandError(
                    f"Channel id {channel.id}, named '{channel.name}', is in realm '{channel.realm.string_id}', not '{realm.string_id}'"
                )
            if not channel.deactivated:
                raise CommandError(
                    f"Channel id {channel.id}, named '{channel.name}', is not deactivated"
                )
            if options["new_name"] is None:
                raise CommandError("--new-name flag is required with --channel-id")
            new_name = options["new_name"]
        else:
            channel_name = options["channel"]
            assert channel_name is not None

            possible_channels = deactivated_streams_by_old_name(realm, channel_name)
            if len(possible_channels) == 0:
                raise CommandError("No matching deactivated channels found!")

            if len(possible_channels) > 1:
                # Print ids of all possible channels, support passing by id
                print("Matching channels:")
                for channel in possible_channels:
                    last_deactivation = (
                        RealmAuditLog.objects.filter(
                            realm=realm,
                            modified_stream=channel,
                            event_type=RealmAuditLog.STREAM_DEACTIVATED,
                        )
                        .order_by("-id")
                        .first()
                    )
                    assert last_deactivation is not None
                    print(
                        f"  ({channel.id}) {channel.name}, deactivated on {last_deactivation.event_time}"
                    )
                raise CommandError(
                    "More than one matching channel found!  Specify which with --channel-id"
                )

            channel = possible_channels[0]
            if options["new_name"] is not None:
                new_name = options["new_name"]
            else:
                new_name = channel_name

        if Stream.objects.filter(realm=realm, name=new_name).exists():
            raise CommandError(
                f"Channel with name '{new_name}' already exists; pass a different --new-name"
            )

        assert channel is not None
        do_unarchive_stream(channel, new_name, acting_user=None)