wechatmp_channel.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236
  1. # -*- coding: utf-8 -*-
  2. import asyncio
  3. import imghdr
  4. import io
  5. import os
  6. import threading
  7. import time
  8. import requests
  9. import web
  10. from wechatpy.crypto import WeChatCrypto
  11. from wechatpy.exceptions import WeChatClientException
  12. from collections import defaultdict
  13. from bridge.context import *
  14. from bridge.reply import *
  15. from channel.chat_channel import ChatChannel
  16. from channel.wechatmp.common import *
  17. from channel.wechatmp.wechatmp_client import WechatMPClient
  18. from common.log import logger
  19. from common.singleton import singleton
  20. from common.utils import split_string_by_utf8_length
  21. from config import conf
  22. from voice.audio_convert import any_to_mp3, split_audio
  23. # If using SSL, uncomment the following lines, and modify the certificate path.
  24. # from cheroot.server import HTTPServer
  25. # from cheroot.ssl.builtin import BuiltinSSLAdapter
  26. # HTTPServer.ssl_adapter = BuiltinSSLAdapter(
  27. # certificate='/ssl/cert.pem',
  28. # private_key='/ssl/cert.key')
  29. @singleton
  30. class WechatMPChannel(ChatChannel):
  31. def __init__(self, passive_reply=True):
  32. super().__init__()
  33. self.passive_reply = passive_reply
  34. self.NOT_SUPPORT_REPLYTYPE = []
  35. appid = conf().get("wechatmp_app_id")
  36. secret = conf().get("wechatmp_app_secret")
  37. token = conf().get("wechatmp_token")
  38. aes_key = conf().get("wechatmp_aes_key")
  39. self.client = WechatMPClient(appid, secret)
  40. self.crypto = None
  41. if aes_key:
  42. self.crypto = WeChatCrypto(token, aes_key, appid)
  43. if self.passive_reply:
  44. # Cache the reply to the user's first message
  45. self.cache_dict = defaultdict(list)
  46. # Record whether the current message is being processed
  47. self.running = set()
  48. # Count the request from wechat official server by message_id
  49. self.request_cnt = dict()
  50. # The permanent media need to be deleted to avoid media number limit
  51. self.delete_media_loop = asyncio.new_event_loop()
  52. t = threading.Thread(target=self.start_loop, args=(self.delete_media_loop,))
  53. t.setDaemon(True)
  54. t.start()
  55. def startup(self):
  56. if self.passive_reply:
  57. urls = ("/wx", "channel.wechatmp.passive_reply.Query")
  58. else:
  59. urls = ("/wx", "channel.wechatmp.active_reply.Query")
  60. app = web.application(urls, globals(), autoreload=False)
  61. port = conf().get("wechatmp_port", 8080)
  62. web.httpserver.runsimple(app.wsgifunc(), ("0.0.0.0", port))
  63. def start_loop(self, loop):
  64. asyncio.set_event_loop(loop)
  65. loop.run_forever()
  66. async def delete_media(self, media_id):
  67. logger.debug("[wechatmp] permanent media {} will be deleted in 10s".format(media_id))
  68. await asyncio.sleep(10)
  69. self.client.material.delete(media_id)
  70. logger.info("[wechatmp] permanent media {} has been deleted".format(media_id))
  71. def send(self, reply: Reply, context: Context):
  72. receiver = context["receiver"]
  73. if self.passive_reply:
  74. if reply.type == ReplyType.TEXT or reply.type == ReplyType.INFO or reply.type == ReplyType.ERROR:
  75. reply_text = reply.content
  76. logger.info("[wechatmp] text cached, receiver {}\n{}".format(receiver, reply_text))
  77. self.cache_dict[receiver].append(("text", reply_text))
  78. elif reply.type == ReplyType.VOICE:
  79. voice_file_path = reply.content
  80. duration, files = split_audio(voice_file_path, 60 * 1000)
  81. if len(files) > 1:
  82. logger.info("[wechatmp] voice too long {}s > 60s , split into {} parts".format(duration / 1000.0, len(files)))
  83. for path in files:
  84. # support: <2M, <60s, mp3/wma/wav/amr
  85. try:
  86. with open(path, "rb") as f:
  87. response = self.client.material.add("voice", f)
  88. logger.debug("[wechatmp] upload voice response: {}".format(response))
  89. f_size = os.fstat(f.fileno()).st_size
  90. time.sleep(1.0 + 2 * f_size / 1024 / 1024)
  91. # todo check media_id
  92. except WeChatClientException as e:
  93. logger.error("[wechatmp] upload voice failed: {}".format(e))
  94. return
  95. media_id = response["media_id"]
  96. logger.info("[wechatmp] voice uploaded, receiver {}, media_id {}".format(receiver, media_id))
  97. self.cache_dict[receiver].append(("voice", media_id))
  98. elif reply.type == ReplyType.IMAGE_URL: # 从网络下载图片
  99. img_url = reply.content
  100. pic_res = requests.get(img_url, stream=True)
  101. image_storage = io.BytesIO()
  102. for block in pic_res.iter_content(1024):
  103. image_storage.write(block)
  104. image_storage.seek(0)
  105. image_type = imghdr.what(image_storage)
  106. filename = receiver + "-" + str(context["msg"].msg_id) + "." + image_type
  107. content_type = "image/" + image_type
  108. try:
  109. response = self.client.material.add("image", (filename, image_storage, content_type))
  110. logger.debug("[wechatmp] upload image response: {}".format(response))
  111. except WeChatClientException as e:
  112. logger.error("[wechatmp] upload image failed: {}".format(e))
  113. return
  114. media_id = response["media_id"]
  115. logger.info("[wechatmp] image uploaded, receiver {}, media_id {}".format(receiver, media_id))
  116. self.cache_dict[receiver].append(("image", media_id))
  117. elif reply.type == ReplyType.IMAGE: # 从文件读取图片
  118. image_storage = reply.content
  119. image_storage.seek(0)
  120. image_type = imghdr.what(image_storage)
  121. filename = receiver + "-" + str(context["msg"].msg_id) + "." + image_type
  122. content_type = "image/" + image_type
  123. try:
  124. response = self.client.material.add("image", (filename, image_storage, content_type))
  125. logger.debug("[wechatmp] upload image response: {}".format(response))
  126. except WeChatClientException as e:
  127. logger.error("[wechatmp] upload image failed: {}".format(e))
  128. return
  129. media_id = response["media_id"]
  130. logger.info("[wechatmp] image uploaded, receiver {}, media_id {}".format(receiver, media_id))
  131. self.cache_dict[receiver].append(("image", media_id))
  132. else:
  133. if reply.type == ReplyType.TEXT or reply.type == ReplyType.INFO or reply.type == ReplyType.ERROR:
  134. reply_text = reply.content
  135. texts = split_string_by_utf8_length(reply_text, MAX_UTF8_LEN)
  136. if len(texts) > 1:
  137. logger.info("[wechatmp] text too long, split into {} parts".format(len(texts)))
  138. for i, text in enumerate(texts):
  139. self.client.message.send_text(receiver, text)
  140. if i != len(texts) - 1:
  141. time.sleep(0.5) # 休眠0.5秒,防止发送过快乱序
  142. logger.info("[wechatmp] Do send text to {}: {}".format(receiver, reply_text))
  143. elif reply.type == ReplyType.VOICE:
  144. try:
  145. file_path = reply.content
  146. file_name = os.path.basename(file_path)
  147. file_type = os.path.splitext(file_name)[1]
  148. if file_type == ".mp3":
  149. file_type = "audio/mpeg"
  150. elif file_type == ".amr":
  151. file_type = "audio/amr"
  152. else:
  153. mp3_file = os.path.splitext(file_path)[0] + ".mp3"
  154. any_to_mp3(file_path, mp3_file)
  155. file_path = mp3_file
  156. file_name = os.path.basename(file_path)
  157. file_type = "audio/mpeg"
  158. logger.info("[wechatmp] file_name: {}, file_type: {} ".format(file_name, file_type))
  159. media_ids = []
  160. duration, files = split_audio(file_path, 60 * 1000)
  161. if len(files) > 1:
  162. logger.info("[wechatmp] voice too long {}s > 60s , split into {} parts".format(duration / 1000.0, len(files)))
  163. for path in files:
  164. # support: <2M, <60s, AMR\MP3
  165. response = self.client.media.upload("voice", (os.path.basename(path), open(path, "rb"), file_type))
  166. logger.debug("[wechatcom] upload voice response: {}".format(response))
  167. media_ids.append(response["media_id"])
  168. os.remove(path)
  169. except WeChatClientException as e:
  170. logger.error("[wechatmp] upload voice failed: {}".format(e))
  171. return
  172. try:
  173. os.remove(file_path)
  174. except Exception:
  175. pass
  176. for media_id in media_ids:
  177. self.client.message.send_voice(receiver, media_id)
  178. time.sleep(1)
  179. logger.info("[wechatmp] Do send voice to {}".format(receiver))
  180. elif reply.type == ReplyType.IMAGE_URL: # 从网络下载图片
  181. img_url = reply.content
  182. pic_res = requests.get(img_url, stream=True)
  183. image_storage = io.BytesIO()
  184. for block in pic_res.iter_content(1024):
  185. image_storage.write(block)
  186. image_storage.seek(0)
  187. image_type = imghdr.what(image_storage)
  188. filename = receiver + "-" + str(context["msg"].msg_id) + "." + image_type
  189. content_type = "image/" + image_type
  190. try:
  191. response = self.client.media.upload("image", (filename, image_storage, content_type))
  192. logger.debug("[wechatmp] upload image response: {}".format(response))
  193. except WeChatClientException as e:
  194. logger.error("[wechatmp] upload image failed: {}".format(e))
  195. return
  196. self.client.message.send_image(receiver, response["media_id"])
  197. logger.info("[wechatmp] Do send image to {}".format(receiver))
  198. elif reply.type == ReplyType.IMAGE: # 从文件读取图片
  199. image_storage = reply.content
  200. image_storage.seek(0)
  201. image_type = imghdr.what(image_storage)
  202. filename = receiver + "-" + str(context["msg"].msg_id) + "." + image_type
  203. content_type = "image/" + image_type
  204. try:
  205. response = self.client.media.upload("image", (filename, image_storage, content_type))
  206. logger.debug("[wechatmp] upload image response: {}".format(response))
  207. except WeChatClientException as e:
  208. logger.error("[wechatmp] upload image failed: {}".format(e))
  209. return
  210. self.client.message.send_image(receiver, response["media_id"])
  211. logger.info("[wechatmp] Do send image to {}".format(receiver))
  212. return
  213. def _success_callback(self, session_id, context, **kwargs): # 线程异常结束时的回调函数
  214. logger.debug("[wechatmp] Success to generate reply, msgId={}".format(context["msg"].msg_id))
  215. if self.passive_reply:
  216. self.running.remove(session_id)
  217. def _fail_callback(self, session_id, exception, context, **kwargs): # 线程异常结束时的回调函数
  218. logger.exception("[wechatmp] Fail to generate reply to user, msgId={}, exception={}".format(context["msg"].msg_id, exception))
  219. if self.passive_reply:
  220. assert session_id not in self.cache_dict
  221. self.running.remove(session_id)