upstream-zulip/zerver/data_import/mattermost.py

"""
spec:
https://docs.mattermost.com/administration/bulk-export.html
"""

import logging
import os
import re
import shutil
from dataclasses import dataclass
from typing import Any, TypeAlias

import orjson
from django.conf import settings
from django.forms.models import model_to_dict
from django.utils.text import slugify
from django.utils.timezone import now as timezone_now

from zerver.data_import.import_util import (
    AttachmentRecordData,
    ImportedBotEmail,
    SubscriberHandler,
    UploadRecordData,
    ZerverFieldsT,
    build_attachment,
    build_direct_message_group,
    build_direct_message_group_subscriptions,
    build_message,
    build_realm,
    build_realm_emoji,
    build_recipient,
    build_stream,
    build_stream_subscriptions,
    build_user_profile,
    build_zerver_realm,
    create_converted_data_files,
    get_attachment_path_and_content,
    get_domain_name_for_import,
    make_subscriber_map,
    make_user_messages,
)
from zerver.data_import.sequencer import NEXT_ID, IdMapper
from zerver.data_import.user_handler import UserHandler
from zerver.lib.emoji import name_to_codepoint
from zerver.lib.export import do_common_export_processes
from zerver.lib.import_realm import validate_and_resolve_relative_path
from zerver.lib.markdown import IMAGE_EXTENSIONS
from zerver.lib.message import truncate_content
from zerver.lib.upload import sanitize_name
from zerver.lib.utils import process_list_in_batches
from zerver.models import Reaction, RealmEmoji, Recipient, UserProfile
from zerver.models.streams import Stream


@dataclass
class ChannelMetadata:
    zulip_channel_id: int
    zulip_recipient_id: int


AddedChannelsT: TypeAlias = dict[str, ChannelMetadata]


def make_realm(realm_id: int, team: dict[str, Any]) -> ZerverFieldsT:
    # set correct realm details
    NOW = float(timezone_now().timestamp())
    domain_name = settings.EXTERNAL_HOST
    realm_subdomain = team["name"]

    zerver_realm = build_zerver_realm(realm_id, realm_subdomain, NOW, "Mattermost")
    realm = build_realm(zerver_realm, realm_id, domain_name, import_source="mattermost")

    # We may override these later.
    realm["zerver_defaultstream"] = []
    realm["zerver_stream"] = []
    realm["zerver_recipient"] = []

    return realm


def process_user(
    user_dict: dict[str, Any], realm_id: int, team_name: str, user_id_mapper: IdMapper[str]
) -> ZerverFieldsT:
    def is_team_admin(user_dict: dict[str, Any]) -> bool:
        if user_dict["teams"] is None:
            return False
        return any(
            team["name"] == team_name and "team_admin" in team["roles"]
            for team in user_dict["teams"]
        )

    def is_team_guest(user_dict: dict[str, Any]) -> bool:
        if user_dict["teams"] is None:
            return False
        for team in user_dict["teams"]:
            if team["name"] == team_name and "team_guest" in team["roles"]:
                return True
        return False

    def get_full_name(user_dict: dict[str, Any]) -> str:
        full_name = "{} {}".format(user_dict["first_name"], user_dict["last_name"])
        if full_name.strip():
            return full_name
        # Bots only have display_name.
        return user_dict.get("display_name") or user_dict["username"]

    avatar_source = UserProfile.DEFAULT_AVATAR_SOURCE
    full_name = get_full_name(user_dict)
    id = user_id_mapper.get(user_dict["username"])
    delivery_email = user_dict["email"]
    email = user_dict["email"]
    short_name = user_dict["username"]
    date_joined = int(timezone_now().timestamp())
    timezone = "UTC"

    if is_team_admin(user_dict):
        role = UserProfile.ROLE_REALM_OWNER
    elif is_team_guest(user_dict):
        role = UserProfile.ROLE_GUEST
    else:
        role = UserProfile.ROLE_MEMBER

    if user_dict.get("is_bot"):
        is_bot = True
        bot_type = UserProfile.DEFAULT_BOT
    else:
        is_bot = False
        bot_type = None

    if user_dict["is_mirror_dummy"]:
        is_active = False
        is_mirror_dummy = True
    else:
        is_active = True
        is_mirror_dummy = False

    return build_user_profile(
        avatar_source=avatar_source,
        date_joined=date_joined,
        delivery_email=delivery_email,
        email=email,
        full_name=full_name,
        id=id,
        is_active=is_active,
        role=role,
        is_bot=is_bot,
        bot_type=bot_type,
        is_mirror_dummy=is_mirror_dummy,
        realm_id=realm_id,
        short_name=short_name,
        timezone=timezone,
    )


def convert_user_data(
    user_handler: UserHandler,
    user_id_mapper: IdMapper[str],
    user_data_map: dict[str, dict[str, Any]],
    realm: ZerverFieldsT,
    realm_id: int,
    team_name: str,
) -> None:
    has_owner = False
    for user_data in user_data_map.values():
        if check_user_in_team(user_data, team_name) or user_data["is_mirror_dummy"]:
            user = process_user(user_data, realm_id, team_name, user_id_mapper)

            user_handler.add_user(user)
            if user["role"] == UserProfile.ROLE_REALM_OWNER:
                has_owner = True
    if not has_owner:
        logging.warning("Converted realm has no owners!")
    user_handler.validate_user_emails()


MATTERMOST_DEFAULT_ANNOUNCEMENTS_CHANNEL_NAME = "Town Square"


def convert_channel_data(
    realm: ZerverFieldsT,
    channel_data: list[ZerverFieldsT],
    user_data_map: dict[str, dict[str, Any]],
    subscriber_handler: SubscriberHandler[frozenset[str]],
    stream_id_mapper: IdMapper[str],
    user_id_mapper: IdMapper[str],
    realm_id: int,
    team_name: str,
) -> AddedChannelsT:
    zerver_realm = realm["zerver_realm"]

    channel_data_list = [d for d in channel_data if d["team"] == team_name]

    channel_members_map: dict[str, list[str]] = {}
    channel_admins_map: dict[str, list[str]] = {}
    added_channels: AddedChannelsT = {}

    def initialize_stream_membership_dicts() -> None:
        for channel in channel_data:
            mattermost_channel_id = channel["name"]
            channel_members_map[mattermost_channel_id] = []
            channel_admins_map[mattermost_channel_id] = []

        for username, user_dict in user_data_map.items():
            teams = user_dict["teams"]
            if user_dict["teams"] is None:
                continue

            for team in teams:
                if team["name"] != team_name:
                    continue
                for channel in team["channels"]:
                    channel_roles = channel["roles"]
                    mattermost_channel_id = channel["name"]
                    if "channel_admin" in channel_roles:
                        channel_admins_map[mattermost_channel_id].append(username)
                    elif "channel_user" in channel_roles:
                        channel_members_map[mattermost_channel_id].append(username)
                    elif "channel_guest" in channel_roles:
                        channel_members_map[mattermost_channel_id].append(username)

    def get_invite_only_value_from_channel_type(channel_type: str) -> bool:
        # Channel can have two types in Mattermost
        # "O" for a public channel.
        # "P" for a private channel.
        if channel_type == "O":
            return False
        elif channel_type == "P":
            return True
        else:  # nocoverage
            raise Exception("unexpected value")

    initialize_stream_membership_dicts()
    channel_name_count: dict[str, int] = {}
    for channel_dict in channel_data_list:
        now = int(timezone_now().timestamp())
        # The "name" field is Mattermost channel's unique identifier.
        mattermost_channel_id = channel_dict["name"]
        stream_id = stream_id_mapper.get(mattermost_channel_id)
        invite_only = get_invite_only_value_from_channel_type(channel_dict["type"])
        channel_display_name = truncate_content(
            channel_dict["display_name"], Stream.MAX_NAME_LENGTH, "…"
        )

        # Channel names which only differ in capitalization are a collision,
        # because we treat them case-insensitively at the database level.
        # So we use .lower() of the name for collision-detection purposes.
        channel_display_name_lower = channel_display_name.lower()
        while channel_display_name_lower in channel_name_count:
            channel_name_count[channel_display_name_lower] += 1
            collision = channel_name_count[channel_display_name_lower]
            count_string = f" ({collision})"

            channel_display_name = (
                truncate_content(
                    channel_display_name, Stream.MAX_NAME_LENGTH - len(count_string), "…"
                )
                + count_string
            )
            # The new generated name could still collide with a pre-existing
            # imported channel.  E.g. channels ["Foo (2)", "foo", "Foo"] will
            # have a collision here while processing the "Foo" name.  This
            # should be very rare, so we opt for an approach that keeps this
            # code simple over prioritizing prettiness of the generated name -
            # so "Foo" will be transformed into "Foo (2) (2)" in this rare case.
            channel_display_name_lower = channel_display_name.lower()
        channel_name_count[channel_display_name_lower] = 1

        realm["zerver_stream"].append(
            build_stream(
                date_created=now,
                realm_id=realm_id,
                name=channel_display_name,
                # Purpose describes how the channel should be used. It is similar to
                # stream description and is shown in channel list to help others decide
                # whether to join.
                # Header text always appears right next to channel name in channel header.
                # Can be used for advertising the purpose of stream, making announcements as
                # well as including frequently used links. So probably not a bad idea to use
                # this as description if the channel purpose is empty.
                description=channel_dict["purpose"] or channel_dict["header"],
                stream_id=stream_id,
                # Mattermost export don't include data of archived(~ deactivated) channels.
                deactivated=False,
                invite_only=invite_only,
            )
        )

        channel_users = {
            *(
                user_id_mapper.get(username)
                for username in channel_admins_map[mattermost_channel_id]
            ),
            *(
                user_id_mapper.get(username)
                for username in channel_members_map[mattermost_channel_id]
            ),
        }

        subscriber_handler.set_info(
            users=channel_users,
            stream_id=stream_id,
        )

        recipient_id = NEXT_ID("recipient")
        recipient = build_recipient(stream_id, recipient_id, Recipient.STREAM)
        realm["zerver_recipient"].append(recipient)

        if channel_dict["display_name"] == MATTERMOST_DEFAULT_ANNOUNCEMENTS_CHANNEL_NAME:
            zerver_realm[0]["new_stream_announcements_stream"] = stream_id
            zerver_realm[0]["zulip_update_announcements_stream"] = stream_id

        added_channels[mattermost_channel_id] = ChannelMetadata(
            zulip_channel_id=stream_id, zulip_recipient_id=recipient_id
        )
    return added_channels


def convert_direct_message_group_data(
    direct_message_group_data: list[ZerverFieldsT],
    user_data_map: dict[str, dict[str, Any]],
    subscriber_handler: SubscriberHandler[frozenset[str]],
    direct_message_group_id_mapper: IdMapper[frozenset[str]],
    user_id_mapper: IdMapper[str],
    realm: ZerverFieldsT,
    realm_id: int,
    team_name: str,
) -> list[ZerverFieldsT]:
    zerver_direct_message_group = []
    for direct_message_group in direct_message_group_data:
        group_member_usernames = (
            direct_message_group["members"]
            if "members" in direct_message_group
            else {
                participant_data["username"]
                for participant_data in direct_message_group["participants"]
            }
        )
        direct_message_group_members = frozenset(
            {username for username in group_member_usernames if user_id_mapper.has(username)}
        )

        if subscriber_handler.get_zulip_recipient_id(direct_message_group_members):
            logging.info("Duplicate direct message group found in the export data. Skipping.")
            continue
        direct_message_group_id = direct_message_group_id_mapper.get(direct_message_group_members)
        direct_message_group_dict = build_direct_message_group(
            direct_message_group_id, len(direct_message_group_members)
        )
        direct_message_group_user_ids = {
            user_id_mapper.get(username) for username in direct_message_group_members
        }
        subscriber_handler.set_info(
            users=direct_message_group_user_ids,
            direct_message_group_id=direct_message_group_id,
        )

        recipient_id = NEXT_ID("recipient")
        recipient = Recipient(
            type_id=direct_message_group_id,
            id=recipient_id,
            type=Recipient.DIRECT_MESSAGE_GROUP,
        )
        recipient_dict = model_to_dict(recipient)
        realm["zerver_recipient"].append(recipient_dict)

        subscriber_handler.add_group_dm_key_to_zulip_recipient_id(
            key=direct_message_group_members,
            group_recipient_id=recipient_id,
        )
        zerver_direct_message_group.append(direct_message_group_dict)
    return zerver_direct_message_group


def build_reactions(
    realm_id: int,
    total_reactions: list[ZerverFieldsT],
    reactions: list[ZerverFieldsT],
    message_id: int,
    user_id_mapper: IdMapper[str],
    zerver_realmemoji: list[ZerverFieldsT],
) -> None:
    realmemoji = {}
    for realm_emoji in zerver_realmemoji:
        realmemoji[realm_emoji["name"]] = realm_emoji["id"]

    # For the Unicode emoji codes, we use equivalent of
    # function 'get_emoji_data' in 'zerver/lib/emoji' here
    for mattermost_reaction in reactions:
        emoji_name = mattermost_reaction["emoji_name"]
        username = mattermost_reaction["user"]
        # Check in Unicode emoji
        if emoji_name in name_to_codepoint:
            emoji_code = name_to_codepoint[emoji_name]
            reaction_type = Reaction.UNICODE_EMOJI
        # Check in realm emoji
        elif emoji_name in realmemoji:
            emoji_code = realmemoji[emoji_name]
            reaction_type = Reaction.REALM_EMOJI
        else:  # nocoverage
            continue

        if not user_id_mapper.has(username):
            continue

        reaction_id = NEXT_ID("reaction")
        reaction = Reaction(
            id=reaction_id,
            emoji_code=emoji_code,
            emoji_name=emoji_name,
            reaction_type=reaction_type,
        )

        reaction_dict = model_to_dict(reaction, exclude=["message", "user_profile"])
        reaction_dict["message"] = message_id
        reaction_dict["user_profile"] = user_id_mapper.get(username)
        total_reactions.append(reaction_dict)


def get_mentioned_user_ids(raw_message: dict[str, Any], user_id_mapper: IdMapper[str]) -> set[int]:
    user_ids = set()
    content = raw_message["content"]

    # usernames can be of the form user.name, user_name, username., username_, user.name_ etc
    matches = re.findall(r"(?<=^|(?<=[^a-zA-Z0-9-_.]))@(([A-Za-z0-9]+[_.]?)+)", content)

    for match in matches:
        possible_username = match[0]
        if user_id_mapper.has(possible_username):
            user_ids.add(user_id_mapper.get(possible_username))
    return user_ids


def process_message_attachments(
    attachments: list[dict[str, Any]],
    realm_id: int,
    message_id: int,
    user_id: int,
    zerver_attachment: list[AttachmentRecordData],
    uploads_list: list[UploadRecordData],
    mattermost_data_dir: str,
    output_dir: str,
) -> tuple[str, bool]:
    has_image = False

    markdown_links = []

    for attachment in attachments:
        attachment_path = attachment["path"]
        data_base_dir = os.path.join(mattermost_data_dir, "data")

        try:
            _, attachment_safe_full_path = validate_and_resolve_relative_path(
                attachment_path,
                base_dir=data_base_dir,
                safe_base_dir=os.path.realpath(data_base_dir),
                field_name_for_error="path",
            )
        except FileNotFoundError:
            logging.info("Message attachment file not found: '%s'; continuing.", attachment_path)
            continue

        file_name = attachment_path.split("/")[-1]
        file_ext = f".{file_name.split('.')[-1]}"

        if file_ext.lower() in IMAGE_EXTENSIONS:
            has_image = True

        attachment_data = get_attachment_path_and_content(
            link_name=file_name, filename=file_name, realm_id=realm_id
        )

        markdown_links.append(attachment_data.markdown_link)

        fileinfo = {
            "name": file_name,
            "size": os.path.getsize(attachment_safe_full_path),
            "created": os.path.getmtime(attachment_safe_full_path),
        }

        uploads_list.append(
            UploadRecordData(
                content_type=None,
                last_modified=fileinfo["created"],
                path=attachment_data.path_id,
                realm_id=realm_id,
                s3_path=attachment_data.path_id,
                size=fileinfo["size"],
                user_profile_id=user_id,
            )
        )

        build_attachment(
            realm_id=realm_id,
            message_ids={message_id},
            user_id=user_id,
            fileinfo=fileinfo,
            s3_path=attachment_data.path_id,
            zerver_attachment=zerver_attachment,
        )

        # Copy the attachment file to output_dir
        attachment_out_path = os.path.join(output_dir, "uploads", attachment_data.path_id)
        os.makedirs(os.path.dirname(attachment_out_path), exist_ok=True)
        shutil.copyfile(attachment_safe_full_path, attachment_out_path)

    content = "\n".join(markdown_links)

    return content, has_image


def process_raw_message_batch(
    realm_id: int,
    raw_messages: list[dict[str, Any]],
    subscriber_map: dict[int, set[int]],
    user_id_mapper: IdMapper[str],
    user_handler: UserHandler,
    added_channels: AddedChannelsT,
    subscriber_handler: SubscriberHandler[frozenset[str]],
    is_pm_data: bool,
    output_dir: str,
    zerver_realmemoji: list[dict[str, Any]],
    total_reactions: list[dict[str, Any]],
    uploads_list: list[UploadRecordData],
    zerver_attachment: list[AttachmentRecordData],
    mattermost_data_dir: str,
) -> None:
    def fix_mentions(content: str, mention_user_ids: set[int]) -> str:
        for user_id in mention_user_ids:
            user = user_handler.get_user(user_id=user_id)
            mattermost_mention = "@{short_name}".format(**user)
            zulip_mention = "@**{full_name}**".format(**user)
            content = content.replace(mattermost_mention, zulip_mention)

        content = content.replace("@channel", "@**all**")
        content = content.replace("@all", "@**all**")
        # We don't have an equivalent for Mattermost's @here mention which mentions all users
        # online in the channel.
        content = content.replace("@here", "@**all**")
        return content

    mention_map: dict[int, set[int]] = {}
    zerver_message = []

    for raw_message in raw_messages:
        message_id = NEXT_ID("message")
        mention_user_ids = get_mentioned_user_ids(raw_message, user_id_mapper)
        mention_map[message_id] = mention_user_ids

        content = fix_mentions(
            content=raw_message["content"],
            mention_user_ids=mention_user_ids,
        )

        date_sent = raw_message["date_sent"]
        sender_user_id = raw_message["sender_id"]
        if "channel_name" in raw_message:
            is_direct_message_type = False
            recipient_id = added_channels[raw_message["channel_name"]].zulip_recipient_id
        elif "direct_message_group_members" in raw_message:
            is_direct_message_type = True
            if group_recipient_id := subscriber_handler.get_zulip_recipient_id(
                raw_message["direct_message_group_members"]
            ):
                recipient_id = group_recipient_id
            else:  # nocoverage
                # Theoretically this happens when none of the group dm participants get
                # converted.
                logging.info(
                    "Skipped a group direct message since none of the participants got converted. Participants: %s",
                    str(raw_message["direct_message_group_members"]),
                )
                continue
        else:
            raise AssertionError(
                "raw_message without channel_name, direct_message_group_members or pm_members key"
            )

        rendered_content = None

        has_attachment = False
        has_image = False
        has_link = False
        if "attachments" in raw_message:
            has_attachment = True
            has_link = True

            attachment_markdown, has_image = process_message_attachments(
                attachments=raw_message["attachments"],
                realm_id=realm_id,
                message_id=message_id,
                user_id=sender_user_id,
                zerver_attachment=zerver_attachment,
                uploads_list=uploads_list,
                mattermost_data_dir=mattermost_data_dir,
                output_dir=output_dir,
            )

            if content:
                content += "\n\n"
            content += attachment_markdown

        topic_name = "imported from mattermost"

        message = build_message(
            content=content,
            message_id=message_id,
            date_sent=date_sent,
            recipient_id=recipient_id,
            realm_id=realm_id,
            rendered_content=rendered_content,
            topic_name=topic_name,
            user_id=sender_user_id,
            is_channel_message=not is_pm_data,
            has_image=has_image,
            has_link=has_link,
            has_attachment=has_attachment,
            is_direct_message_type=is_direct_message_type,
        )
        zerver_message.append(message)
        build_reactions(
            realm_id,
            total_reactions,
            raw_message["reactions"],
            message_id,
            user_id_mapper,
            zerver_realmemoji,
        )

    zerver_usermessage = make_user_messages(
        zerver_message=zerver_message,
        subscriber_map=subscriber_map,
        mention_map=mention_map,
    )

    message_json = dict(
        zerver_message=zerver_message,
        zerver_usermessage=zerver_usermessage,
    )

    dump_file_id = NEXT_ID("dump_file_id" + str(realm_id))
    message_file = f"/messages-{dump_file_id:06}.json"
    create_converted_data_files(message_json, output_dir, message_file)


def process_posts(
    num_teams: int,
    team_name: str,
    realm_id: int,
    post_data: list[dict[str, Any]],
    added_channels: AddedChannelsT,
    subscriber_handler: SubscriberHandler[frozenset[str]],
    subscriber_map: dict[int, set[int]],
    output_dir: str,
    is_pm_data: bool,
    masking_content: bool,
    user_id_mapper: IdMapper[str],
    user_handler: UserHandler,
    zerver_realmemoji: list[dict[str, Any]],
    total_reactions: list[dict[str, Any]],
    uploads_list: list[UploadRecordData],
    zerver_attachment: list[AttachmentRecordData],
    mattermost_data_dir: str,
) -> None:
    post_data_list = []
    for post in post_data:
        if "team" not in post:
            # Mattermost doesn't specify a team for direct messages
            # in its export format.  This line of code requires that
            # we only be importing data from a single team (checked
            # elsewhere) -- we just assume it's the target team.
            post_team = team_name
        else:
            post_team = post["team"]
        if post_team == team_name:
            post_data_list.append(post)

    def message_to_dict(post_dict: dict[str, Any]) -> dict[str, Any] | None:
        sender_username = post_dict["user"]
        if user_id_mapper.has(sender_username):
            sender_id = user_id_mapper.get(sender_username)
        else:  # nocoverage
            return None
        content = post_dict["message"]

        if masking_content:
            content = re.sub(r"[a-z]", "x", content)
            content = re.sub(r"[A-Z]", "X", content)

        if "reactions" in post_dict:
            reactions = post_dict["reactions"] or []
        else:
            reactions = []

        message_dict = dict(
            sender_id=sender_id,
            content=content,
            date_sent=int(post_dict["create_at"] / 1000),
            reactions=reactions,
        )
        if "channel" in post_dict:
            message_dict["channel_name"] = post_dict["channel"]
        elif "channel_members" in post_dict:
            # This case is for handling posts from direct messages and direct message,
            # groups not channels. Direct messages and direct message groups are known
            # as direct_channels in Slack and hence the name channel_members.
            channel_members: list[str] = [
                username
                for username in post_dict["channel_members"]
                if user_id_mapper.has(username)
            ]
            message_dict["direct_message_group_members"] = frozenset(channel_members)
        else:
            raise AssertionError("Post without channel or channel_members key.")

        if post_dict.get("attachments"):
            message_dict["attachments"] = post_dict["attachments"]

        return message_dict

    raw_messages = []
    for post_dict in post_data_list:
        message_dict = message_to_dict(post_dict)
        if message_dict is None:  # nocoverage
            continue
        raw_messages.append(message_dict)
        message_replies = post_dict["replies"]
        # Replies to a message in Mattermost are stored in the main message object.
        # For now, we just append the replies immediately after the original message.
        if message_replies is not None:
            for reply in message_replies:
                if "channel" in post_dict:
                    reply["channel"] = post_dict["channel"]
                else:  # nocoverage
                    reply["channel_members"] = post_dict["channel_members"]
                reply_dict = message_to_dict(reply)
                if reply_dict is None:  # nocoverage
                    continue
                raw_messages.append(reply_dict)

    def process_batch(lst: list[dict[str, Any]]) -> None:
        process_raw_message_batch(
            realm_id=realm_id,
            raw_messages=lst,
            subscriber_map=subscriber_map,
            user_id_mapper=user_id_mapper,
            user_handler=user_handler,
            added_channels=added_channels,
            subscriber_handler=subscriber_handler,
            is_pm_data=is_pm_data,
            output_dir=output_dir,
            zerver_realmemoji=zerver_realmemoji,
            total_reactions=total_reactions,
            uploads_list=uploads_list,
            zerver_attachment=zerver_attachment,
            mattermost_data_dir=mattermost_data_dir,
        )

    chunk_size = 1000

    process_list_in_batches(
        lst=raw_messages,
        chunk_size=chunk_size,
        process_batch=process_batch,
    )


def write_message_data(
    num_teams: int,
    team_name: str,
    realm_id: int,
    post_data: dict[str, list[dict[str, Any]]],
    zerver_recipient: list[ZerverFieldsT],
    subscriber_map: dict[int, set[int]],
    output_dir: str,
    masking_content: bool,
    added_channels: AddedChannelsT,
    subscriber_handler: SubscriberHandler[frozenset[str]],
    user_id_mapper: IdMapper[str],
    user_handler: UserHandler,
    zerver_realmemoji: list[dict[str, Any]],
    total_reactions: list[dict[str, Any]],
    uploads_list: list[UploadRecordData],
    zerver_attachment: list[AttachmentRecordData],
    mattermost_data_dir: str,
) -> None:
    if num_teams == 1:
        post_types = ["channel_post", "direct_post"]
    else:
        post_types = ["channel_post"]
        logging.warning(
            "Skipping importing direct message groups and DMs since there are multiple teams in the export"
        )

    for post_type in post_types:
        process_posts(
            num_teams=num_teams,
            team_name=team_name,
            realm_id=realm_id,
            post_data=post_data[post_type],
            added_channels=added_channels,
            subscriber_handler=subscriber_handler,
            subscriber_map=subscriber_map,
            output_dir=output_dir,
            is_pm_data=post_type == "direct_post",
            masking_content=masking_content,
            user_id_mapper=user_id_mapper,
            user_handler=user_handler,
            zerver_realmemoji=zerver_realmemoji,
            total_reactions=total_reactions,
            uploads_list=uploads_list,
            zerver_attachment=zerver_attachment,
            mattermost_data_dir=mattermost_data_dir,
        )


def write_emoticon_data(
    realm_id: int, custom_emoji_data: list[dict[str, Any]], data_dir: str, output_dir: str
) -> list[ZerverFieldsT]:
    """
    This function does most of the work for processing emoticons, the bulk
    of which is copying files.  We also write a json file with metadata.
    Finally, we return a list of RealmEmoji dicts to our caller.

    In our data_dir we have a pretty simple setup:

        The exported JSON file will have emoji rows if it contains any custom emoji
            {
                "type": "emoji",
                "emoji": {"name": "peerdium", "image": "exported_emoji/h15ni7kf1bnj7jeua4qhmctsdo/image"}
            }
            {
                "type": "emoji",
                "emoji": {"name": "tick", "image": "exported_emoji/7u7x8ytgp78q8jir81o9ejwwnr/image"}
            }

        exported_emoji/ - contains a bunch of image files:
            exported_emoji/7u7x8ytgp78q8jir81o9ejwwnr/image
            exported_emoji/h15ni7kf1bnj7jeua4qhmctsdo/image

    We move all the relevant files to Zulip's more nested
    directory structure.
    """

    logging.info("Starting to process emoticons")

    flat_data = [
        dict(
            path=d["image"],
            name=d["name"],
        )
        for d in custom_emoji_data
    ]

    emoji_folder = os.path.join(output_dir, "emoji")
    os.makedirs(emoji_folder, exist_ok=True)

    def process(data: ZerverFieldsT) -> ZerverFieldsT | None:
        source_sub_path = data["path"]

        try:
            _, safe_source_path = validate_and_resolve_relative_path(
                source_sub_path,
                base_dir=data_dir,
                safe_base_dir=os.path.realpath(data_dir),
                field_name_for_error="path",
            )
        except FileNotFoundError:
            logging.info("Emoji file no found: '%s'; continuing.", source_sub_path)
            return None

        target_fn = sanitize_name(data["name"])
        target_sub_path = RealmEmoji.PATH_ID_TEMPLATE.format(
            realm_id=realm_id,
            emoji_file_name=target_fn,
        )
        target_path = os.path.join(emoji_folder, target_sub_path)

        os.makedirs(os.path.dirname(target_path), exist_ok=True)

        target_path = os.path.abspath(target_path)

        shutil.copyfile(safe_source_path, target_path)

        return dict(
            path=target_sub_path,
            s3_path=target_sub_path,
            file_name=target_fn,
            realm_id=realm_id,
            name=data["name"],
        )

    emoji_records = [record for data in flat_data if (record := process(data))]

    create_converted_data_files(emoji_records, output_dir, "/emoji/records.json")

    realmemoji = [
        build_realm_emoji(
            realm_id=realm_id,
            name=rec["name"],
            id=NEXT_ID("realmemoji"),
            file_name=rec["file_name"],
        )
        for rec in emoji_records
    ]
    logging.info("Done processing emoticons")

    return realmemoji


def create_username_to_user_mapping(
    user_data_list: list[dict[str, Any]],
) -> dict[str, dict[str, Any]]:
    username_to_user = {}
    for user in user_data_list:
        username_to_user[user["username"]] = user
    return username_to_user


def check_user_in_team(user: dict[str, Any], team_name: str) -> bool:
    if user["teams"] is None:
        # This is null for users not on any team
        return False
    return any(team["name"] == team_name for team in user["teams"])


def backfill_user_data_from_posts(
    num_teams: int,
    team_name: str,
    mattermost_data: dict[str, Any],
    username_to_user: dict[str, dict[str, Any]],
) -> None:
    # This function might looks like a great place to label admin users. But
    # that won't be fully correct since we are iterating only though posts and
    # it covers only users that have sent at least one message.

    def backfill_user_data(post: dict[str, Any]) -> None:
        user_data = username_to_user.get(post["user"])
        if user_data is None:  # nocoverage
            # This is probably deleted user, we currently don't support converting
            # this type of user yet.
            return
        if user_data.get("is_bot"):
            # The export data doesn't state which teams or channels a bot user
            # is a member of. So, they'll only be assigned to channels and teams
            # where their messages are found.
            if "team" not in post:
                return

            if not (
                bot_team_data := next(
                    (team for team in user_data["teams"] if team["name"] == post["team"]), None
                )
            ):
                user_data["teams"].append(
                    {
                        "name": post["team"],
                        "roles": "team_user",
                        "channels": [],
                    }
                )
                bot_team_data = user_data["teams"][-1]

            if "channel" in post and all(
                channel["name"] != post["channel"] for channel in bot_team_data["channels"]
            ):
                bot_team_data["channels"].append(
                    {
                        "name": post["channel"],
                        "roles": "channel_user",
                    }
                )
            return

        if not check_user_in_team(user_data, team_name):
            user_data["is_mirror_dummy"] = True

    for post in mattermost_data["post"]["channel_post"]:
        post_team = post["team"]
        if post_team == team_name:
            backfill_user_data(post)

    if num_teams == 1:
        for post in mattermost_data["post"]["direct_post"]:
            assert "team" not in post
            backfill_user_data(post)


def reset_mirror_dummy_users(username_to_user: dict[str, dict[str, Any]]) -> None:
    for user in username_to_user.values():
        user["is_mirror_dummy"] = False


DEFAULT_SINGLE_TEAM_OBJECT = {
    "name": "zulip-1",
    "display_name": "Converted Mattermost teams",
    "type": "O",
    "description": "This organization contains multiple Mattermost teams.",
    "allow_open_invite": False,
}

COMPILED_CHANNEL_ID_FORMAT = "{id}-{team}"


def mattermost_data_file_to_dict(
    mattermost_data_file: str, combine_into_one_realm: bool = False
) -> dict[str, Any]:
    # If combine_into_one_realm=True, we want to make sure these things
    # happen here:
    #   1. The only mattermost team object is DEFAULT_SINGLE_TEAM_OBJECT
    #
    #   2. Update all objects to have DEFAULT_SINGLE_TEAM_OBJECT as the team
    #      it is associated to.
    #
    #   3. Since Mattermost objects uses the "name" field as their unique ID,
    #      mark all objects' "name" field with their original "team" ID to
    #      make sure they stay unique when combined.

    mattermost_data: dict[str, Any] = {}
    mattermost_data["version"] = []
    mattermost_data["team"] = []
    mattermost_data["channel"] = []
    mattermost_data["user"] = []
    mattermost_data["post"] = {"channel_post": [], "direct_post": []}
    mattermost_data["emoji"] = []
    mattermost_data["direct_channel"] = []
    mattermost_data["role"] = []

    if combine_into_one_realm:
        mattermost_data["team"] = [DEFAULT_SINGLE_TEAM_OBJECT]

    with open(mattermost_data_file, "rb") as fp:
        for line in fp:
            row = orjson.loads(line)
            data_type = row["type"]
            if data_type == "post":
                post = row["post"]
                if combine_into_one_realm:
                    post["channel"] = COMPILED_CHANNEL_ID_FORMAT.format(
                        id=post["channel"], team=post["team"]
                    )
                    post["team"] = DEFAULT_SINGLE_TEAM_OBJECT["name"]
                mattermost_data["post"]["channel_post"].append(post)
            elif data_type == "direct_post":
                mattermost_data["post"]["direct_post"].append(row["direct_post"])
            elif data_type == "bot":
                bot_data = row["bot"]
                bot_data["is_bot"] = True
                bot_data["first_name"] = ""
                bot_data["last_name"] = ""
                if "email" not in bot_data:
                    bot_username = bot_data["username"]
                    bot_data["email"] = ImportedBotEmail.get_email(
                        bot_data,
                        get_domain_name_for_import(),
                        # Mattermost exports don't provide a nice ID, so we have to do
                        # with the username field.
                        bot_id=slugify(bot_username),
                        bot_name_getter=lambda d: d["username"],
                    )
                # Exported bot data doesn't originally include which teams or channels they are a
                # part of. This data will be backfilled later.
                # In the special case of combine_into_one_realm=True, we fill this in right now,
                # setting DEFAULT_SINGLE_TEAM_OBJECT as the team for each bot.
                bot_data["teams"] = (
                    []
                    if not combine_into_one_realm
                    else [
                        {
                            "name": DEFAULT_SINGLE_TEAM_OBJECT["name"],
                            "roles": "team_user",
                            "channels": [],
                        }
                    ]
                )
                mattermost_data["user"].append(bot_data)
            elif data_type == "user" and combine_into_one_realm:
                if not row[data_type]["teams"]:
                    # "teams" is null for users who are not a member of any
                    # team. Keep it as is, so that such users are treated
                    # just like in the non-combined code path: they only get
                    # a mirror dummy account, backfilled from their posts.
                    mattermost_data[data_type].append(row[data_type])
                    continue
                all_user_channels: list[dict[str, Any]] = []
                all_user_team_roles: set[str] = set()
                # Admin users in individual teams will have administrator role
                # in the combined realm.
                for team in row[data_type]["teams"]:
                    for channel in team["channels"]:
                        channel["name"] = COMPILED_CHANNEL_ID_FORMAT.format(
                            id=channel["name"], team=team["name"]
                        )
                    all_user_channels += team["channels"]
                    all_user_team_roles.update(team["roles"].split(" "))
                # If a user is a guest in a team but a normal user in at least one
                # other team, they will be converted into a normal user in the combined
                # realm.
                if {"team_guest", "team_user"}.issubset(all_user_team_roles):
                    all_user_team_roles.remove("team_guest")
                row[data_type]["teams"] = [
                    {
                        "name": DEFAULT_SINGLE_TEAM_OBJECT["name"],
                        "roles": " ".join(all_user_team_roles),
                        "channels": all_user_channels,
                    }
                ]
                mattermost_data[data_type].append(row[data_type])
            elif data_type == "team" and combine_into_one_realm:
                continue
            elif data_type == "channel" and combine_into_one_realm:
                row[data_type]["name"] = COMPILED_CHANNEL_ID_FORMAT.format(
                    id=row[data_type]["name"], team=row[data_type]["team"]
                )
                row[data_type]["team"] = DEFAULT_SINGLE_TEAM_OBJECT["name"]
                mattermost_data[data_type].append(row[data_type])
            else:
                if (
                    combine_into_one_realm
                    and isinstance(row[data_type], dict)
                    and "team" in row[data_type]
                ):
                    raise AssertionError(
                        f"Found unexpected '{data_type}' object while compiling into combined realm. {row}"
                    )
                mattermost_data[data_type].append(row[data_type])
    return mattermost_data


def do_convert_data(
    mattermost_data_dir: str,
    output_dir: str,
    masking_content: bool,
    combine_into_one_realm: bool = False,
) -> None:
    username_to_user: dict[str, dict[str, Any]] = {}

    os.makedirs(output_dir, exist_ok=True)
    if os.listdir(output_dir):  # nocoverage
        raise Exception("Output directory should be empty!")

    # Older versions of Mattermost CLI generate export.json while newer versions of
    # mmctl generate import.jsonl.
    import_jsonl_file = os.path.join(mattermost_data_dir, "import.jsonl")
    export_json_file = os.path.join(mattermost_data_dir, "export.json")

    if os.path.exists(import_jsonl_file):
        mattermost_data = mattermost_data_file_to_dict(import_jsonl_file, combine_into_one_realm)
    elif os.path.exists(export_json_file):
        mattermost_data = mattermost_data_file_to_dict(export_json_file, combine_into_one_realm)
    else:
        raise AssertionError(
            f"Missing import.jsonl or export.json file in {mattermost_data_dir}. Files: {os.listdir(mattermost_data_dir)!s}",
        )

    username_to_user = create_username_to_user_mapping(mattermost_data["user"])

    for team in mattermost_data["team"]:
        realm_id = NEXT_ID("realm_id")
        team_name = team["name"]

        user_handler = UserHandler()
        subscriber_handler = SubscriberHandler[frozenset[str]]()
        user_id_mapper = IdMapper[str]()
        stream_id_mapper = IdMapper[str]()
        direct_message_group_id_mapper = IdMapper[frozenset[str]]()

        logging.info("Generating data for %s", team_name)
        realm = make_realm(realm_id, team)
        realm_output_dir = os.path.join(output_dir, team_name)

        reset_mirror_dummy_users(username_to_user)
        backfill_user_data_from_posts(
            len(mattermost_data["team"]), team_name, mattermost_data, username_to_user
        )

        convert_user_data(
            user_handler=user_handler,
            user_id_mapper=user_id_mapper,
            user_data_map=username_to_user,
            realm=realm,
            realm_id=realm_id,
            team_name=team_name,
        )

        added_channels = convert_channel_data(
            realm=realm,
            channel_data=mattermost_data["channel"],
            user_data_map=username_to_user,
            subscriber_handler=subscriber_handler,
            stream_id_mapper=stream_id_mapper,
            user_id_mapper=user_id_mapper,
            realm_id=realm_id,
            team_name=team_name,
        )

        zerver_direct_message_group: list[ZerverFieldsT] = []
        if len(mattermost_data["team"]) == 1:
            zerver_direct_message_group = convert_direct_message_group_data(
                direct_message_group_data=mattermost_data["direct_channel"],
                user_data_map=username_to_user,
                subscriber_handler=subscriber_handler,
                direct_message_group_id_mapper=direct_message_group_id_mapper,
                user_id_mapper=user_id_mapper,
                realm=realm,
                realm_id=realm_id,
                team_name=team_name,
            )
            realm["zerver_huddle"] = zerver_direct_message_group

        stream_subscriptions = build_stream_subscriptions(
            get_users=subscriber_handler.get_users,
            zerver_recipient=realm["zerver_recipient"],
            zerver_stream=realm["zerver_stream"],
        )

        direct_message_group_subscriptions = build_direct_message_group_subscriptions(
            get_users=subscriber_handler.get_users,
            zerver_recipient=realm["zerver_recipient"],
            zerver_direct_message_group=zerver_direct_message_group,
        )

        zerver_subscription = stream_subscriptions + direct_message_group_subscriptions
        realm["zerver_subscription"] = zerver_subscription

        zerver_realmemoji = write_emoticon_data(
            realm_id=realm_id,
            custom_emoji_data=mattermost_data["emoji"],
            data_dir=mattermost_data_dir,
            output_dir=realm_output_dir,
        )
        realm["zerver_realmemoji"] = zerver_realmemoji

        subscriber_map = make_subscriber_map(
            zerver_subscription=zerver_subscription,
        )

        total_reactions: list[dict[str, Any]] = []
        uploads_list: list[UploadRecordData] = []
        zerver_attachment: list[AttachmentRecordData] = []

        write_message_data(
            num_teams=len(mattermost_data["team"]),
            team_name=team_name,
            realm_id=realm_id,
            post_data=mattermost_data["post"],
            zerver_recipient=realm["zerver_recipient"],
            subscriber_map=subscriber_map,
            output_dir=realm_output_dir,
            masking_content=masking_content,
            added_channels=added_channels,
            subscriber_handler=subscriber_handler,
            user_id_mapper=user_id_mapper,
            user_handler=user_handler,
            zerver_realmemoji=zerver_realmemoji,
            total_reactions=total_reactions,
            uploads_list=uploads_list,
            zerver_attachment=zerver_attachment,
            mattermost_data_dir=mattermost_data_dir,
        )
        realm["zerver_reaction"] = total_reactions
        realm["zerver_userprofile"] = user_handler.get_all_users()
        realm["sort_by_date"] = True

        create_converted_data_files(realm, realm_output_dir, "/realm.json")
        # Mattermost currently doesn't support exporting avatars
        create_converted_data_files([], realm_output_dir, "/avatars/records.json")

        # Export message attachments
        attachment: dict[str, list[Any]] = {"zerver_attachment": zerver_attachment}
        create_converted_data_files(attachment, realm_output_dir, "/attachment.json")
        create_converted_data_files(uploads_list, realm_output_dir, "/uploads/records.json")

        do_common_export_processes(realm_output_dir)