ForcePilot/backend/package/yuxi/channel/extensions/bluebubbles/chat.py

import logging

from yuxi.channel.extensions.bluebubbles.client import BlueBubblesClient
from yuxi.channel.extensions.bluebubbles.outbound import (
    mark_chat_read,
    mark_chat_unread,
    send_typing_indicator,
    stop_typing_indicator,
    edit_message,
    unsend_message,
    rename_group,
    set_group_icon,
    remove_group_icon,
    add_participant,
    remove_participant,
    leave_group,
    delete_chat,
    delete_message,
    force_notify,
    send_multipart_message,
    schedule_message,
    list_scheduled_messages,
    update_scheduled_message,
    delete_scheduled_message,
)

logger = logging.getLogger(__name__)


class ChatOperations:
    def __init__(self, client: BlueBubblesClient, private_api_available: bool = False):
        self.client = client
        self.private_api_available = private_api_available

    async def mark_read(self, chat_guid: str) -> dict:
        if not self.private_api_available:
            return {}
        return await mark_chat_read(self.client, chat_guid)

    async def mark_unread(self, chat_guid: str) -> dict:
        if not self.private_api_available:
            return {}
        return await mark_chat_unread(self.client, chat_guid)

    async def send_typing(self, chat_guid: str) -> dict:
        if not self.private_api_available:
            return {}
        return await send_typing_indicator(self.client, chat_guid)

    async def stop_typing(self, chat_guid: str) -> dict:
        if not self.private_api_available:
            return {}
        return await stop_typing_indicator(self.client, chat_guid)

    async def edit(self, chat_guid: str, message_guid: str, text: str) -> dict:
        if not self.private_api_available:
            raise RuntimeError("Private API required for message editing")
        return await edit_message(self.client, chat_guid, message_guid, text)

    async def unsend(self, chat_guid: str, message_guid: str) -> dict:
        if not self.private_api_available:
            raise RuntimeError("Private API required for message unsend")
        return await unsend_message(self.client, chat_guid, message_guid)

    async def rename_chat(self, chat_guid: str, name: str) -> dict:
        if not self.private_api_available:
            raise RuntimeError("Private API required for group rename")
        return await rename_group(self.client, chat_guid, name)

    async def set_icon(self, chat_guid: str, icon_path: str) -> dict:
        if not self.private_api_available:
            raise RuntimeError("Private API required for group icon")
        return await set_group_icon(self.client, chat_guid, icon_path)

    async def remove_icon(self, chat_guid: str) -> dict:
        if not self.private_api_available:
            raise RuntimeError("Private API required for group icon removal")
        return await remove_group_icon(self.client, chat_guid)

    async def add_member(self, chat_guid: str, address: str) -> dict:
        if not self.private_api_available:
            raise RuntimeError("Private API required for group management")
        return await add_participant(self.client, chat_guid, address)

    async def remove_member(self, chat_guid: str, address: str) -> dict:
        if not self.private_api_available:
            raise RuntimeError("Private API required for group management")
        return await remove_participant(self.client, chat_guid, address)

    async def leave(self, chat_guid: str) -> dict:
        if not self.private_api_available:
            raise RuntimeError("Private API required for group management")
        return await leave_group(self.client, chat_guid)

    async def delete(self, chat_guid: str) -> dict:
        if not self.private_api_available:
            raise RuntimeError("Private API required for chat deletion")
        return await delete_chat(self.client, chat_guid)

    async def delete_msg(self, chat_guid: str, message_guid: str) -> dict:
        if not self.private_api_available:
            raise RuntimeError("Private API required for message deletion")
        return await delete_message(self.client, chat_guid, message_guid)

    async def notify(self, message_guid: str) -> dict:
        if not self.private_api_available:
            raise RuntimeError("Private API required for force notify")
        return await force_notify(self.client, message_guid)

    async def send_multipart(
        self,
        chat_guid: str,
        text: str,
        file_paths: list[str],
        *,
        subject: str | None = None,
    ) -> dict:
        return await send_multipart_message(
            self.client,
            chat_guid,
            text,
            file_paths,
            private_api_available=self.private_api_available,
            subject=subject,
        )

    async def schedule(
        self,
        chat_guid: str,
        text: str,
        scheduled_at: str,
        *,
        subject: str | None = None,
    ) -> dict:
        return await schedule_message(
            self.client,
            chat_guid,
            text,
            scheduled_at,
            private_api_available=self.private_api_available,
            subject=subject,
        )

    async def list_scheduled(self) -> list[dict]:
        return await list_scheduled_messages(self.client)

    async def update_schedule(self, schedule_id: str, scheduled_at: str) -> dict:
        return await update_scheduled_message(self.client, schedule_id, scheduled_at)

    async def delete_schedule(self, schedule_id: str) -> dict:
        return await delete_scheduled_message(self.client, schedule_id)