Pulpcode

捕获，搅碎，拼接，吞咽

0%

如何实现一个微信机器人

发表于 分类于 Valine：

之前公司有一个需求，需要搭建一个微信机器人，查阅了相关资料，整理了一些Demo之后，我大概实现了一个python的微信机器人聊天程序，这里就把经验拿出来分享下。

基本原理

首先你要明白，微信并没有给你提供一套接口，一套文档，让你开发聊天机器人。如果你想实现一个机器人，能够被拉到群里，与群里的人聊天，就必须用点hack的办法。

这个原理就是web版本的微信，想想你登录web版的微信，在浏览器输入地址后，先会给你返回一个二维码。然后你用你的手机扫描了此二维码，接着就登录成功了。

这个扫二维码登录的过程，其实就是一个授权的过程，即你授权给web界面，使得可以通过web页面，用http的方式与微信服务器建立交互。所以微信机器人的实现方式就是基于此。你创建一个账号，然后这个账号就是微信机器人，你通过扫码的方式登录，使得可以通过http的方式与微信服务器建立交互。就这样，我们的服务托管了你的机器人账户，从而与群里的人进行交流，解析他们的聊天记录，匹配关键字，生成相应的回复信息。

交互流程

首先想想和微信服务器建立交互的时序图。

weixin01

所以当我们要用一个服务来托管你的微信机器人，就需要维护一个与微信服务器通信的HTTP Session。

weixin02

这里要注意的是，http是上下文无关的，所以你与微信服务器的交互必须要通过cookie建立上下文，还有http协议是一问一答的形式，所以服务器并不能主动推送数据给客户端，所以你只能使用轮训的方式，从微信服务器获取消息信息。

基于python tornado服务的实现思路

网上有一个现成的库，提供tornado异步客户端的Session方式，其实就是使用RequestsCookieJar,这个库来保存Cookie。
我通过这个库来包装了一个Session，方便我调用get和post。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
from httpclient_session import Session

class RequestSession(object):
    """
    一个能维护异步请求的session,传递cookie.
    """
    def __init__(self):
        self._session = Session()

    @tornado.gen.coroutine
    def retry_request_get(self, url, timeout=None):
        """
        此请求如果失败,将会重试,直到成功为止(最多三次)
        :return:
        """
        try:
            headers = HTTPHeaders({'User-Agent': 'Mozilla/5.0 (X11; Linux i686; U;) Gecko/20070322 Kazehakase/0.4.5'})
            request = HTTPRequest(
                url=url,
                headers=headers,
                connect_timeout=timeout,
                request_timeout=timeout
            )
            response = yield self._session.fetch(request)
        except Exception:
            logger.error('request error: %s' % url, exc_info=True)
            raise tornado.gen.Return((False, None))
        else:
            raise tornado.gen.Return((True, response.body))

    @tornado.gen.coroutine
    def retry_request_post(self, url, body, timeout=None):
        """
        此请求如果失败,将会重试,直到成功为止(最多三次)
        :return:
        """
        try:
            headers = HTTPHeaders({'User-Agent': 'Mozilla/5.0 (X11; Linux i686; U;) Gecko/20070322 Kazehakase/0.4.5'})
            request = HTTPRequest(
                url=url,
                body=body,
                method="POST",
                headers=headers,
                connect_timeout=timeout,
                request_timeout=timeout
            )
            response = yield self._session.fetch(request)
        except Exception:
            logger.error('request error: %s' % url, exc_info=True)
            raise tornado.gen.Return((False, None))
        else:
            raise tornado.gen.Return((True, response.body))

请求加入的头部参数，是为了伪装成浏览器，那么对于这个例子，我创建的每一个机器人类，都会维护一个自己的RequestSession。

交互过程

先需要从一个get请求获取uuid。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@tornado.gen.coroutine
def get_uuid(self):
    """
    获取微信web登录验证所需的uuuid
    :return:
    """
    url = https://login.weixin.qq.com/jslogin
    params = {
            'appid': 'wx782c26e4c19acffb',
            'fun': 'new',
            'lang': 'zh_CN',
            '_': int(time.time()) * 1000 + random.randint(1, 999),
    }
    url = "%s?%s" % (url, urllib.urlencode(params))
    logger.debug("fetch uuid: %s" % url)
    ok, body = yield self._session.retry_request_get(url)
    regx = r'window.QRLogin.code = (\d+); window.QRLogin.uuid = "(\S+?)"'

    if ok:
        pm = re.search(regx, body)
        if pm:
            code = pm.group(1)
            uuid = pm.group(2)
            if code == "200":
                self._uuid = uuid
                raise tornado.gen.Return(True)
    raise tornado.gen.Return(False)

然后通过这个uuid，配上一个url，来生成一个用于登录的微信二维码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def gen_login_qrcode_png(self):
    """
    生成二维码登录图片,返回一个io流对象
    :return:
    """
    url = 'https://login.weixin.qq.com/l/'
    logger.debug("login url: %s" % url)
    qr = qrcode.QRCode(
        version=1,
        error_correction=qrcode.constants.ERROR_CORRECT_L,
        box_size=10,
        border=4,
    )
    qr.add_data(url)
    qr.make(fit=True)

    out = BytesIO()
    qr_img = qr.make_image()
    qr_img.save(out, 'PNG')
    return out

之后你的服务就需要一直不停的轮训微信服务器，来知道客户登录的状态：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@tornado.gen.coroutine
def wait_for_login(self):
    """
    等待用户登录
    :return:
    """
    TIP_WAIT_FOR_SCAN_QR_CODE = 1
    # 等待用户扫描二维码
    TIP_WAIT_CONFIRM_LOGIN = 0
    # 等待用户确认登录
    MAX_RETRY_TIMES = 10
    # 最大尝试次数
    TRY_LATER_SECS = 1
    # sleep时间(秒)
    UNKONWN = 'unkonwn'
    # 未知
    SUCCESS = '200'
    # 200: confirmed
    SCANED = '201'
    # 201: scaned
    TIMEOUT = '408'
    # 408: timeout

    tip = TIP_WAIT_FOR_SCAN_QR_CODE
    retry_time = MAX_RETRY_TIMES
    code = UNKONWN

    while retry_time > 0:
        url = "%s?tip=%s&uuid=%s&_=%s" % (https://login.weixin.qq.com/cgi-bin/mmwebwx-bin/login, tip, self._uuid, int(time.time()))
        logger.debug("get login status url: %s" % url)
        ok, body = yield self._session.retry_request_get(url)
        logger.debug("get login status response: %s" % body)
        if ok:
            param = re.search(r'window.code=(\d+);', body)
            code = param.group(1)
            if code == SCANED:
                tip = TIP_WAIT_CONFIRM_LOGIN
            elif code == SUCCESS:
                param = re.search(r'window.redirect_uri="(\S+?)";', body)
                self._redirect_uri = param.group(1) + '&fun=new'
                self._base_uri = self._redirect_uri[:self._redirect_uri.rfind('/')]
                temp_host = self._base_uri[8:]
                self._base_host = temp_host[:temp_host.find("/")]
                logger.info(u"微信登录成功")
                raise tornado.gen.Return(True)
            elif code == TIMEOUT:
                tip = TIP_WAIT_FOR_SCAN_QR_CODE
                retry_time -= 1
                yield tornado.gen.sleep(TRY_LATER_SECS)
            else:
                tip = TIP_WAIT_FOR_SCAN_QR_CODE
                retry_time -= 1
                yield tornado.gen.sleep(TRY_LATER_SECS)

    raise tornado.gen.Return(code == SUCCESS)

之后就要根据之前获取到的redirect_uri，来获取一些，skey，sid，in，pass_ticket等字段，这些参数每次与微信服务器交互都要用到。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
@tornado.gen.coroutine
def login(self):
    """
    登录并获取相关信息
    :return:
    """
    if len(self._redirect_uri) < 4:
        logger.error("Login failed due to network problem, please try again.")
        raise tornado.gen.Return(False)

    ok, body = yield self._session.retry_request_get(self._redirect_uri)
    logger.debug("login info: %s" % body)
    if ok:
        doc = xml.dom.minidom.parseString(body)
        root = doc.documentElement

        for node in root.childNodes:
            if node.nodeName == 'skey':
                self._skey = node.childNodes[0].data
            elif node.nodeName == 'wxsid':
                self._sid = node.childNodes[0].data
            elif node.nodeName == 'wxuin':
                self._uin = node.childNodes[0].data
            elif node.nodeName == 'pass_ticket':
                self._pass_ticket = node.childNodes[0].data

        if '' in (self._skey, self._sid, self._uin, self._pass_ticket):
            raise tornado.gen.Return(False)

        self._base_request["Uin"] = self._uin
        self._base_request["Sid"] = self._sid
        self._base_request["Skey"] = self._skey
        self._base_request["DeviceID"] = self._device_id

        raise tornado.gen.Return(True)
    else:
        raise tornado.gen.Return(False)

@tornado.gen.coroutine
def main(self):
    """
    机器人的运行主逻辑
    TODO: 这里的主逻辑可能不对
    :return:
    """
    if self._main_lock == False:
        self._main_lock = True
        if self.status == BaseRobot.STATUS_LIVE:
            code, msg = yield self._wxClient.dispatch_msg()
            if code == WxClient.WX_RETURN_CODE_LOGINOUT or code == WxClient.WX_RETURN_CODE_RELOGINOTHER:
                self.status = BaseRobot.STATUS_DEAD
                logger.info("robot stop...")

            self.send_message()
        else:
            logger.info("robot dead.")
        self._main_lock = False

@tornado.gen.coroutine
def dispatch_msg(self):
    """
    处理消息(根据不同消息类型,分类处理)
    :return:
    """
    retcode, selector = yield self.sync_check()
    if retcode == WxClient.WX_RETURN_CODE_LOGINOUT:
        raise tornado.gen.Return((WxClient.WX_RETURN_CODE_LOGINOUT, None))
    elif retcode == WxClient.WX_RETURN_CODE_RELOGINOTHER:
        raise tornado.gen.Return((WxClient.WX_RETURN_CODE_RELOGINOTHER, None))
    elif retcode == WxClient.WX_RETURN_CODE_MESSAGE:
        if selector == WxClient.SELECTOR_NEW_MESSAGE:
            r = yield self.sync()
            if r is not None:
                self.handle_msg(r)
                raise tornado.gen.Return((WxClient.WX_RETURN_CODE_MESSAGE, None))
        elif selector == WxClient.SELECTOR_UNKNOWN:
            r = yield self.sync()
            if r is not None:
                self.handle_msg(r)
                raise tornado.gen.Return((WxClient.WX_RETURN_CODE_MESSAGE, None))
        elif selector == WxClient.SELECTOR_ADDRESS_LIST_UPDATE:
            r = yield self.sync()
            if r is not None:
                raise tornado.gen.Return((WxClient.WX_RETURN_CODE_MESSAGE, None))
        elif selector == WxClient.SELECTOR_MAYBE_RED_PACKET:
            r = yield self.sync()
            if r is not None:
                raise tornado.gen.Return((WxClient.WX_RETURN_CODE_MESSAGE, None))
        elif selector == WxClient.SELECTOR_USE_PHONE_OPERATE:
            r = yield self.sync()
            if r is not None:
                self.handle_msg(r)
                raise tornado.gen.Return((WxClient.WX_RETURN_CODE_MESSAGE, None))
        elif selector == WxClient.SELECTOR_ZERO:
            pass
        else:
            logger.warning("unknown retcode, selector: %s, %s" % (retcode, selector))
            raise tornado.gen.Return((WxClient.WX_RETURN_CODE_UNKNOWN, None))

    logger.warning("unknown sync check: %s, %s" % (retcode, selector))
    raise tornado.gen.Return((WxClient.WX_RETURN_CODE_UNKNOWN, None))

之后的代码只需要从服务器不停的轮训，获取数据，在根据消息类型，做不同的处理就行了。

注意事项

首先这个服务其实写起来比我想象的要复杂，因为在服务器维护每个机器人的session，然后不停的轮训这件事，一个服务器还好说，如果部署到多台服务器，那就会麻烦很多，比如你的session一定要入库的，否则一重启什么都没了，那么如果入库，那由哪个服务器的哪个线程用来与服务器建立通信呢，所以这里我只是做了单机部署，因为这篇博客的意图，其实通过微信机器人程序，总结HTTP的一些知识。

还有就是这个服务可能会很不稳定，因为我并没有做过测试来验证，web版的微信到底能登录多长时间。所以它有可能在登录一段时间后就掉线了，而且这种“破解”类的程序，微信随时都有可能封你，尤其你使用了类似“加好友”这样的接口。

不仅仅是每次登录获取的uuid和之后的各种token不一样，实际上你每次重新登录，用户的id，和组的id，甚至你自己的id，都是重新生成的。

微信web是单点登录的，所以你在另一个web页面登录微信的话，这个session就会被微信服务器关闭的。