import itertools
import os
import random
from collections import defaultdict
from collections.abc import Mapping, Sequence
from datetime import datetime, timedelta
from typing import Any

import bmemcached
import orjson
from django.conf import settings
from django.contrib.sessions.models import Session
from django.core.files.base import File
from django.core.management import call_command
from django.core.management.base import CommandParser
from django.core.validators import validate_email
from django.db import connection
from django.db.models import F
from django.db.models.signals import post_delete
from django.utils.timezone import now as timezone_now
from typing_extensions import override

from scripts.lib.zulip_tools import get_or_create_dev_uuid_var_path
from zerver.actions.channel_folders import check_add_channel_folder
from zerver.actions.create_realm import do_create_realm
from zerver.actions.custom_profile_fields import (
    do_update_user_custom_profile_data_if_changed,
    try_add_realm_custom_profile_field,
    try_add_realm_default_custom_profile_field,
)
from zerver.actions.message_send import build_message_send_dict, do_send_messages
from zerver.actions.realm_emoji import check_add_realm_emoji
from zerver.actions.realm_linkifiers import do_add_linkifier
from zerver.actions.realm_settings import (
    do_set_realm_moderation_request_channel,
    do_set_realm_property,
)
from zerver.actions.scheduled_messages import check_schedule_message
from zerver.actions.streams import bulk_add_subscriptions
from zerver.actions.user_groups import create_user_group_in_database
from zerver.actions.user_settings import do_change_user_setting
from zerver.actions.users import do_change_user_role
from zerver.lib.bulk_create import bulk_create_streams
from zerver.lib.digest import DIGEST_CUTOFF
from zerver.lib.generate_test_data import create_test_data, generate_topics
from zerver.lib.management import ZulipBaseCommand
from zerver.lib.onboarding import create_if_missing_realm_internal_bots
from zerver.lib.onboarding_steps import ALL_ONBOARDING_STEPS
from zerver.lib.push_notifications import logger as push_notifications_logger
from zerver.lib.remote_server import get_realms_info_for_push_bouncer
from zerver.lib.server_initialization import create_internal_realm, create_users
from zerver.lib.storage import static_path
from zerver.lib.stream_color import STREAM_ASSIGNMENT_COLORS
from zerver.lib.stream_subscription import bulk_create_stream_subscriptions
from zerver.lib.types import AnalyticsDataUploadLevel, ProfileFieldData
from zerver.lib.users import add_service
from zerver.lib.utils import generate_api_key
from zerver.models import (
    AlertWord,
    Client,
    CustomProfileField,
    DefaultStream,
    DirectMessageGroup,
    Draft,
    Message,
    Reaction,
    Realm,
    RealmAuditLog,
    RealmDomain,
    RealmUserDefault,
    Recipient,
    Service,
    Stream,
    Subscription,
    UserMessage,
    UserPresence,
    UserProfile,
)
from zerver.models.alert_words import flush_alert_word
from zerver.models.clients import get_client
from zerver.models.groups import NamedUserGroup, SystemGroups
from zerver.models.onboarding_steps import OnboardingStep
from zerver.models.realm_audit_logs import AuditLogEventType
from zerver.models.realms import get_realm
from zerver.models.recipients import get_or_create_direct_message_group
from zerver.models.streams import get_stream
from zerver.models.users import get_user, get_user_by_delivery_email, get_user_profile_by_id
from zilencer.models import RemoteRealm, RemoteZulipServer, RemoteZulipServerAuditLog
from zilencer.views import update_remote_realm_data_for_server

# Disable the push notifications bouncer to avoid enqueuing updates in
# maybe_enqueue_audit_log_upload during early setup.
settings.ZULIP_SERVICE_PUSH_NOTIFICATIONS = False
settings.ZULIP_SERVICE_SUBMIT_USAGE_STATISTICS = False
settings.ZULIP_SERVICE_SECURITY_ALERTS = False
settings.ANALYTICS_DATA_UPLOAD_LEVEL = AnalyticsDataUploadLevel.NONE
settings.USING_TORNADO = False
# Disable using memcached caches to avoid 'unsupported pickle
# protocol' errors if `populate_db` is run with a different Python
# from `run-dev`.
default_cache = settings.CACHES["default"]
settings.CACHES["default"] = {
    "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
}

DEFAULT_EMOJIS = [
    ("+1", "1f44d"),
    ("smiley", "1f603"),
    ("eyes", "1f440"),
    ("crying_cat_face", "1f63f"),
    ("arrow_up", "2b06"),
    ("confetti_ball", "1f38a"),
    ("hundred_points", "1f4af"),
]


def clear_database() -> None:
    # Hacky function only for use inside populate_db.  Designed to
    # allow running populate_db repeatedly in series to work without
    # flushing memcached or clearing the database manually.

    # With `zproject.test_settings`, we aren't using real memcached
    # and; we only need to flush memcached if we're populating a
    # database that would be used with it (i.e. zproject.dev_settings).
    if default_cache["BACKEND"] == "zerver.lib.singleton_bmemcached.SingletonBMemcached":
        memcached_client = bmemcached.Client(
            (default_cache["LOCATION"],),
            **default_cache["OPTIONS"],
        )
        try:
            memcached_client.flush_all()
        finally:
            memcached_client.disconnect_all()

    model: Any = None  # Hack because mypy doesn't know these are model classes

    # The after-delete signal on this just updates caches, and slows
    # down the deletion noticeably.  Remove the signal and replace it
    # after we're done.
    post_delete.disconnect(flush_alert_word, sender=AlertWord)
    for model in [
        Message,
        Stream,
        AlertWord,
        UserProfile,
        Recipient,
        Realm,
        Subscription,
        DirectMessageGroup,
        UserMessage,
        Client,
        DefaultStream,
        RemoteRealm,
        RemoteZulipServer,
    ]:
        model.objects.all().delete()
    Session.objects.all().delete()
    post_delete.connect(flush_alert_word, sender=AlertWord)


def subscribe_users_to_streams(realm: Realm, stream_dict: dict[str, dict[str, Any]]) -> None:
    subscriptions_to_add = []
    event_time = timezone_now()
    # Backdate few channel subscriptions to support digest previews in dev.
    # This ensures sample data appears in `/digest` by making subscriptions
    # appear old enough to pass the cutoff check in `get_user_stream_map`.
    event_time_for_digest = event_time - timedelta(days=DIGEST_CUTOFF)
    all_subscription_logs = []
    subscriber_count_changes: dict[int, set[int]] = defaultdict(set)
    profiles = UserProfile.objects.select_related("realm").filter(realm=realm)
    for i, stream_name in enumerate(stream_dict):
        stream = Stream.objects.get(name=stream_name, realm=realm)
        recipient = Recipient.objects.get(type=Recipient.STREAM, type_id=stream.id)
        for profile in profiles:
            # Subscribe to some streams.
            s = Subscription(
                recipient=recipient,
                user_profile=profile,
                is_user_active=profile.is_active,
                color=STREAM_ASSIGNMENT_COLORS[i % len(STREAM_ASSIGNMENT_COLORS)],
            )
            subscriptions_to_add.append(s)
            if profile.is_active:
                subscriber_count_changes[stream.id].add(profile.id)

            log = RealmAuditLog(
                realm=profile.realm,
                modified_user=profile,
                modified_stream=stream,
                event_last_message_id=0,
                event_type=AuditLogEventType.SUBSCRIPTION_CREATED,
                event_time=event_time_for_digest if i < 4 else event_time,
            )
            all_subscription_logs.append(log)
    bulk_create_stream_subscriptions(subs=subscriptions_to_add, streams=subscriber_count_changes)
    RealmAuditLog.objects.bulk_create(all_subscription_logs)


def create_alert_words(realm_id: int) -> None:
    user_ids = UserProfile.objects.filter(
        realm_id=realm_id,
        is_bot=False,
        is_active=True,
    ).values_list("id", flat=True)

    alert_words = [
        "algorithms",
        "complexity",
        "founded",
        "galaxy",
        "grammar",
        "illustrious",
        "natural",
        "objective",
        "people",
        "robotics",
        "study",
    ]

    recs: list[AlertWord] = []
    for user_id in user_ids:
        random.shuffle(alert_words)
        recs.extend(
            AlertWord(realm_id=realm_id, user_profile_id=user_id, word=word)
            for word in alert_words[:4]
        )

    AlertWord.objects.bulk_create(recs)


class Command(ZulipBaseCommand):
    help = "Populate a test database"

    @override
    def add_arguments(self, parser: CommandParser) -> None:
        parser.add_argument(
            "-n", "--num-messages", type=int, default=1000, help="The number of messages to create."
        )

        parser.add_argument(
            "-o",
            "--oldest-message-days",
            type=int,
            default=5,
            help="The start of the time range where messages could have been sent.",
        )

        parser.add_argument(
            "-b",
            "--batch-size",
            type=int,
            default=1000,
            help="How many messages to process in a single batch",
        )

        parser.add_argument(
            "--extra-users", type=int, default=0, help="The number of extra users to create"
        )

        parser.add_argument(
            "--extra-bots", type=int, default=0, help="The number of extra bots to create"
        )

        parser.add_argument(
            "--extra-streams", type=int, default=0, help="The number of extra streams to create"
        )

        parser.add_argument("--max-topics", type=int, help="The number of maximum topics to create")

        parser.add_argument(
            "--direct-message-groups",
            dest="num_direct_message_groups",
            type=int,
            default=3,
            help="The number of direct message groups to create.",
        )

        parser.add_argument(
            "--personals",
            dest="num_personals",
            type=int,
            default=6,
            help="The number of personal pairs to create.",
        )

        parser.add_argument("--threads", type=int, default=1, help="The number of threads to use.")

        parser.add_argument(
            "--percent-direct-message-groups",
            type=float,
            default=15,
            help="The percent of messages to be direct message groups.",
        )

        parser.add_argument(
            "--percent-personals",
            type=float,
            default=15,
            help="The percent of messages to be personals.",
        )

        parser.add_argument(
            "--stickiness",
            type=float,
            default=20,
            help="The percent of messages to repeat recent folks.",
        )

        parser.add_argument(
            "--nodelete",
            action="store_false",
            dest="delete",
            help="Whether to delete all the existing messages.",
        )

        parser.add_argument(
            "--test-suite",
            action="store_true",
            help="Configures populate_db to create a deterministic data set for the backend tests.",
        )

    @override
    def handle(self, *args: Any, **options: Any) -> None:
        # Suppress spammy output from the push notifications logger
        push_notifications_logger.disabled = True

        if options["percent_direct_message_groups"] + options["percent_personals"] > 100:
            self.stderr.write("Error!  More than 100% of messages allocated.\n")
            return

        # Get consistent data for backend tests.
        if options["test_suite"]:
            random.seed(0)

            with connection.cursor() as cursor:
                # Sometimes bugs relating to confusing recipient.id for recipient.type_id
                # or <object>.id for <object>.recipient_id remain undiscovered by the test suite
                # due to these numbers happening to coincide in such a way that it makes tests
                # accidentally pass. By bumping the Recipient.id sequence by a large enough number,
                # we can have those ids in a completely different range of values than object ids,
                # eliminating the possibility of such coincidences.
                cursor.execute("SELECT setval('zerver_recipient_id_seq', 100)")

        if options["max_topics"] is None:
            # If max_topics is not set, we use a default that's big
            # enough "show all topics" should appear, and scales slowly
            # with the number of messages.
            options["max_topics"] = 8 + options["num_messages"] // 1000

        if options["delete"]:
            # Start by clearing all the data in our database
            clear_database()

            # Create our three default realms
            # Could in theory be done via zerver.actions.create_realm.do_create_realm, but
            # welcome-bot (needed for do_create_realm) hasn't been created yet
            create_internal_realm()
            zulip_realm = do_create_realm(
                string_id="zulip",
                name="Zulip Dev",
                emails_restricted_to_domains=False,
                description="The Zulip development environment default organization."
                "  It's great for testing!",
                invite_required=False,
                plan_type=Realm.PLAN_TYPE_SELF_HOSTED,
                org_type=Realm.ORG_TYPES["business"]["id"],
                enable_read_receipts=True,
                enable_spectator_access=True,
            )
            RealmDomain.objects.create(realm=zulip_realm, domain="zulip.com")
            assert zulip_realm.new_stream_announcements_stream is not None
            zulip_realm.new_stream_announcements_stream.name = "Verona"
            zulip_realm.new_stream_announcements_stream.description = "A city in Italy"
            zulip_realm.new_stream_announcements_stream.save(update_fields=["name", "description"])

            realm_user_default = RealmUserDefault.objects.get(realm=zulip_realm)
            realm_user_default.enter_sends = True
            realm_user_default.email_address_visibility = (
                RealmUserDefault.EMAIL_ADDRESS_VISIBILITY_ADMINS
            )
            realm_user_default.save()

            if options["test_suite"]:
                mit_realm = do_create_realm(
                    string_id="zephyr",
                    name="MIT",
                    emails_restricted_to_domains=True,
                    invite_required=False,
                    plan_type=Realm.PLAN_TYPE_SELF_HOSTED,
                    org_type=Realm.ORG_TYPES["business"]["id"],
                )
                RealmDomain.objects.create(realm=mit_realm, domain="mit.edu")

                lear_realm = do_create_realm(
                    string_id="lear",
                    name="Lear & Co.",
                    emails_restricted_to_domains=False,
                    invite_required=False,
                    plan_type=Realm.PLAN_TYPE_SELF_HOSTED,
                    org_type=Realm.ORG_TYPES["business"]["id"],
                )

            # Realms should have matching RemoteRealm entries - simulating having realms registered
            # with the bouncer, which is going to be the primary case for modern servers. Tests
            # wanting to have missing registrations, or simulating legacy server scenarios,
            # should delete RemoteRealms to explicit set things up.

            assert isinstance(settings.ZULIP_ORG_ID, str)
            assert isinstance(settings.ZULIP_ORG_KEY, str)
            server = RemoteZulipServer.objects.create(
                uuid=settings.ZULIP_ORG_ID,
                api_key=settings.ZULIP_ORG_KEY,
                hostname=settings.EXTERNAL_HOST,
                last_updated=timezone_now(),
                contact_email="remotezulipserver@zulip.com",
            )
            RemoteZulipServerAuditLog.objects.create(
                event_type=AuditLogEventType.REMOTE_SERVER_CREATED,
                server=server,
                event_time=server.last_updated,
            )
            update_remote_realm_data_for_server(server, get_realms_info_for_push_bouncer())

            # Create test Users (UserProfiles are automatically created,
            # as are subscriptions to the ability to receive personals).
            names = [
                ("Zoe", "ZOE@zulip.com"),
                ("Othello, the Moor of Venice", "othello@zulip.com"),
                ("Iago", "iago@zulip.com"),
                ("Prospero from The Tempest", "prospero@zulip.com"),
                ("Cordelia, Lear's daughter", "cordelia@zulip.com"),
                ("King Hamlet", "hamlet@zulip.com"),
                ("aaron", "AARON@zulip.com"),
                ("Polonius", "polonius@zulip.com"),
                ("Desdemona", "desdemona@zulip.com"),
                ("शिव", "shiva@zulip.com"),
            ]

            # For testing really large batches:
            # Create extra users with semi realistic names to make search
            # functions somewhat realistic.  We'll still create 1000 users
            # like Extra222 User for some predictability.
            num_names = options["extra_users"]
            num_boring_names = 300

            for i in range(min(num_names, num_boring_names)):
                full_name = f"Extra{i:03} User"
                names.append((full_name, f"extrauser{i}@zulip.com"))

            if num_names > num_boring_names:
                fnames = [
                    "Amber",
                    "Arpita",
                    "Bob",
                    "Cindy",
                    "Daniela",
                    "Dan",
                    "Dinesh",
                    "Faye",
                    "François",
                    "George",
                    "Hank",
                    "Irene",
                    "James",
                    "Janice",
                    "Jenny",
                    "Jill",
                    "John",
                    "Kate",
                    "Katelyn",
                    "Kobe",
                    "Lexi",
                    "Manish",
                    "Mark",
                    "Matt",
                    "Mayna",
                    "Michael",
                    "Pete",
                    "Peter",
                    "Phil",
                    "Phillipa",
                    "Preston",
                    "Sally",
                    "Scott",
                    "Sandra",
                    "Steve",
                    "Stephanie",
                    "Vera",
                ]
                mnames = ["de", "van", "von", "Shaw", "T."]
                lnames = [
                    "Adams",
                    "Agarwal",
                    "Beal",
                    "Benson",
                    "Bonita",
                    "Davis",
                    "George",
                    "Harden",
                    "James",
                    "Jones",
                    "Johnson",
                    "Jordan",
                    "Lee",
                    "Leonard",
                    "Singh",
                    "Smith",
                    "Patel",
                    "Towns",
                    "Wall",
                ]
                non_ascii_names = [
                    "Günter",
                    "أحمد",
                    "Magnús",
                    "आशी",
                    "イツキ",
                    "语嫣",
                    "அருண்",
                    "Александр",
                    "José",
                ]
                # to imitate emoji insertions in usernames
                raw_emojis = ["😎", "😂", "🐱‍👤"]

            for i in range(num_boring_names, num_names):
                fname = random.choice(fnames) + str(i)
                full_name = fname
                if random.random() < 0.7:
                    if random.random() < 0.3:
                        full_name += " " + random.choice(non_ascii_names)
                    else:
                        full_name += " " + random.choice(mnames)
                    if random.random() < 0.1:
                        full_name += f" {random.choice(raw_emojis)} "
                    else:
                        full_name += " " + random.choice(lnames)
                email = fname.lower().encode("ascii", "ignore").decode("ascii") + "@zulip.com"
                validate_email(email)
                names.append((full_name, email))

            create_users(zulip_realm, names, tos_version=settings.TERMS_OF_SERVICE_VERSION)

            # Add time zones to some users. Ideally, this would be
            # done in the initial create_users calls, but the
            # tuple-based interface for that function doesn't support
            # doing so.
            def assign_time_zone_by_delivery_email(delivery_email: str, new_time_zone: str) -> None:
                u = get_user_by_delivery_email(delivery_email, zulip_realm)
                u.timezone = new_time_zone
                u.save(update_fields=["timezone"])

            # Note: Hamlet keeps default time zone of "".
            assign_time_zone_by_delivery_email("AARON@zulip.com", "US/Pacific")
            assign_time_zone_by_delivery_email("othello@zulip.com", "US/Pacific")
            assign_time_zone_by_delivery_email("ZOE@zulip.com", "US/Eastern")
            assign_time_zone_by_delivery_email("iago@zulip.com", "US/Eastern")
            assign_time_zone_by_delivery_email("desdemona@zulip.com", "Canada/Newfoundland")
            assign_time_zone_by_delivery_email("polonius@zulip.com", "Asia/Shanghai")  # China
            assign_time_zone_by_delivery_email("shiva@zulip.com", "Asia/Kolkata")  # India
            assign_time_zone_by_delivery_email("cordelia@zulip.com", "UTC")

            iago = get_user_by_delivery_email("iago@zulip.com", zulip_realm)
            do_change_user_role(iago, UserProfile.ROLE_REALM_ADMINISTRATOR, acting_user=None)
            iago.is_staff = True
            iago.save(update_fields=["is_staff"])

            # We need to create at least two test draft for Iago for the sake
            # of the cURL tests. Two since one will be deleted.
            Draft.objects.create(
                user_profile=iago,
                recipient=None,
                topic="Release Notes",
                content="Release 4.0 will contain ...",
                last_edit_time=timezone_now(),
            )
            Draft.objects.create(
                user_profile=iago,
                recipient=None,
                topic="Release Notes",
                content="Release 4.0 will contain many new features such as ... ",
                last_edit_time=timezone_now(),
            )

            desdemona = get_user_by_delivery_email("desdemona@zulip.com", zulip_realm)
            do_change_user_role(desdemona, UserProfile.ROLE_REALM_OWNER, acting_user=None)

            shiva = get_user_by_delivery_email("shiva@zulip.com", zulip_realm)
            do_change_user_role(shiva, UserProfile.ROLE_MODERATOR, acting_user=None)

            polonius = get_user_by_delivery_email("polonius@zulip.com", zulip_realm)
            do_change_user_role(polonius, UserProfile.ROLE_GUEST, acting_user=None)

            # These bots are directly referenced from code and thus
            # are needed for the test suite.
            zulip_realm_bots = [
                ("Zulip Default Bot", "default-bot@zulip.com"),
                *(
                    (f"Extra Bot {i}", f"extrabot{i}@zulip.com")
                    for i in range(options["extra_bots"])
                ),
            ]

            create_users(
                zulip_realm, zulip_realm_bots, bot_type=UserProfile.DEFAULT_BOT, bot_owner=desdemona
            )

            zoe = get_user_by_delivery_email("zoe@zulip.com", zulip_realm)
            zulip_webhook_bots = [
                ("Zulip Webhook Bot", "webhook-bot@zulip.com"),
            ]
            # If a stream is not supplied in the webhook URL, the webhook
            # will (in some cases) send the notification as a PM to the
            # owner of the webhook bot, so bot_owner can't be None
            create_users(
                zulip_realm,
                zulip_webhook_bots,
                bot_type=UserProfile.INCOMING_WEBHOOK_BOT,
                bot_owner=zoe,
            )
            aaron = get_user_by_delivery_email("AARON@zulip.com", zulip_realm)

            zulip_outgoing_bots = [
                ("Outgoing Webhook", "outgoing-webhook@zulip.com"),
            ]
            create_users(
                zulip_realm,
                zulip_outgoing_bots,
                bot_type=UserProfile.OUTGOING_WEBHOOK_BOT,
                bot_owner=aaron,
            )
            outgoing_webhook = get_user("outgoing-webhook@zulip.com", zulip_realm)
            add_service(
                "outgoing-webhook",
                user_profile=outgoing_webhook,
                interface=Service.GENERIC,
                base_url="http://127.0.0.1:5002",
                token=generate_api_key(),
            )

            # Add the realm internal bots to each realm.
            create_if_missing_realm_internal_bots()

            # Create streams.
            zulip_discussion_channel_name = str(Realm.ZULIP_DISCUSSION_CHANNEL_NAME)
            zulip_sandbox_channel_name = str(Realm.ZULIP_SANDBOX_CHANNEL_NAME)

            stream_list = [
                "Verona",
                "Denmark",
                "Scotland",
                "Venice",
                "Rome",
                "core team",
                zulip_discussion_channel_name,
                zulip_sandbox_channel_name,
            ]
            stream_dict: dict[str, dict[str, Any]] = {
                "Denmark": {"description": "A Scandinavian country"},
                "Scotland": {"description": "Located in the United Kingdom", "creator": iago},
                "Venice": {"description": "A northeastern Italian city", "creator": polonius},
                "Rome": {"description": "Yet another Italian city", "is_web_public": True},
                "core team": {
                    "description": "A private channel for core team members",
                    "invite_only": True,
                    "creator": desdemona,
                },
            }

            bulk_create_streams(zulip_realm, stream_dict)
            recipient_streams: list[int] = [
                Stream.objects.get(name=name, realm=zulip_realm).id for name in stream_list
            ]

            # Create subscriptions to streams.  The following
            # algorithm will give each of the users a different but
            # deterministic subset of the streams (given a fixed list
            # of users). For the test suite, we have a fixed list of
            # subscriptions to make sure test data is consistent
            # across platforms.

            subscriptions_list: list[tuple[UserProfile, Recipient]] = []
            profiles: Sequence[UserProfile] = list(
                UserProfile.objects.select_related("realm").filter(is_bot=False).order_by("email")
            )

            if options["test_suite"]:
                subscriptions_map = {
                    "AARON@zulip.com": ["Verona"],
                    "cordelia@zulip.com": ["Verona"],
                    "hamlet@zulip.com": [
                        "Verona",
                        "Denmark",
                        "core team",
                        zulip_discussion_channel_name,
                        zulip_sandbox_channel_name,
                    ],
                    "iago@zulip.com": [
                        "Verona",
                        "Denmark",
                        "Scotland",
                        "core team",
                        zulip_discussion_channel_name,
                        zulip_sandbox_channel_name,
                    ],
                    "othello@zulip.com": ["Verona", "Denmark", "Scotland"],
                    "prospero@zulip.com": ["Verona", "Denmark", "Scotland", "Venice"],
                    "ZOE@zulip.com": ["Verona", "Denmark", "Scotland", "Venice", "Rome"],
                    "polonius@zulip.com": ["Verona"],
                    "desdemona@zulip.com": [
                        "Verona",
                        "Denmark",
                        "Venice",
                        "core team",
                        zulip_discussion_channel_name,
                        zulip_sandbox_channel_name,
                    ],
                    "shiva@zulip.com": ["Verona", "Denmark", "Scotland"],
                }

                for profile in profiles:
                    email = profile.delivery_email
                    if email not in subscriptions_map:
                        raise Exception(f"Subscriptions not listed for user {email}")

                    for stream_name in subscriptions_map[email]:
                        stream = Stream.objects.get(name=stream_name, realm=zulip_realm)
                        r = Recipient.objects.get(type=Recipient.STREAM, type_id=stream.id)
                        subscriptions_list.append((profile, r))
            else:
                num_streams = len(recipient_streams)
                num_users = len(profiles)
                for i, profile in enumerate(profiles):
                    # Subscribe to some streams.
                    fraction = float(i) / num_users
                    num_recips = int(num_streams * fraction) + 1

                    for type_id in recipient_streams[:num_recips]:
                        r = Recipient.objects.get(type=Recipient.STREAM, type_id=type_id)
                        subscriptions_list.append((profile, r))

            subscriptions_to_add: list[Subscription] = []
            subscriber_count_changes: dict[int, set[int]] = defaultdict(set)
            event_time = timezone_now()
            all_subscription_logs: list[RealmAuditLog] = []

            i = 0
            for profile, recipient in subscriptions_list:
                i += 1
                color = STREAM_ASSIGNMENT_COLORS[i % len(STREAM_ASSIGNMENT_COLORS)]
                s = Subscription(
                    recipient=recipient,
                    user_profile=profile,
                    is_user_active=profile.is_active,
                    color=color,
                )

                subscriptions_to_add.append(s)
                if profile.is_active:
                    subscriber_count_changes[recipient.type_id].add(profile.id)

                log = RealmAuditLog(
                    realm=profile.realm,
                    modified_user=profile,
                    modified_stream_id=recipient.type_id,
                    event_last_message_id=0,
                    event_type=AuditLogEventType.SUBSCRIPTION_CREATED,
                    event_time=event_time,
                )
                all_subscription_logs.append(log)

            bulk_create_stream_subscriptions(
                subs=subscriptions_to_add, streams=subscriber_count_changes
            )
            RealmAuditLog.objects.bulk_create(all_subscription_logs)

            # Create custom profile field data
            phone_number = try_add_realm_custom_profile_field(
                zulip_realm, "Phone number", CustomProfileField.SHORT_TEXT, hint=""
            )
            biography = try_add_realm_custom_profile_field(
                zulip_realm,
                "Biography",
                CustomProfileField.LONG_TEXT,
                hint="What are you known for?",
            )
            favorite_food = try_add_realm_custom_profile_field(
                zulip_realm,
                "Favorite food",
                CustomProfileField.SHORT_TEXT,
                hint="Or drink, if you'd prefer",
            )
            field_data: ProfileFieldData = {
                "0": {"text": "Vim", "order": "1"},
                "1": {"text": "Emacs", "order": "2"},
            }
            favorite_editor = try_add_realm_custom_profile_field(
                zulip_realm, "Favorite editor", CustomProfileField.SELECT, field_data=field_data
            )
            birthday = try_add_realm_custom_profile_field(
                zulip_realm, "Birthday", CustomProfileField.DATE
            )
            favorite_website = try_add_realm_custom_profile_field(
                zulip_realm,
                "Favorite website",
