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
customized_filter = get_app_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)
- 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 NotImplementedError(
f'Not implemented: {type(self).__module__}.{type(self).__name__}.handle_action()'
)
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[StrMap] = None) -> List[Command]:
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 NotImplementedError(
f'Not implemented: {type(self).__module__}.{type(self).__name__}.filter_content()'
)
class AppCustomizedFilter(CustomizedContentFilter):
def __init__(self):
super().__init__()
self.__default_handler = BaseCustomizedContentHandler()
self.__handlers: MutableMapping[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 CustomizedFilterExtension:
@property
def customized_filter(self) -> CustomizedContentFilter:
raise NotImplementedError(
f'Not implemented: {type(self).__module__}.{type(self).__name__}.customized_filter getter'
)
@customized_filter.setter
def customized_filter(self, delegate: CustomizedContentFilter):
raise NotImplementedError(
f'Not implemented: {type(self).__module__}.{type(self).__name__}.customized_filter setter'
)
shared_message_extensions.customized_filter = AppCustomizedFilter()
def customized_extensions() -> CustomizedFilterExtension:
return shared_message_extensions
def get_app_filter() -> AppCustomizedFilter:
ext = customized_extensions()
customized_filter = ext.customized_filter
if not isinstance(customized_filter, AppCustomizedFilter):
customized_filter = AppCustomizedFilter()
ext.customized_filter = customized_filter
return customized_filter
- 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_map()
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='chat.dim.group',
# mod='history',
# handler=GroupHistoryHandler()
# )
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 or msg_type == 'application':
return CustomizedContentProcessor(facebook=self.facebook, messenger=self.messenger)
elif msg_type == ContentType.CUSTOMIZED or msg_type == '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:
- ContentType.APPLICATION
- ContentType.CUSTOMIZED
and then set your creator for GeneralContentProcessorFactory in the MessageProcessor.
Copyright © 2018-2026 Albert Moky
1 maintainer
Metadata
1 maintainer
Release files for dimsdk 2.5.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| dimsdk-2.5.0.tar.gz | 40.7 kB | Without provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| dimsdk-2.5.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 120.6 kB
Release files / dimsdk-2.5.0.tar.gz
| Download URL | dimsdk-2.5.0.tar.gz |
|---|---|
| Size | 40.7 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
9652bb017961886510dffeb2379906a8e117ec80b931802d984ed635781e7168
|
|
BLAKE2b-256 checksum How to use checksums |
e77a4abf4a3cb4ea80a402640cb5f2572f8b920c3cd4925f52ea5d68f5c71a3c
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/1.13.0 pkginfo/1.5.0.1 requests/2.21.0 setuptools/68.0.0 requests-toolbelt/0.9.1 tqdm/4.32.2 CPython/3.7.0b3
|
Release files / dimsdk-2.5.0-py3-none-any.whl
| Download URL | dimsdk-2.5.0-py3-none-any.whl |
|---|---|
| Size | 79.9 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
0ae10381b85be78c0c9c2acd1a691ac6a12e8a52b79049decc2017cf56a026e8
|
|
BLAKE2b-256 checksum How to use checksums |
e3c258118f272053cefe32f49311551f68304660281e8332091c077a2b8d3b2c
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/1.13.0 pkginfo/1.5.0.1 requests/2.21.0 setuptools/68.0.0 requests-toolbelt/0.9.1 tqdm/4.32.2 CPython/3.7.0b3
|
Security for dimsdk 2.5.0
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.
