<commit_before>import io
import logging
import mimetypes
import os
import re
import threading
import time
import html
from binascii import crc32

import itchat
import magic
import xmltodict
from PIL import Image

import config
from channel import EFBChannel, EFBMsg, MsgType, MsgSource, TargetType, ChannelType
from channelExceptions import EFBMessageTypeNotSupported, EFBMessageError, EFBChatNotFound
from utils import extra


def incomeMsgMeta(func):
    def wcFunc(self, msg, isGroupChat=False):
        logger = logging.getLogger("plugins.%s.incomeMsgMeta" % self.channel_id)
        mobj = func(self, msg, isGroupChat)
        mobj.uid = msg.get("MsgId", time.time())
        me = msg['FromUserName'] == self.itchat.loginInfo['User']['UserName']
        logger.debug("me, %s", me)
        if me:
            msg['FromUserName'], msg['ToUserName'] = msg['ToUserName'], msg['FromUserName']
        FromUser = self.search_user(UserName=msg['FromUserName']) or \
                       self.search_user(UserName=msg['FromUserName'], refresh=True) or \
                       [{"NickName": "Chat not found. (UE01)", "RemarkName": "Chat not found. (UE01)", "Uin": 0}]
        FromUser = FromUser[0]
        logger.debug("From user, %s", FromUser)
        if isGroupChat:
            logger.debug("groupchat")
            if me:
                msg['ActualUserName'] = msg['ToUserName']
                member = {"NickName": self.itchat.loginInfo['User']['NickName'], "DisplayName": "You", "Uin": self.itchat.loginInfo['User']['Uin']}
            else:
                logger.debug("search_user")
                member = self.search_user(UserName=msg['FromUserName'], ActualUserName=msg['ActualUserName'])[0]['MemberList'][0]
                logger.debug("search_user.done")
            mobj.source = MsgSource.Group
            logger.debug("Group. member: %s", member)
            mobj.origin = {
                'name': html.unescape(FromUser['NickName']),
                'alias': html.unescape(FromUser['RemarkName'] or FromUser['NickName']),
                'uid': self.get_uid(UserName=msg.get('FromUserName', None),
                                    NickName=html.unescape(FromUser.get('NickName', "s")),
                                    alias=html.unescape(FromUser.get('RemarkName', "s")),
                                    Uin=FromUser.get('Uin', None))
            }
            mobj.member = {
                'name': html.unescape(member['NickName']),
                'alias': html.unescape(member['DisplayName']),
                'uid': self.get_uid(UserName=msg.get('ActualUserName', None),
                                    NickName=html.unescape(member.get('NickName', "")),
                                    alias=html.unescape(member.get('DisplayName', "")),
                                    Uin=member.get('Uin', None))
            }
            logger.debug("origin: %s\nmember: %s\n", mobj.origin, mobj.member)
        else:
            if me:
                mobj.text = mobj.text or ""
                mobj.text = "You: " + mobj.text
            mobj.source = MsgSource.User
            mobj.origin = {
                'name': html.unescape(FromUser['NickName']),
                'alias': html.unescape(FromUser['RemarkName'] or FromUser['NickName']),
                'uid': self.get_uid(UserName=msg.get('FromUserName', None),
                                    NickName=html.unescape(FromUser.get('NickName', "")),
                                    alias=html.unescape(FromUser.get('RemarkName', "")),
                                    Uin=FromUser.get('Uin', None))
            }
        mobj.destination = {
            'name': self.itchat.loginInfo['User']['NickName'],
            'alias': self.itchat.loginInfo['User']['NickName'],
            'uid': self.get_uid(UserName=msg['ToUserName'])
        }
        logger.debug("dest: %s", mobj.destination)
        logger.info("WeChat incoming message:\nType: %s\nText: %s\nUserName: %s\nuid: %s\nname: %s" %
                    (mobj.type, msg['Text'], msg['FromUserName'], mobj.origin['uid'], mobj.origin['name']))
        self.queue.put(mobj)

    return wcFunc


class WeChatChannel(EFBChannel):
    """
    EFB Channel - WeChat (slave)
    Based on itchat (modified by <NAME>)

    Author: <NAME> <https://github.com/blueset>
    """
    channel_name = "WeChat Slave"
    channel_emoji = "💬"
    channel_id = "eh_wechat_slave"
    channel_type = ChannelType.Slave

    supported_message_types = {MsgType.Text, MsgType.Sticker, MsgType.Image,
                               MsgType.File, MsgType.Video, MsgType.Link, MsgType.Audio}
    logger = logging.getLogger("plugins.%s.WeChatChannel" % channel_id)
    qr_uuid = ""
    done_reauth = threading.Event()
    _stop_polling = False

    SYSTEM_USERNAMES = ["filehelper", "newsapp", "fmessage", "weibo", "qqmail", "fmessage", "tmessage", "qmessage",
                         "qqsync", "floatbottle", "lbsapp", "shakeapp", "medianote", "qqfriend", "readerapp",
                         "blogapp", "facebookapp", "masssendapp", "meishiapp", "feedsapp", "voip", "blogappweixin",
                         "weixin"]

    def __init__(self, queue, mutex):
        super().__init__(queue, mutex)
        self.itchat = itchat.new_instance()
        itchat.set_logging(loggingLevel=logging.getLogger().level, showOnCmd=False)
        self.itchat_msg_register()
        with mutex:
            self.itchat.auto_login(enableCmdQR=2,
                                   hotReload=True,
                                   statusStorageDir="storage/%s.pkl" % self.channel_id,
                                   exitCallback=self.exit_callback,
                                   qrCallback=self.console_qr_code)
        mimetypes.init(files=["mimetypes"])
        self.logger.info("EWS Inited!!!\n---")

    #
    # Utilities
    #

    def console_qr_code(self, uuid, status, qrcode):
        status = int(status)
        if status == 201:
            QR = 'Tap "Confirm" to continue.'
            return self.logger.critical(QR)
        elif status == 200:
            QR = "Successfully authorized."
            return self.logger.critical(QR)
        elif uuid != self.qr_uuid:
            # 0: First QR code
            # 408: Updated QR code
            QR = "WeChat: Scan QR code with WeChat to continue. (%s, %s)\n" % (uuid, status)
            if status == 408:
                QR += "Previous code expired. Please scan the new one.\n"
            QR += "\n"
            time.sleep(0.5)
            img = Image.open(io.BytesIO(qrcode)).convert("1")
            img = img.resize((img.size[0] // 10, img.size[1] // 10))
            for i in range(img.size[0]):
                for j in range(img.size[1]):
                        if img.getpixel((i, j)) > 0:
                            QR += "\x1b[7m  \x1b[0m"
                        else:
                            QR += "\x1b[49m  \x1b[0m"
                QR += "\n"
            QR += "\nIf you cannot read the QR code above, " \
                  "Please generate a QR code with any tool utsing the following URL:\n" \
                  "https://login.weixin.qq.com/l/" + uuid
            return self.logger.critical(QR)
        self.qr_uuid = uuid

    def master_qr_code(self, uuid, status, qrcode):
        status = int(status)
        msg = EFBMsg(self)
        msg.type = MsgType.Text
        msg.source = MsgSource.System<commit_msg>EWS: Load QR Code from WeChat Web API (#22 pangxiezhou/wechat_qrcode_new_url)<commit_after>import io
import logging
import mimetypes
import os
import re
import threading
import time
import html
from binascii import crc32

import itchat
import magic
import xmltodict
from PIL import Image

import config
from channel import EFBChannel, EFBMsg, MsgType, MsgSource, TargetType, ChannelType
from channelExceptions import EFBMessageTypeNotSupported, EFBMessageError, EFBChatNotFound
from utils import extra


def incomeMsgMeta(func):
    def wcFunc(self, msg, isGroupChat=False):
        logger = logging.getLogger("plugins.%s.incomeMsgMeta" % self.channel_id)
        mobj = func(self, msg, isGroupChat)
        mobj.uid = msg.get("MsgId", time.time())
        me = msg['FromUserName'] == self.itchat.loginInfo['User']['UserName']
        logger.debug("me, %s", me)
        if me:
            msg['FromUserName'], msg['ToUserName'] = msg['ToUserName'], msg['FromUserName']
        FromUser = self.search_user(UserName=msg['FromUserName']) or \
                       self.search_user(UserName=msg['FromUserName'], refresh=True) or \
                       [{"NickName": "Chat not found. (UE01)", "RemarkName": "Chat not found. (UE01)", "Uin": 0}]
        FromUser = FromUser[0]
        logger.debug("From user, %s", FromUser)
        if isGroupChat:
            logger.debug("groupchat")
            if me:
                msg['ActualUserName'] = msg['ToUserName']
                member = {"NickName": self.itchat.loginInfo['User']['NickName'], "DisplayName": "You", "Uin": self.itchat.loginInfo['User']['Uin']}
            else:
                logger.debug("search_user")
                member = self.search_user(UserName=msg['FromUserName'], ActualUserName=msg['ActualUserName'])[0]['MemberList'][0]
                logger.debug("search_user.done")
            mobj.source = MsgSource.Group
            logger.debug("Group. member: %s", member)
            mobj.origin = {
                'name': html.unescape(FromUser['NickName']),
                'alias': html.unescape(FromUser['RemarkName'] or FromUser['NickName']),
                'uid': self.get_uid(UserName=msg.get('FromUserName', None),
                                    NickName=html.unescape(FromUser.get('NickName', "s")),
                                    alias=html.unescape(FromUser.get('RemarkName', "s")),
                                    Uin=FromUser.get('Uin', None))
            }
            mobj.member = {
                'name': html.unescape(member['NickName']),
                'alias': html.unescape(member['DisplayName']),
                'uid': self.get_uid(UserName=msg.get('ActualUserName', None),
                                    NickName=html.unescape(member.get('NickName', "")),
                                    alias=html.unescape(member.get('DisplayName', "")),
                                    Uin=member.get('Uin', None))
            }
            logger.debug("origin: %s\nmember: %s\n", mobj.origin, mobj.member)
        else:
            if me:
                mobj.text = mobj.text or ""
                mobj.text = "You: " + mobj.text
            mobj.source = MsgSource.User
            mobj.origin = {
                'name': html.unescape(FromUser['NickName']),
                'alias': html.unescape(FromUser['RemarkName'] or FromUser['NickName']),
                'uid': self.get_uid(UserName=msg.get('FromUserName', None),
                                    NickName=html.unescape(FromUser.get('NickName', "")),
                                    alias=html.unescape(FromUser.get('RemarkName', "")),
                                    Uin=FromUser.get('Uin', None))
            }
        mobj.destination = {
            'name': self.itchat.loginInfo['User']['NickName'],
            'alias': self.itchat.loginInfo['User']['NickName'],
            'uid': self.get_uid(UserName=msg['ToUserName'])
        }
        logger.debug("dest: %s", mobj.destination)
        logger.info("WeChat incoming message:\nType: %s\nText: %s\nUserName: %s\nuid: %s\nname: %s" %
                    (mobj.type, msg['Text'], msg['FromUserName'], mobj.origin['uid'], mobj.origin['name']))
        self.queue.put(mobj)

    return wcFunc


class WeChatChannel(EFBChannel):
    """
    EFB Channel - WeChat (slave)
    Based on itchat (modified by <NAME>)

    Author: <NAME> <https://github.com/blueset>
    """
    channel_name = "WeChat Slave"
    channel_emoji = "💬"
    channel_id = "eh_wechat_slave"
    channel_type = ChannelType.Slave

    supported_message_types = {MsgType.Text, MsgType.Sticker, MsgType.Image,
                               MsgType.File, MsgType.Video, MsgType.Link, MsgType.Audio}
    logger = logging.getLogger("plugins.%s.WeChatChannel" % channel_id)
    qr_uuid = ""
    done_reauth = threading.Event()
    _stop_polling = False

    SYSTEM_USERNAMES = ["filehelper", "newsapp", "fmessage", "weibo", "qqmail", "fmessage", "tmessage", "qmessage",
                         "qqsync", "floatbottle", "lbsapp", "shakeapp", "medianote", "qqfriend", "readerapp",
                         "blogapp", "facebookapp", "masssendapp", "meishiapp", "feedsapp", "voip", "blogappweixin",
                         "weixin"]

    def __init__(self, queue, mutex):
        super().__init__(queue, mutex)
        self.itchat = itchat.new_instance()
        itchat.set_logging(loggingLevel=logging.getLogger().level, showOnCmd=False)
        self.itchat_msg_register()
        with mutex:
            self.itchat.auto_login(enableCmdQR=2,
                                   hotReload=True,
                                   statusStorageDir="storage/%s.pkl" % self.channel_id,
                                   exitCallback=self.exit_callback,
                                   qrCallback=self.console_qr_code)
        mimetypes.init(files=["mimetypes"])
        self.logger.info("EWS Inited!!!\n---")

    #
    # Utilities
    #

    def console_qr_code(self, uuid, status, qrcode):
        status = int(status)
        if status == 201:
            QR = 'Tap "Confirm" to continue.'
            return self.logger.critical(QR)
        elif status == 200:
            QR = "Successfully authorized."
            return self.logger.critical(QR)
        elif uuid != self.qr_uuid:
            # 0: First QR code
            # 408: Updated QR code
            QR = "WeChat: Scan QR code with WeChat to continue. (%s, %s)\n" % (uuid, status)
            if status == 408:
                QR += "Previous code expired. Please scan the new one.\n"
            QR += "\n"
            time.sleep(0.5)
            img = Image.open(io.BytesIO(qrcode)).convert("1")
            img = img.resize((img.size[0] // 10, img.size[1] // 10))
            for i in range(img.size[0]):
                for j in range(img.size[1]):
                        if img.getpixel((i, j)) > 0:
                            QR += "\x1b[7m  \x1b[0m"
                        else:
                            QR += "\x1b[49m  \x1b[0m"
                QR += "\n"
            QR += "\nIf you cannot read the QR code above, " \
                  "Please open the following URL:\n" \
                  "https://login.weixin.qq.com/qrcode/" + uuid
            return self.logger.critical(QR)
        self.qr_uuid = uuid

    def master_qr_code(self, uuid, status, qrcode):
        status = int(status)
        msg = EFBMsg(self)
        msg.type = MsgType.Text
        msg.source = MsgSource.System
{'fraction_non_alphanumeric': 0.13729040912139504, 'fraction_numerical': 0.006170355466130114, 'mean_word_length': 1.9202898550724639, 'pattern_counts': {'":': 12, '<': 9, '<?xml version=': 0, '>': 11, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5774276', 'n_tokens_mistral': 4653, 'n_tokens_neox': 4430, 'n_words': 867}