|
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236 |
-
-
- """
- wechat channel
- """
-
- import io
- import json
- import os
- import threading
- import time
-
- import requests
-
- from bridge.context import *
- from bridge.reply import *
- from channel.chat_channel import ChatChannel
- from channel.wechat.wechat_message import *
- from common.expired_dict import ExpiredDict
- from common.log import logger
- from common.singleton import singleton
- from common.time_check import time_checker
- from config import conf, get_appdata_dir
- from lib import itchat
- from lib.itchat.content import *
- from plugins import *
-
-
- @itchat.msg_register([TEXT, VOICE, PICTURE, NOTE])
- def handler_single_msg(msg):
- try:
- cmsg = WeChatMessage(msg, False)
- except NotImplementedError as e:
- logger.debug("[WX]single message {} skipped: {}".format(msg["MsgId"], e))
- return None
- WechatChannel().handle_single(cmsg)
- return None
-
-
- @itchat.msg_register([TEXT, VOICE, PICTURE, NOTE], isGroupChat=True)
- def handler_group_msg(msg):
- try:
- cmsg = WeChatMessage(msg, True)
- except NotImplementedError as e:
- logger.debug("[WX]group message {} skipped: {}".format(msg["MsgId"], e))
- return None
- WechatChannel().handle_group(cmsg)
- return None
-
-
- def _check(func):
- def wrapper(self, cmsg: ChatMessage):
- msgId = cmsg.msg_id
- if msgId in self.receivedMsgs:
- logger.info("Wechat message {} already received, ignore".format(msgId))
- return
- self.receivedMsgs[msgId] = cmsg
- create_time = cmsg.create_time
- if (
- conf().get("hot_reload") == True
- and int(create_time) < int(time.time()) - 60
- ):
- logger.debug("[WX]history message {} skipped".format(msgId))
- return
- return func(self, cmsg)
-
- return wrapper
-
-
-
-
-
- def qrCallback(uuid, status, qrcode):
-
- if status == "0":
- try:
- from PIL import Image
-
- img = Image.open(io.BytesIO(qrcode))
- _thread = threading.Thread(target=img.show, args=("QRCode",))
- _thread.setDaemon(True)
- _thread.start()
- except Exception as e:
- pass
-
- import qrcode
-
- url = f"https://login.weixin.qq.com/l/{uuid}"
-
- qr_api1 = "https://api.isoyu.com/qr/?m=1&e=L&p=20&url={}".format(url)
- qr_api2 = (
- "https://api.qrserver.com/v1/create-qr-code/?size=400×400&data={}".format(
- url
- )
- )
- qr_api3 = "https://api.pwmqr.com/qrcode/create/?url={}".format(url)
- qr_api4 = "https://my.tv.sohu.com/user/a/wvideo/getQRCode.do?text={}".format(
- url
- )
- print("You can also scan QRCode in any website below:")
- print(qr_api3)
- print(qr_api4)
- print(qr_api2)
- print(qr_api1)
-
- qr = qrcode.QRCode(border=1)
- qr.add_data(url)
- qr.make(fit=True)
- qr.print_ascii(invert=True)
-
-
- @singleton
- class WechatChannel(ChatChannel):
- NOT_SUPPORT_REPLYTYPE = []
-
- def __init__(self):
- super().__init__()
- self.receivedMsgs = ExpiredDict(60 * 60 * 24)
-
- def startup(self):
- itchat.instance.receivingRetryCount = 600
-
- hotReload = conf().get("hot_reload", False)
- status_path = os.path.join(get_appdata_dir(), "itchat.pkl")
- try:
- itchat.auto_login(
- enableCmdQR=2,
- hotReload=hotReload,
- statusStorageDir=status_path,
- qrCallback=qrCallback,
- )
- except Exception as e:
- if hotReload:
- logger.error("Hot reload failed, try to login without hot reload")
- itchat.logout()
- os.remove(status_path)
- itchat.auto_login(
- enableCmdQR=2, hotReload=hotReload, qrCallback=qrCallback
- )
- else:
- raise e
- self.user_id = itchat.instance.storageClass.userName
- self.name = itchat.instance.storageClass.nickName
- logger.info(
- "Wechat login success, user_id: {}, nickname: {}".format(
- self.user_id, self.name
- )
- )
-
- itchat.run()
-
-
-
-
-
-
-
-
-
-
-
-
-
- @time_checker
- @_check
- def handle_single(self, cmsg: ChatMessage):
- if cmsg.ctype == ContextType.VOICE:
- if conf().get("speech_recognition") != True:
- return
- logger.debug("[WX]receive voice msg: {}".format(cmsg.content))
- elif cmsg.ctype == ContextType.IMAGE:
- logger.debug("[WX]receive image msg: {}".format(cmsg.content))
- elif cmsg.ctype == ContextType.PATPAT:
- logger.debug("[WX]receive patpat msg: {}".format(cmsg.content))
- elif cmsg.ctype == ContextType.TEXT:
- logger.debug(
- "[WX]receive text msg: {}, cmsg={}".format(
- json.dumps(cmsg._rawmsg, ensure_ascii=False), cmsg
- )
- )
- else:
- logger.debug("[WX]receive msg: {}, cmsg={}".format(cmsg.content, cmsg))
- context = self._compose_context(
- cmsg.ctype, cmsg.content, isgroup=False, msg=cmsg
- )
- if context:
- self.produce(context)
-
- @time_checker
- @_check
- def handle_group(self, cmsg: ChatMessage):
- if cmsg.ctype == ContextType.VOICE:
- if conf().get("speech_recognition") != True:
- return
- logger.debug("[WX]receive voice for group msg: {}".format(cmsg.content))
- elif cmsg.ctype == ContextType.IMAGE:
- logger.debug("[WX]receive image for group msg: {}".format(cmsg.content))
- elif cmsg.ctype in [ContextType.JOIN_GROUP, ContextType.PATPAT]:
- logger.debug("[WX]receive note msg: {}".format(cmsg.content))
- elif cmsg.ctype == ContextType.TEXT:
-
- pass
- else:
- logger.debug("[WX]receive group msg: {}".format(cmsg.content))
- context = self._compose_context(
- cmsg.ctype, cmsg.content, isgroup=True, msg=cmsg
- )
- if context:
- self.produce(context)
-
-
- def send(self, reply: Reply, context: Context):
- receiver = context["receiver"]
- if reply.type == ReplyType.TEXT:
- itchat.send(reply.content, toUserName=receiver)
- logger.info("[WX] sendMsg={}, receiver={}".format(reply, receiver))
- elif reply.type == ReplyType.ERROR or reply.type == ReplyType.INFO:
- itchat.send(reply.content, toUserName=receiver)
- logger.info("[WX] sendMsg={}, receiver={}".format(reply, receiver))
- elif reply.type == ReplyType.VOICE:
- itchat.send_file(reply.content, toUserName=receiver)
- logger.info("[WX] sendFile={}, receiver={}".format(reply.content, receiver))
- elif reply.type == ReplyType.IMAGE_URL:
- img_url = reply.content
- pic_res = requests.get(img_url, stream=True)
- image_storage = io.BytesIO()
- for block in pic_res.iter_content(1024):
- image_storage.write(block)
- image_storage.seek(0)
- itchat.send_image(image_storage, toUserName=receiver)
- logger.info("[WX] sendImage url={}, receiver={}".format(img_url, receiver))
- elif reply.type == ReplyType.IMAGE:
- image_storage = reply.content
- image_storage.seek(0)
- itchat.send_image(image_storage, toUserName=receiver)
- logger.info("[WX] sendImage, receiver={}".format(receiver))
|