dimsdk 2.4.0

Decentralized Instant Messaging SDK

Decentralized Instant Messaging (Python SDK)

Dependencies

• Latest Versions

Name	Version	Description
Ming Ke Ming (名可名)		Decentralized User Identity Authentication
Dao Ke Dao (道可道)		Universal Message Module
DIMP (去中心化通讯协议)		Decentralized Instant Messaging Protocol

Extensions

Content

extends CustomizedContent

ContentProcessor

class CustomizedContentProcessor(BaseContentProcessor):
    """
        Customized Content Processing Unit
        ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
        Handle content for application customized
    """

    # def __init__(self, facebook: Facebook, messenger: Messenger):
    #     super().__init__(facebook=facebook, messenger=messenger)

    # Override
    async def process_content(self, content: Content, r_msg: ReliableMessage) -> List[Content]:
        assert isinstance(content, CustomizedContent), 'customized content error: %s' % content
        ext = message_extensions()
        customized_filter = ext.customized_filter
        # get handler for 'app' & 'mod'
        handler = customized_filter.filter_content(content=content, msg=r_msg)
        return await handler.handle_action(content=content, msg=r_msg, messenger=self.messenger)


def message_extensions() -> Union[MessageExtensions, CustomizedFilterExtensions]:
    return shared_message_extensions

• CustomizedContentHandler

class CustomizedContentHandler(ABC):
    """
        Handler for Customized Content
        ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    """

    @abstractmethod
    async def handle_action(self, content: CustomizedContent, msg: ReliableMessage,
                            messenger: Messenger) -> List[Content]:
        """
        Do your job

        @param content:   customized content
        @param msg:       network message
        @param messenger: message transceiver
        @return contents
        """
        raise NotImplemented


class BaseCustomizedContentHandler(CustomizedContentHandler):
    """
        Default Handler
        ~~~~~~~~~~~~~~~
    """

    # Override
    async def handle_action(self, content: CustomizedContent, msg: ReliableMessage,
                            messenger: Messenger) -> List[Content]:
        # app = content.application
        app = content.get_str(key='app')
        mod = content.module
        act = content.action
        text = 'Content not support.'
        return self._respond_receipt(text=text, content=content, envelope=msg.envelope, extra={
            'template': 'Customized content (app: ${app}, mod: ${mod}, act: ${act}) not support yet!',
            'replacements': {
                'app': app,
                'mod': mod,
                'act': act,
            }
        })

    #
    #   Convenient responding
    #

    # noinspection PyMethodMayBeStatic
    def _respond_receipt(self, text: str, envelope: Envelope, content: Optional[Content],
                         extra: Optional[Dict] = None) -> List[ReceiptCommand]:
        return [
            # create base receipt command with text & original envelope
            BaseContentProcessor.create_receipt(text=text, envelope=envelope, content=content, extra=extra)
        ]

• CustomizedContentFilter

class CustomizedContentFilter(ABC):

    @abstractmethod
    def filter_content(self, content: CustomizedContent, msg: ReliableMessage) -> CustomizedContentHandler:
        raise NotImplemented


class AppCustomizedFilter(CustomizedContentFilter):

    def __init__(self):
        super().__init__()
        self.__default_handler = BaseCustomizedContentHandler()
        self.__handlers: Dict[str, CustomizedContentHandler] = {}

    def set_content_handler(self, app: str, mod: str, handler: CustomizedContentHandler):
        key = '%s:%s' % (app, mod)
        self.__handlers[key] = handler

    # protected
    def get_content_handler(self, app: str, mod: str) -> Optional[CustomizedContentHandler]:
        key = '%s:%s' % (app, mod)
        return self.__handlers.get(key)

    # Override
    def filter_content(self, content: CustomizedContent, msg: ReliableMessage) -> CustomizedContentHandler:
        # app = content.application
        app = content.get_str(key='app', default='')
        mod = content.module
        handler = self.get_content_handler(app=app, mod=mod)
        if handler is not None:
            return handler
        # if the application has too many modules, I suggest you to
        # use different handler to do the jobs for each module.
        return self.__default_handler


class CustomizedFilterExtensions:

    @property
    def customized_filter(self) -> CustomizedContentFilter:
        raise NotImplemented

    @customized_filter.setter
    def customized_filter(self, delegate: CustomizedContentFilter):
        raise NotImplemented


shared_message_extensions.customized_filter = AppCustomizedFilter()

• Example for group querying

class GroupHistoryHandler(BaseCustomizedContentHandler):
    """ Command Transform:

        +===============================+===============================+
        |      Customized Content       |      Group Query Command      |
        +-------------------------------+-------------------------------+
        |   "type" : i2s(0xCC)          |   "type" : i2s(0x88)          |
        |   "sn"   : 123                |   "sn"   : 123                |
        |   "time" : 123.456            |   "time" : 123.456            |
        |   "app"  : "chat.dim.group"   |                               |
        |   "mod"  : "history"          |                               |
        |   "act"  : "query"            |                               |
        |                               |   "command"   : "query"       |
        |   "group"     : "{GROUP_ID}"  |   "group"     : "{GROUP_ID}"  |
        |   "last_time" : 0             |   "last_time" : 0             |
        +===============================+===============================+
    """

    # Override
    async def handle_action(self, content: CustomizedContent, msg: ReliableMessage,
                            messenger: Messenger) -> List[Content]:
        if content.group is None:
            text = 'Group command error.'
            return self._respond_receipt(text=text, envelope=msg.envelope, content=content)
        act = content.action
        if act == GroupHistory.ACT_QUERY:
            # assert GroupHistory.APP == content.application
            assert GroupHistory.MOD == content.module
            return await self.__transform_query_command(content=content, msg=msg, messenger=messenger)
        else:
            # assert False, 'unknown action: %s, %s, sender: %s' % (act, content, sender)
            return await super().handle_action(content=content, msg=msg, messenger=messenger)

    async def __transform_query_command(self, content: CustomizedContent, msg: ReliableMessage,
                                        messenger: Messenger) -> List[Content]:
        info = content.copy_dict()
        info['type'] = ContentType.COMMAND
        info['command'] = QueryCommand.QUERY
        query = Content.parse(content=info)
        if isinstance(query, QueryCommand):
            return await messenger.process_content(content=query, r_msg=msg)
        # else:
        #     assert False, 'query command error: %s, %s, sender: %s' % (query, content, sender)
        text = 'Query command error.'
        return self._respond_receipt(text=text, envelope=msg.envelope, content=content)


# def register_customized_handlers():
#     app_filter = get_app_filter()
#     # 'chat.dim.group:history'
#     app_filter.set_content_handler(app=GroupHistory.APP,
#                                    mod=GroupHistory.MOD,
#                                    handler=GroupHistoryHandler()
#                                    )
# 
# 
# def get_app_filter() -> AppCustomizedFilter:
#     ext = message_extensions()
#     app_filter = ext.customized_filter
#     if not isinstance(app_filter, AppCustomizedFilter):
#         app_filter = AppCustomizedFilter()
#         ext.customized_filter = app_filter
#     return app_filter

ContentProcessorCreator

from typing import Optional

from dimsdk import *

from .handshake import *
from .customized import *


class ClientContentProcessorCreator(BaseContentProcessorCreator):

    # Override
    def create_content_processor(self, msg_type: str) -> Optional[ContentProcessor]:
        # application customized
        if msg_type == ContentType.APPLICATION:
            return CustomizedContentProcessor(facebook=self.facebook, messenger=self.messenger)
        elif msg_type == ContentType.CUSTOMIZED:
            return CustomizedContentProcessor(facebook=self.facebook, messenger=self.messenger)
        
        # ...
        
        # others
        return super().create_content_processor(msg_type=msg_type)

    # Override
    def create_command_processor(self, msg_type: str, cmd: str) -> Optional[ContentProcessor]:
        # handshake
        if cmd == HandshakeCommand.HANDSHAKE:
            return HandshakeCommandProcessor(facebook=self.facebook, messenger=self.messenger)
        
        # ...
        
        # others
        return super().create_command_processor(msg_type=msg_type, cmd=cmd)

Usage

To let your CustomizedContentProcessor start to work, you must override BaseContentProcessorCreator for message types:

1. ContentType.APPLICATION
2. ContentType.CUSTOMIZED

and then set your creator for GeneralContentProcessorFactory in the MessageProcessor.

---

Copyright © 2018-2026 Albert Moky