树莓派实战:微信机器人(itchat实现)

为您推荐:小红书API     抖音查询API      chatGPT API

背景

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

什么是树莓派?树莓派是一个信用卡大小的单板计算机,ARM架构CPU,可以7×24跑Linux服务器,连接各种扩展硬件,所以发挥想象力,就能做很多有意思的事情。
树莓派实战:微信机器人(itchat实现)

需求

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

itchat简介

itchat是一个开源的微信个人号接口,使用不到三十行的代码,就可以完成一个能够处理所有信息的微信机器人。
github地址:https://github.com/littlecodersh/ItChat

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

下面分点介绍如何实现有趣的功能。

功能实现

1、自动回复

首先得注册消息处理函数,即对不同类型的消息做处理。微信消息分为文本、图片、语音、视频、好友申请等,可通过itchat的Python语法糖来注册不同类型消息的处理函数,有点类似Java里的注解。
如果是文本消息,可以识别其中的关键字,不同的关键字对应不同的逻辑处理。
默认是处理单聊的消息,也可以处理群聊的消息。
下面给出一个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'@%su2005I 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对话还是不错的,如果是自由对话,那大概率前言不搭后语。

树莓派实战:微信机器人(itchat实现)

# 调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') 

树莓派实战:微信机器人(itchat实现)

发起视频

楼主尝试了几个常见的免费视频聊天软件,都无法支持,主要原因是树莓派是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/wechat_robot
除了上面提到的几个功能实现,还增加了健身打卡、睡觉打卡的功能。现在,微信机器人的功能已经越来越丰富了。

# -*- 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'@%su2005%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()  

发表评论

相关文章