关于raspberry-pi:树莓派实战微信机器人itchat实现

56次阅读

共计 6478 个字符,预计需要花费 17 分钟才能阅读完成。

背景

楼主有一台树莓派 4B 开发板(8G 内存版),是目前的顶配机型。这一年来的业余时间,除了写 Java、架构方面的文章,也陆续折腾了不少树莓派上的好玩小我的项目,在此新开一个树莓派实战的文章系列,分享给粉丝和读者。

什么是树莓派?树莓派是一个信用卡大小的单板计算机,ARM 架构 CPU,低功耗,能够 7×24 跑 Linux 服务器,连贯各种扩大硬件,所以施展想象力,就能做很多有意思的事件。

需要

你有没有想过,领有一个微信机器人,能够主动回复、AI 聊天、定时发送天气预报、管制摄像头等等。应用树莓派 + 开源库 itchat,就能实现上述所有需要。
为什么强调要用树莓派呢?因为它能 7×24 在线,能够把 itchat 客户端当作一个不停服的 server。
特地阐明:本文仅供学习用,请勿用于任何商业和其它用处。

itchat 简介

itchat 是一个开源的微信集体号接口,应用不到三十行的代码,就能够实现一个可能解决所有信息的微信机器人。
github 地址:https://github.com/littlecode…

你肯定对原理感到好奇。其实能够概括为一句话:itchat 实质上是一个微信网页版客户端,它实现了微信网页版的协定 / 语义,通过 http 来通信。具体源码能够看 components 包里的文件。

上面分点介绍如何实现乏味的性能。

性能实现

1、主动回复

首先得注册音讯处理函数,即对不同类型的音讯做解决。微信音讯分为文本、图片、语音、视频、好友申请等,可通过 itchat 的 Python 语法糖来注册不同类型音讯的处理函数,有点相似 Java 里的注解。
如果是文本音讯,能够辨认其中的关键字,不同的关键字对应不同的逻辑解决。默认是解决单聊的音讯,也能够解决群聊的音讯。
运行程序后,会弹出一个二维码,扫码即可登录,而后 itchat 程序就跑起来了。另外需注意,发消息给本人是没用的,变通的方法是发消息给文件传输助手 filehelper,成果等同于发消息给本人。
上面给出一个 demo,并加以正文。

import itchat, time
from itchat.content import *

# 注册音讯处理函数,回复文本、地图、名片、备注、分享类型的音讯
@itchat.msg_register([TEXT, MAP, CARD, NOTE, SHARING])
def text_reply(msg):
    # 回复以下音讯:音讯类型,音讯内容文本
    itchat.send('%s: %s' % (msg.type, msg.text))
    # 依据不同的关键字,回复不同的音讯
    if '你好' in msg.text:
        itchat.send('你好啊')
    elif '拜拜' in msg.text:
        itchat.send('下次聊')

# 注册音讯处理函数,当收到图片、语音、附件、视频类型的音讯时,下载内容
@itchat.msg_register([PICTURE, RECORDING, ATTACHMENT, VIDEO])
def download_files(msg):
    # 下载文件
    msg.download(msg.fileName)
    typeSymbol = {
        PICTURE: 'img',
        VIDEO: 'vid', }.get(msg.type, 'fil')
    return '@%s@%s' % (typeSymbol, msg.fileName)

# 注册音讯处理函数,解决好友申请音讯
@itchat.msg_register(FRIENDS)
def add_friend(msg):
    # 主动通过对方的好友申请
    msg.user.verify()
    # 而后发送问候语
    msg.user.send('Nice to meet you!')

# 下面几个都是单聊,加上 isGroupChat=True 就能解决群聊音讯
@itchat.msg_register(TEXT, isGroupChat=True)
def text_reply(msg):
    # 当在群聊被 at 时才回复,个别都会加上此条件,否则可能回复群内所有音讯
    if msg.isAt:
        # 回复时,也 at 对应的人音讯
        msg.user.send(u'@%s\u2005I received: %s' % (msg.actualNickName, msg.text))

# 会弹出一个二维码,扫码即可登录
itchat.auto_login(True)
# 运行 itchat 客户端,debug=True 会打印日志
itchat.run(True)

2、AI 聊天

有了第 1 步的根底,要实现 AI 聊天,就须要引入另外的 AI 本地库、或者在线 API 了,应用在线 API 更简略,只须要管制传参、解析响应即可。楼主应用了一个叫青云客的 API,可收费应用(本人简略试用的前提下,非商用),带关键字命令的 AI 对话还是不错的,如果是自在对话,那大概率前言不搭后语。

# 调 API 来进行 AI 聊天,只有一个文本参数
def ai_chat(msg):
    url = 'http://api.qingyunke.com/api.php?key=free&appid=0&msg=%s' % msg
    response = requests.get(url)
    return response.json()["content"].replace('{br}', '\n') # 响应里的换行是{br},替换为微信可辨认的 \n 换行

3、定时发送天气预报

有了第 2 步的根底,要获取天气预报信息,只须要在 AI 聊天的申请里传某地天气即可,比方:上海天气、北京天气。当然,你也能够通过爬天气预报网页的字段,失去更详尽的天气预报信息,此处就不多探讨了。
定时发天气预报,要解决 2 个关键问题。

  • 一是如何执行定时工作。此处选用 Python 库 apscheduler。当然,也能够写一个 Python 脚本,而后通过操作系统的 crontab 在指定的工夫执行该脚本,不过还有更优雅的形式,在 Python 主程序内启动定时工作。能够应用 Python 库 apscheduler 来实现定时工作的调度,相似于 Java 的 ScheduledThreadPool。
  • 二是如何发送音讯到指定的群。itchat 曾经提供了便捷的 API 来依据群名搜寻具体的群。
from apscheduler.schedulers.blocking import BlockingScheduler

# 发送天气预报信息到群里
def weather_report():
    msg = ai_chat('上海天气')
    # 获取所有群聊
    itchat.get_chatrooms(update=True)
    # 依据群名,搜寻具体的群
    chatrooms = itchat.search_chatrooms(name='< 此处改为理论的群名 >')
    chatroom = itchat.update_chatroom(chatrooms[0]['UserName'])
    # 发送音讯,到指定的群
    itchat.send_msg(msg=msg,toUserName=chatroom['UserName'])

if __name__ == '__main__':
    itchat.auto_login(hotReload=True)
    # itchat 启动后是否阻塞,此处改为否(默认为是),相当于 itchat 在新启动的线程中运行,不阻塞主程序
    itchat.run(blockThread=False)

    # 定时工作
    scheduler = BlockingScheduler()
    # 指定在每天早上 9 点调用 weather_report 函数
    scheduler.add_job(weather_report, 'cron', day_of_week='*', hour=9, minute=0, second=0)
    scheduler.start()

4、管制摄像头,拍照、视频看看家里

树莓派 4B 有 2 个 USB 3.0 高速接口、2 个 USB 2.0 接口,只须要其中一个连贯上 USB 摄像头即可,个别 2.0 接口即可,3.0 接口留给外接硬盘。
想要通过摄像头看到家里,要解决的关键问题是,应用什么拍照软件?应用什么视频聊天软件?

拍照

能够应用 fswebcam 来拍照,能够指定图像分辨率,也能够不指定,默认的分辨率较低。
装置:sudo apt install fswebcam

    img_file = '%d.jpg' % timestamp
    # 调用 fswebcam 拍照
    os.system('fswebcam %s' % img_file)
    # 发送照片至本人的文件传输助手,因为通常发给本人会失败
    itchat.send_image(img_file, toUserName='filehelper')

发动视频

楼主尝试了几个常见的免费视频聊天软件,都无奈反对,次要起因是树莓派是 ARM CPU 架构,支流软件基本上只在 amd64、x86 CPU 架构下发行。比方 QQ、Skype、网页版 Jitsi Meet 等都无奈发动视频聊天。
最终,楼主发现了一个较为完满的解决方案,就是应用 linphone:

  • 发动视频:在树莓派上装置并关上 linphone 程序,也在手机上安装并关上 linphone app。这样通过微信就能够让树莓派上的 linphone 发动视频通话,手机端就能接到电话了。
  • 挂断视频:须要通过微信机器人,在树莓派上被动退出 linphone,否则后续不能持续发动视频。
    下载最新的 linphone 可能无奈失常工作,得应用 sudo apt install linphone 来装置旧的稳定版。
    # 先退出 linphone(如以后有在运行),再启动 linphone
    os.system('linphonecsh exit; linphonecsh init -V -c .linphonerc')
    time.sleep(1)
    # 应用 linphone 命令行拨打视频通话
    os.system('linphonecsh generic"call < 替换成理论的 linphone 账号,需注册 >"')

残缺代码

以下是楼主写的几个实用例子,并加以正文。
残缺代码已上传至 github:https://github.com/topcoding/…
除了下面提到的几个性能实现,还减少了健身打卡、睡觉打卡的性能。当初,微信机器人的性能曾经越来越丰盛了。

# -*- coding: utf-8 -*-

import itchat
import sqlite3
import os
import time
import requests
from apscheduler.schedulers.blocking import BlockingScheduler


PUNCH_TYPE_WORKOUT = 1
PUNCH_TYPE_SLEEP = 2

ai_chat_switch = True

AI_CHATROOM_WHITELIST = ['< 替换成理论的群名 >']


def save_db(punch_type, owner, timestamp = None):
    conn = sqlite3.connect('punch-card.db')
    cursor = conn.cursor()
    if timestamp is None:
        punch_time = (int) (time.time())
    else:
        punch_time = timestamp
    cursor.execute("insert into punch_card(punch_type, owner, updated_at) values(%d,'%s', %d)"
                   % (punch_type, owner, punch_time))
    conn.commit()
    conn.close()

@itchat.msg_register(itchat.content.TEXT)
def text_reply(msg):
    print(msg)
    timestamp = (int) (time.time())
    global ai_chat_switch
    if msg.text == '健身打卡':
        save_db(PUNCH_TYPE_WORKOUT, msg.User.NickName, timestamp)
        itchat.send('%s,您好,您于 %s 健身打卡胜利' % (msg.User.NickName, time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())), toUserName='filehelper')
    elif msg.text == '睡觉打卡':
        save_db(PUNCH_TYPE_SLEEP, msg.User.NickName, timestamp)
        itchat.send('%s,您好,您于 %s 睡觉打卡胜利' % (msg.User.NickName, time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())), toUserName='filehelper')
    elif msg.text == '拍照':
        img_file = '%d.jpg' % timestamp
        os.system('fswebcam %s' % img_file)
        itchat.send_image(img_file, toUserName='filehelper')
    elif msg.text == '看看家里':
        os.system('linphonecsh exit; linphonecsh init -V -c .linphonerc')
        time.sleep(1)
        os.system('linphonecsh generic"call < 替换成理论的 linphone 账号,需注册 >"')
    elif msg.text == '挂断视频':
        os.system('linphonecsh exit')
    elif msg.text == '群聊':
        ai_chat_switch = True
    elif msg.text == '群聊勾销':
        ai_chat_switch = False
    else:
        # do nothing
        pass

@itchat.msg_register('Text', isGroupChat = True)
def group_reply(msg):
    if ai_chat_switch and msg['isAt'] and msg['User']['NickName'] in AI_CHATROOM_WHITELIST:
        print(msg)
        return u'@%s\u2005%s' % (msg['ActualNickName'], ai_chat(msg))

def ai_chat(msg):
    url = 'http://api.qingyunke.com/api.php?key=free&appid=0&msg=%s' % msg
    response = requests.get(url)
    return response.json()["content"].replace('{br}', '\n')

def weather_report():
    msg = ai_chat('上海天气')
    itchat.get_chatrooms(update=True)
    chatrooms = itchat.search_chatrooms(name='< 替换成理论的群名 >')
    chatroom = itchat.update_chatroom(chatrooms[0]['UserName'])
    itchat.send_msg(msg=msg,toUserName=chatroom['UserName'])


if __name__ == '__main__':
    itchat.auto_login(hotReload=True)
    itchat.run(blockThread=False)

    scheduler = BlockingScheduler()
    scheduler.add_job(weather_report, 'cron', day_of_week='*', hour=9, minute=0, second=0)
    scheduler.start()

更多例子

能够参考 itchat 提供的教程文档:https://github.com/littlecode…

正文完
 0