緣由
由於我個人有個小博客，然后還錄制了一些課程，所有就建立了對應的微信群，但是微信群的二維碼是變化的，總不能經常換了。所以就想搞一個微信機器人，自動加 微信，自動拉群。廢話不多說，先上圖
效果展示
大家可以試試效果，效果相當棒
需求
其他人可以通過二維碼加我好友 自動通過
好友通過之后 主動發送一些邀請入群鏈接和介紹信
回復 邀請 、加群 可以發送邀請入群鏈接
所有聊天數據都要存儲起來 並且可以通過Web展示
根據用戶輸入信息 回復相關內容
技術選型
Python3
Flask：輕量級的MVC 框架
itchat ：實現微信API相關接口
MySQL：存儲相關微信內容
圖靈機器人：調用API實現主動場景回復
架構
本人沒有使用flask 默認的MVC方式，使用了我自己實現過后的：結構分層明了的MVC框架 。結構如下
每一個文件夾都有自己明確的作用
├── application.py ├── common │ ├── libs │ └── models ├── config │ ├── base_setting.py │ ├── local_setting.py │ ├── production_setting.py ├── controllers │ ├── index.py │ ├── member.py ├── interceptors │ ├── Auth.py │ ├── errorHandler.py ├── jobs │ ├── launcher.py │ ├── movie.py │ └── tasks ├── manager.py ├── requirement.txt ├── static │ ├── js │ └── plugins ├── templates │ ├── common │ ├── index.html │ ├── info.html │ └── member ├── test │ └── apsch_test.py ├── tornado_server.py └── www.py
實現源碼
這里將主要實現源碼如下
# -*- coding: utf-8 -*-
from application import app,db
import itchat,signal,json,time,os,random
import multiprocessing
from ctypes import c_bool, c_int
from common.libs.tuling123.TuLingService import TuLingService
from common.libs.wxbot.WeChatBotService import WeChatBotService
from common.models.queue import QueueList
from common.libs.DateUtil import getCurrentTime
import traceback,requests
'''
python manage.py runjob -m wechatbot/index
'''
class JobTask():
def __init__(self):
pass
def run(self, params):
try:
# 增加重試連接次數
# requests使用了urllib3庫，默認的http connection是keep-alive的，requests設置False關閉。
s = requests.session()
s.keep_alive = False
itchat.auto_login( enableCmdQR=2,hotReload=True )
threads = []
t_queue_consumer = multiprocessing.Process(target = syncConsumer, args=("t_queue_consumer",) )
t_queue_consumer.daemon = True
threads.append( t_queue_consumer )
app.logger.info( 'Parent process {0} is Running'.format( os.getpid() ) )
for i in range( len(threads) ):
threads[i].start()
#初始化所有的朋友，存入隊列，然后進程做事情
queue_msg = {
'type': 'friends',
'data': itchat.get_friends( update=True )
}
global_queue.put(queue_msg)
##必須要把群聊保存到通訊錄才行，不會無法獲取到的
#https://github.com/littlecodersh/ItChat/issues/480
queue_msg = {
'type': 'group',
'data': itchat.get_chatrooms(update=True)
}
global_queue.put(queue_msg)
itchat.run()
except Exception as err:
app.logger.error( "=========itchat:run error========" )
traceback.print_exc()
exit( 15 )
@staticmethod
def exit( signum, frame ):
global sendmail_flag
app.logger.info( "手動退出~~" )
app.logger.info( "signum:%s,frame:%s,sendmail_flag:%s"%( signum,frame,sendmail_flag ) )
if not sendmail_flag:
sendmail_flag = True
#itchat.logout()#執行這個退出之后下次就必須要在掃碼才能登陸
from common.libs.queue.QueueListService import QueueListService
notice_users = QueueListService.getNoticeUsers(ids=[app.config['NOTICE_WECHAT_USER']['family_host_man']])
data = {
"msg": "微信機器人 倒了，需要人為干預，修復好~~",
"title": "【提醒】微信機器人 倒了",
"notice_users": notice_users
}
QueueListService.addQueueMsg(queue_name="reminder", data=data, type=2)
exit( 15 )
# http://itchat.readthedocs.io/zh/latest/
# 加入圖靈機器人：https://www.kancloud.cn/turing/web_api/522992
# 關於進程的文章：https://mozillazg.com/2017/07/python-how-to-generate-kill-clean-zombie-process.html
@itchat.msg_register(itchat.content.INCOME_MSG)
def handle_msg(msg):
queue_msg = {
'type':'msg',
'data':msg
}
global_queue.put( queue_msg )
if msg.Type == itchat.content.TEXT and msg.FromUserName != "newsapp":
if msg.Type == itchat.content.TEXT and ( "邀請" in msg.Text or "加群" in msg.Text ) :
# 同時發送加群邀請鏈接
itchat.add_member_into_chatroom( get_group_id("編程浪子小天地"), [{'UserName':msg.FromUserName }],
useInvitation=True )
return None
'''
需要過濾掉幾種情況，例如自己發送的不要自己回復自己了
'''
tmp_msg = []
tmp_msg.append( TuLingService.getRobotAnswer( msg.Text ) )
tmp_msg.extend( getAd() )
msg.user.send( "\n".join( tmp_msg ) )
elif msg.Type == itchat.content.FRIENDS:
msg.user.verify()
tmp_msg = [
"歡迎進入編程浪子機器人世界~~",
"你可以輸入想對我說的話，會有彩蛋",
"個人博客：http://www.54php.cn",
"回復 邀請 、 加群 關鍵字 可以加入群",
"獲取慕課課程賬號，請關注公眾號"
]
tmp_msg.extend( getAd( is_rand = False ) )
msg.user.send( "\n".join( tmp_msg ) )
itchat.send_image( app.root_path + "/web/static/images/wechat/coderonin.jpg" ,msg['RecommendInfo']['UserName'] )
itchat.send_image( app.root_path + "/web/static/images/wechat/mina.jpg" ,msg['RecommendInfo']['UserName'] )
#同時發送加群邀請鏈接
itchat.add_member_into_chatroom( get_group_id("編程浪子小天地"), [{'UserName': msg['RecommendInfo']['UserName'] }], useInvitation=True)
elif msg.Type in [ itchat.content.SYSTEM, itchat.content.NOTE ]:
#note表示添加好友通過后的備注內容
pass
else:
if msg.Type == "Sharing" and msg.Text == "邀請你加入群聊":
return None
tmp_msg = [ "我還是個孩子~~無法識別你的發的什么哦" ]
tmp_msg.extend(getAd())
msg.user.send( "\n".join( tmp_msg ) )
return None
@itchat.msg_register(itchat.content.INCOME_MSG, isGroupChat=True)
def handle_group_msg(msg):
app.logger.info("type:{0},text:{1}".format(msg.Type, msg.Text))
app.logger.info(msg)
return None
def getAd( is_rand = True ):
ad_urls = [
"http://t.cn/AiK3JhDK",
"http://t.cn/AiK3JLD5",
"http://t.cn/AiK3JqKX",
"http://t.cn/AiK3JfqQ",
"http://t.cn/AiK3JXWa",
"http://t.cn/AiK3JNvY",
"http://t.cn/AiKS9XZR",
"http://t.cn/AiK3JQe3",
"http://t.cn/AiK3JuJi",
"http://t.cn/AiK3JeOC",
"http://t.cn/AiK3ivyk",
"http://t.cn/AiK3izxl",
"http://t.cn/AiK3iLVH"
]
tmp_msg = [
"",
"==================",
"望大家多多支持，關注公眾號：CodeRonin",
"支持浪子點擊文章最下面AD：" + ad_urls[ random.randint(1, len( ad_urls ) ) - 1 ]
]
if is_rand and random.randint(1,10) < 8:
tmp_msg = []
return tmp_msg
# 獲得群聊id
def get_group_id(group_name):
group_list = itchat.search_chatrooms(name=group_name)
return group_list[0]['UserName']
#捕獲退出信號，例如control + c
signal.signal(signal.SIGINT, JobTask.exit )
signal.signal(signal.SIGTERM, JobTask.exit )
'''
存儲到數據庫我使用線程處理，這樣異步不影響返回
queue模塊有三種隊列及構造函數:
Python queue模塊的FIFO隊列先進先出。 class queue.Queue(maxsize)
LIFO類似於堆，即先進后出。 class queue.LifoQueue(maxsize)
還有一種是優先級隊列級別越低越先出來。 class queue.PriorityQueue(maxsize)
'''
manager = multiprocessing.Manager()
global_queue = multiprocessing.Queue()
sendmail_flag = multiprocessing.Value( c_bool,False )
def syncConsumer(name):
with app.app_context(): # http://flask-sqlalchemy.pocoo.org/2.3/contexts/
while True:
try:
app.logger.info( "syncConsumer,pid:%s"%( os.getpid() ) )
queue_msg = global_queue.get()
# get_nowait()不會阻塞等待
app.logger.info('[%s]取到了[%s]' % (name, queue_msg['type']))
if queue_msg['type'] in [ "friends","group" ]:
type = 2 if queue_msg['type'] == "group" else 1
WeChatBotService.initContact( queue_msg['data'],type )
elif queue_msg['type'] == "msg":
WeChatBotService.addHistory(queue_msg['data'])
time.sleep(1)
except Exception as err:
traceback.print_exc()
exit( 15 )
總結
通過借助於 itchat 和圖片機器人 ，然后基於 Python3 + Flask MVC + MySQL 就可以實現一個自己的微信機器人 ，過程其實相當簡單，但是需要大家對 Python3 + Flask MVC 有一定的了解，並且本人根據項目經驗總結出來了一套個性化的 ：結構分層明了、高並發 MVC框架。
廣告
如果您對Python Flask學習感興趣 ，而您整好基礎薄弱，可以看看我的兩個課程，僅供您參考
Python Flask 從入門到精通： https://coding.imooc.com/class/399.html
Python Flask構建微信小程序訂餐系統： https://coding.imooc.com/class/265.html
原文地址： 使用Python Flask 開發微信機器人
標簽： 微信 wechat 機器人 robot python flask mooc 慕課