You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
преди 1 година
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129
  1. # -*- coding: utf-8 -*-
  2. import web
  3. import time
  4. import json
  5. import requests
  6. import threading
  7. from common.singleton import singleton
  8. from common.log import logger
  9. from common.expired_dict import ExpiredDict
  10. from config import conf
  11. from bridge.reply import *
  12. from bridge.context import *
  13. from channel.chat_channel import ChatChannel
  14. from channel.wechatmp.common import *
  15. # If using SSL, uncomment the following lines, and modify the certificate path.
  16. # from cheroot.server import HTTPServer
  17. # from cheroot.ssl.builtin import BuiltinSSLAdapter
  18. # HTTPServer.ssl_adapter = BuiltinSSLAdapter(
  19. # certificate='/ssl/cert.pem',
  20. # private_key='/ssl/cert.key')
  21. @singleton
  22. class WechatMPChannel(ChatChannel):
  23. def __init__(self, passive_reply = True):
  24. super().__init__()
  25. self.passive_reply = passive_reply
  26. self.running = set()
  27. self.received_msgs = ExpiredDict(60*60*24)
  28. if self.passive_reply:
  29. self.NOT_SUPPORT_REPLYTYPE = [ReplyType.IMAGE, ReplyType.VOICE]
  30. self.cache_dict = dict()
  31. self.query1 = dict()
  32. self.query2 = dict()
  33. self.query3 = dict()
  34. else:
  35. # TODO support image
  36. self.NOT_SUPPORT_REPLYTYPE = [ReplyType.IMAGE, ReplyType.VOICE]
  37. self.app_id = conf().get('wechatmp_app_id')
  38. self.app_secret = conf().get('wechatmp_app_secret')
  39. self.access_token = None
  40. self.access_token_expires_time = 0
  41. self.access_token_lock = threading.Lock()
  42. self.get_access_token()
  43. def startup(self):
  44. if self.passive_reply:
  45. urls = ('/wx', 'channel.wechatmp.SubscribeAccount.Query')
  46. else:
  47. urls = ('/wx', 'channel.wechatmp.ServiceAccount.Query')
  48. app = web.application(urls, globals(), autoreload=False)
  49. port = conf().get('wechatmp_port', 8080)
  50. web.httpserver.runsimple(app.wsgifunc(), ('0.0.0.0', port))
  51. def wechatmp_request(self, method, url, **kwargs):
  52. r = requests.request(method=method, url=url, **kwargs)
  53. r.raise_for_status()
  54. r.encoding = "utf-8"
  55. ret = r.json()
  56. if "errcode" in ret and ret["errcode"] != 0:
  57. raise WeChatAPIException("{}".format(ret))
  58. return ret
  59. def get_access_token(self):
  60. # return the access_token
  61. if self.access_token:
  62. if self.access_token_expires_time - time.time() > 60:
  63. return self.access_token
  64. # Get new access_token
  65. # Do not request access_token in parallel! Only the last obtained is valid.
  66. if self.access_token_lock.acquire(blocking=False):
  67. # Wait for other threads that have previously obtained access_token to complete the request
  68. # This happens every 2 hours, so it doesn't affect the experience very much
  69. time.sleep(1)
  70. self.access_token = None
  71. url="https://api.weixin.qq.com/cgi-bin/token"
  72. params={
  73. "grant_type": "client_credential",
  74. "appid": self.app_id,
  75. "secret": self.app_secret
  76. }
  77. data = self.wechatmp_request(method='get', url=url, params=params)
  78. self.access_token = data['access_token']
  79. self.access_token_expires_time = int(time.time()) + data['expires_in']
  80. logger.info("[wechatmp] access_token: {}".format(self.access_token))
  81. self.access_token_lock.release()
  82. else:
  83. # Wait for token update
  84. while self.access_token_lock.locked():
  85. time.sleep(0.1)
  86. return self.access_token
  87. def send(self, reply: Reply, context: Context):
  88. if self.passive_reply:
  89. receiver = context["receiver"]
  90. self.cache_dict[receiver] = reply.content
  91. logger.info("[send] reply to {} saved to cache: {}".format(receiver, reply))
  92. else:
  93. receiver = context["receiver"]
  94. reply_text = reply.content
  95. url="https://api.weixin.qq.com/cgi-bin/message/custom/send"
  96. params = {
  97. "access_token": self.get_access_token()
  98. }
  99. json_data = {
  100. "touser": receiver,
  101. "msgtype": "text",
  102. "text": {"content": reply_text}
  103. }
  104. self.wechatmp_request(method='post', url=url, params=params, data=json.dumps(json_data, ensure_ascii=False).encode('utf8'))
  105. logger.info("[send] Do send to {}: {}".format(receiver, reply_text))
  106. return
  107. def _success_callback(self, session_id, context, **kwargs): # 线程异常结束时的回调函数
  108. logger.debug("[wechatmp] Success to generate reply, msgId={}".format(context['msg'].msg_id))
  109. if self.passive_reply:
  110. self.running.remove(session_id)
  111. def _fail_callback(self, session_id, exception, context, **kwargs): # 线程异常结束时的回调函数
  112. logger.exception("[wechatmp] Fail to generate reply to user, msgId={}, exception={}".format(context['msg'].msg_id, exception))
  113. if self.passive_reply:
  114. assert session_id not in self.cache_dict
  115. self.running.remove(session_id)