示例代码:
import argparse
import logging
from dingtalk_stream import AckMessage
import dingtalk_stream
def setup_logger():
logger = logging.getLogger()
handler = logging.StreamHandler()
handler.setFormatter(
logging.Formatter(’%(asctime)s %(name)-8s %(levelname)-8s %(message)s [%(filename)s:%(lineno)d]’))
logger.addHandler(handler)
logger.setLevel(logging.INFO)
return logger
def define_options():
parser = argparse.ArgumentParser()
parser.add_argument(
’–client_id’, dest=‘client_id’, required=True,
help=‘app_key or suite_key from https://open-dev.digntalk.com’
)
parser.add_argument(
’–client_secret’, dest=‘client_secret’, required=True,
help=‘app_secret or suite_secret from https://open-dev.digntalk.com’
)
options = parser.parse_args()
return options
class EchoTextHandler(dingtalk_stream.ChatbotHandler):
def init(self, logger: logging.Logger = None):
super(dingtalk_stream.ChatbotHandler, self).init()
if logger:
self.logger = logger
async def process(self, callback: dingtalk_stream.CallbackMessage):
incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
print(incoming_message)
text = incoming_message.text.content.strip()
self.reply_text(text, incoming_message)
return AckMessage.STATUS_OK, 'OK'
def main():
logger = setup_logger()
options = define_options()
credential = dingtalk_stream.Credential(options.client_id, options.client_secret)
client = dingtalk_stream.DingTalkStreamClient(credential)
client.register_callback_handler(dingtalk_stream.chatbot.ChatbotMessage.TOPIC, EchoTextHandler(logger))
client.start_forever()
if name == ‘main’:
main()
出错信息:
python ./src/echo_text.py --client_id=“ding”Xxxxxxxxx --client_secret="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
2025-05-17 07:57:27,030 dingtalk_stream.client INFO open connection, url=https://api.dingtalk.com/v1.0/gateway/connections/open [stream.py:151]
2025-05-17 07:57:27,161 dingtalk_stream.client INFO endpoint is {‘endpoint’: ‘wss://wss-open-connection.dingtalk.com:443/connect’, ‘ticket’: ‘852473fc-32b1-11f0-8bff-5215a8ef9827’} [stream.py:73]
— Logging error —
Traceback (most recent call last):
File “/Users/imooc/Library/Caches/pypoetry/virtualenvs/dingtalk-langchain-G04N6Vrq-py3.12/lib/python3.12/site-packages/dingtalk_stream/stream.py”, line 76, in start
async with websockets.connect(uri) as websocket:
File “/Users/imooc/Library/Caches/pypoetry/virtualenvs/dingtalk-langchain-G04N6Vrq-py3.12/lib/python3.12/site-packages/websockets/asyncio/client.py”, line 587, in aenter
return await self
^^^^^^^^^^
File “/Users/imooc/Library/Caches/pypoetry/virtualenvs/dingtalk-langchain-G04N6Vrq-py3.12/lib/python3.12/site-packages/websockets/asyncio/client.py”, line 541, in await_impl
self.connection = await self.create_connection()
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File “/Users/imooc/Library/Caches/pypoetry/virtualenvs/dingtalk-langchain-G04N6Vrq-py3.12/lib/python3.12/site-packages/websockets/asyncio/client.py”, line 467, in create_connection
_, connection = await loop.create_connection(factory, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/base_events.py”, line 1146, in create_connection
transport, protocol = await self._create_connection_transport(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/base_events.py”, line 1179, in _create_connection_transport
await waiter
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/sslproto.py”, line 575, in _on_handshake_complete
raise handshake_exc
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/sslproto.py”, line 557, in _do_handshake
self._sslobj.do_handshake()
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/ssl.py”, line 917, in do_handshake
self._sslobj.do_handshake()
ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1000)
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/logging/init.py”, line 1160, in emit
msg = self.format(record)
^^^^^^^^^^^^^^^^^^^
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/logging/init.py”, line 999, in format
return fmt.format(record)
^^^^^^^^^^^^^^^^^^
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/logging/init.py”, line 703, in format
record.message = record.getMessage()
^^^^^^^^^^^^^^^^^^^
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/logging/init.py”, line 392, in getMessage
msg = msg % self.args~~~~~~^
TypeError: not all arguments converted during string formatting
Call stack:
File “/Users/imooc/ai_learn/dingtalk_langchain/./src/echo_text.py”, line 52, in
main()
File “/Users/imooc/ai_learn/dingtalk_langchain/./src/echo_text.py”, line 49, in main
client.start_forever()
File “/Users/imooc/Library/Caches/pypoetry/virtualenvs/dingtalk-langchain-G04N6Vrq-py3.12/lib/python3.12/site-packages/dingtalk_stream/stream.py”, line 144, in start_forever
asyncio.run(self.start())
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/runners.py”, line 194, in run
return runner.run(main)
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/runners.py”, line 118, in run
return self._loop.run_until_complete(task)
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/base_events.py”, line 671, in run_until_complete
self.run_forever()
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/base_events.py”, line 638, in run_forever
self._run_once()
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/base_events.py”, line 1971, in _run_once
handle._run()
File “/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/events.py”, line 84, in _run
self._context.run(self._callback, *self._args)
File “/Users/imooc/Library/Caches/pypoetry/virtualenvs/dingtalk-langchain-G04N6Vrq-py3.12/lib/python3.12/site-packages/dingtalk_stream/stream.py”, line 91, in start
self.logger.exception(‘unknown exception’, e)
Message: 'unknown exception’
Arguments: (SSLCertVerificationError(1, ‘[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1000)’),)
2025-05-17 07:57:30,280 dingtalk_stream.client INFO open connection, url=https://api.dingtalk.com/v1.0/gateway/connections/open [stream.py:151]
2025-05-17 07:57:30,379 dingtalk_stream.client INFO endpoint is {‘endpoint’: ‘wss://wss-open-connection.dingtalk.com:443/connect’, ‘ticket’: ‘870e6f3f-32b1-11f0-8bff-5215a8ef9827’} [stream.py:73]
登录后可查看更多问答,登录/注册