微信自動(dòng)聊天機(jī)器狗,配置ChatGPT,比Siri還智能!
“大家好，我是TheWeiJun；
最近看見(jiàn)微信里各個(gè)群聊都在聊 ChatGPT，甚至有的大佬們都把 C hatGPT接入了微信群聊，于是就有粉絲來(lái)找小編，希望能出一期 C hatGPT的文章； 故今天這篇文章我將手把手教大家如何實(shí)現(xiàn)并自定義自己的聊天機(jī)器人。
碼字不易，在閱讀的同時(shí)記得給我一個(gè) star！
”
“特別聲明： 本公眾號(hào)文章只作為學(xué)術(shù)研究，不作為其它不法用途；如有侵權(quán)請(qǐng)聯(lián)系作者刪除。
”
OpenAI推出了ChatGPT，小明想把ChatGPT接入到微信群聊中，平日里閑下來(lái)沒(méi)事的時(shí)候，和GPT聊聊八卦、談?wù)勅松?、增長(zhǎng)一下學(xué)習(xí)知識(shí)。甚至在女朋友不知情的情況下，偷偷看看漂亮小姐姐的圖片和視頻；而這些功能都希望能通過(guò)微信機(jī)器人去實(shí)現(xiàn)。
為了滿(mǎn)足小明同學(xué)的要求，今天給大家分享一個(gè)ChatGPT與itchat打造的全網(wǎng)最強(qiáng)機(jī)器狗。
相信我，一定要看到最后，你肯定不會(huì)后悔。
機(jī)器狗id名片關(guān)注走起：
一、前言介紹
- 什么是chatGPT？
ChatGPT是美國(guó)人工智能研究實(shí)驗(yàn)室OpenAI新推出的一種人工智能技術(shù)驅(qū)動(dòng)的自然語(yǔ)言處理工具，使用了Transformer神經(jīng)網(wǎng)絡(luò)架構(gòu)，也是GPT-3.5架構(gòu)，這是一種用于處理序列數(shù)據(jù)的模型，擁有語(yǔ)言理解和文本生成能力，尤其是它會(huì)通過(guò)連接大量的語(yǔ)料庫(kù)來(lái)訓(xùn)練模型，這些語(yǔ)料庫(kù)包含了真實(shí)世界中的對(duì)話(huà)，使得ChatGPT具備上知天文下知地理，還能根據(jù)聊天的上下文進(jìn)行互動(dòng)的能力，做到與真正人類(lèi)幾乎無(wú)異的聊天場(chǎng)景進(jìn)行交流。ChatGPT不單是聊天機(jī)器人，還能進(jìn)行撰寫(xiě)郵件、視頻腳本、文案、翻譯、代碼等任務(wù)。
- 什么是itchat？
itchat是一個(gè)開(kāi)源的微信個(gè)人號(hào)接口，使用python調(diào)用微信從未如此簡(jiǎn)單。使用不到三十行的代碼，你就可以完成一個(gè)能夠處理所有信息的微信機(jī)器人。當(dāng)然，該api的使用遠(yuǎn)不止一個(gè)機(jī)器人，如今微信已經(jīng)成為了個(gè)人社交的很大一部分。
二、環(huán)境準(zhǔn)備
為了實(shí)現(xiàn)一個(gè)微信機(jī)器狗，我們需要準(zhǔn)備如下的環(huán)境和工具包：
- 微信號(hào)一個(gè)，需要實(shí)名認(rèn)證（itchat會(huì)獲取登錄后的skey，不然無(wú)法登錄成功）
- 安裝itchat第三方工具包
- Python環(huán)境為Python3；最好是3.7以上版本
- 有服務(wù)器的可以部署到服務(wù)器，沒(méi)有的本地環(huán)境也可以
附上環(huán)境安裝命令：
pip3 install itchat-uos==1.5.0.dev0
三、ChatGPT 接入
由于 ChatGPT 有種種限制，比如國(guó)內(nèi)無(wú)法訪(fǎng)問(wèn)，使用需要額外申請(qǐng) ChatGPT 賬號(hào)，不科學(xué)使用 API Key 會(huì)導(dǎo)致封號(hào)等等，為了讓大家更便捷地對(duì)接 ChatGPT，這里推薦一個(gè)接口 - 來(lái)自 zhishuyun.com。
有了這個(gè)接口，你不需要再自己擁有 ChatGPT 賬號(hào)、API Key 等各種麻煩的事情，對(duì)接也十分方便。
具體的接口申請(qǐng)方式可以參考文檔：https://data.zhishuyun.com/documents/b91e6309-f549-4d98-b92f-51cfbfb1dad7[1]
有了它，通過(guò)簡(jiǎn)單的幾行代碼便可以實(shí)現(xiàn)調(diào)用，代碼如下：
import requests
url = 'https://api.zhishuyun.com/chatgpt?token={token}'
headers = {
'content-type': 'application/json',
'accept': 'application/json'
}
body = {
"question": "如何學(xué)好英語(yǔ)"
}
r = requests.post(url, headers=headers, json=body, verify=False)
print(r.json())
輸出效果如下：
{'answer': '學(xué)好英語(yǔ)需要付出持續(xù)的努力和時(shí)間，以下是一些建議：\n\n1. 建立一個(gè)有效的學(xué)習(xí)計(jì)劃：根據(jù)自己的學(xué)習(xí)目標(biāo)，安排每周的學(xué)習(xí)時(shí)間表，并制定學(xué)習(xí)計(jì)劃，包括聽(tīng)、說(shuō)、讀、寫(xiě)等方面的練習(xí)。\n\n2. 多聽(tīng)多說(shuō)：聽(tīng)英語(yǔ)廣播、電視節(jié)目、電影等可以提高聽(tīng)力和口語(yǔ)能力。同時(shí)，多練習(xí)口語(yǔ)，積極參加英語(yǔ)角、口語(yǔ)班等活動(dòng)。\n\n3. 多讀多寫(xiě)：讀英語(yǔ)書(shū)籍、報(bào)紙、雜志等可以提高閱讀能力，寫(xiě)英語(yǔ)作文、日記、郵件等可以提高寫(xiě)作能力。同時(shí)，可以找英語(yǔ)老師或者朋友幫忙修改作文。\n\n4. 學(xué)習(xí)語(yǔ)法和詞匯：掌握英語(yǔ)語(yǔ)法和詞匯是學(xué)好英語(yǔ)的基礎(chǔ)，可以通過(guò)學(xué)習(xí)語(yǔ)法和背單詞來(lái)提高自己的英語(yǔ)水平。\n\n5. 創(chuàng)造英語(yǔ)環(huán)境：在日常生活中創(chuàng)造英語(yǔ)環(huán)境，例如使用英語(yǔ)APP、參加英語(yǔ)俱樂(lè)部等，可以提高英語(yǔ)應(yīng)用能力。\n\n6. 堅(jiān)持不懈：學(xué)好英語(yǔ)需要長(zhǎng)期的堅(jiān)持和努力，不要輕易放棄。同時(shí)，要保持積極的心態(tài)，相信自己能夠?qū)W好英語(yǔ)。\n\n總之，學(xué)好英語(yǔ)需要綜合的學(xué)習(xí)策略和方法，同時(shí)需要持續(xù)的努力和堅(jiān)持。'}
ChatGPT接口OK后，我們把它和itchat進(jìn)行關(guān)聯(lián)，相關(guān)代碼如下所示：
import itchat
import json
from itchat.content import *
from channel.channel import Channel
from concurrent.futures import ThreadPoolExecutor
from common.log import logger
from common.tmp_dir import TmpDir
from config import conf
import requests
import io
thread_pool = ThreadPoolExecutor(max_workers=8)
@itchat.msg_register(TEXT)
def handler_single_msg(msg):
WechatChannel().handle_text(msg)
return None
@itchat.msg_register(TEXT, isGroupChat=True)
def handler_group_msg(msg):
WechatChannel().handle_group(msg)
return None
@itchat.msg_register(VOICE)
def handler_single_voice(msg):
WechatChannel().handle_voice(msg)
return None
class WechatChannel(Channel):
def __init__(self):
pass
def startup(self):
# login by scan QRCode
itchat.auto_login(enableCmdQR=2)
# start message listener
itchat.run()
def handle_voice(self, msg):
if conf().get('speech_recognition') != True :
return
logger.debug("[WX]receive voice msg: " + msg['FileName'])
thread_pool.submit(self._do_handle_voice, msg)
def _do_handle_voice(self, msg):
from_user_id = msg['FromUserName']
other_user_id = msg['User']['UserName']
if from_user_id == other_user_id:
file_name = TmpDir().path() + msg['FileName']
msg.download(file_name)
query = super().build_voice_to_text(file_name)
if conf().get('voice_reply_voice'):
self._do_send_voice(query, from_user_id)
else:
self._do_send_text(query, from_user_id)
def handle_text(self, msg):
logger.debug("[WX]receive text msg: " + json.dumps(msg, ensure_ascii=False))
content = msg['Text']
self._handle_single_msg(msg, content)
def _handle_single_msg(self, msg, content):
from_user_id = msg['FromUserName']
to_user_id = msg['ToUserName'] # 接收人id
other_user_id = msg['User']['UserName'] # 對(duì)手方id
match_prefix = self.check_prefix(content, conf().get('single_chat_prefix'))
if "」\n- - - - - - - - - - - - - - -" in content:
logger.debug("[WX]reference query skipped")
return
if from_user_id == other_user_id and match_prefix is not None:
# 好友向自己發(fā)送消息
if match_prefix != '':
str_list = content.split(match_prefix, 1)
if len(str_list) == 2:
content = str_list[1].strip()
img_match_prefix = self.check_prefix(content, conf().get('image_create_prefix'))
if img_match_prefix:
content = content.split(img_match_prefix, 1)[1].strip()
thread_pool.submit(self._do_send_img, content, from_user_id)
else :
thread_pool.submit(self._do_send_text, content, from_user_id)
elif to_user_id == other_user_id and match_prefix:
# 自己給好友發(fā)送消息
str_list = content.split(match_prefix, 1)
if len(str_list) == 2:
content = str_list[1].strip()
img_match_prefix = self.check_prefix(content, conf().get('image_create_prefix'))
if img_match_prefix:
content = content.split(img_match_prefix, 1)[1].strip()
thread_pool.submit(self._do_send_img, content, to_user_id)
else:
thread_pool.submit(self._do_send_text, content, to_user_id)
def handle_group(self, msg):
logger.debug("[WX]receive group msg: " + json.dumps(msg, ensure_ascii=False))
group_name = msg['User'].get('NickName', None)
group_id = msg['User'].get('UserName', None)
if not group_name:
return ""
origin_content = msg['Content']
content = msg['Content']
content_list = content.split(' ', 1)
context_special_list = content.split('\u2005', 1)
if len(context_special_list) == 2:
content = context_special_list[1]
elif len(content_list) == 2:
content = content_list[1]
if "」\n- - - - - - - - - - - - - - -" in content:
logger.debug("[WX]reference query skipped")
return ""
config = conf()
match_prefix = (msg['IsAt'] and not config.get("group_at_off", False)) or self.check_prefix(origin_content, config.get('group_chat_prefix')) \
or self.check_contain(origin_content, config.get('group_chat_keyword'))
if ('ALL_GROUP' in config.get('group_name_white_list') or group_name in config.get('group_name_white_list') or self.check_contain(group_name, config.get('group_name_keyword_white_list'))) and match_prefix:
img_match_prefix = self.check_prefix(content, conf().get('image_create_prefix'))
if img_match_prefix:
content = content.split(img_match_prefix, 1)[1].strip()
thread_pool.submit(self._do_send_img, content, group_id)
else:
thread_pool.submit(self._do_send_group, content, msg)
def send(self, msg, receiver):
itchat.send(msg, toUserName=receiver)
logger.info('[WX] sendMsg={}, receiver={}'.format(msg, receiver))
def _do_send_voice(self, query, reply_user_id):
try:
if not query:
return
context = dict()
context['from_user_id'] = reply_user_id
reply_text = super().build_reply_content(query, context)
if reply_text:
replyFile = super().build_text_to_voice(reply_text)
itchat.send_file(replyFile, toUserName=reply_user_id)
logger.info('[WX] sendFile={}, receiver={}'.format(replyFile, reply_user_id))
except Exception as e:
logger.exception(e)
def _do_send_text(self, query, reply_user_id):
try:
if not query:
return
context = dict()
context['session_id'] = reply_user_id
reply_text = super().build_reply_content(query, context)
if reply_text:
self.send(conf().get("single_chat_reply_prefix") + reply_text, reply_user_id)
except Exception as e:
logger.exception(e)
def _do_send_img(self, query, reply_user_id):
try:
if not query:
return
context = dict()
context['type'] = 'IMAGE_CREATE'
img_url = super().build_reply_content(query, context)
if not img_url:
return
# 圖片下載
pic_res = requests.get(img_url, stream=True)
image_storage = io.BytesIO()
for block in pic_res.iter_content(1024):
image_storage.write(block)
image_storage.seek(0)
# 圖片發(fā)送
itchat.send_image(image_storage, reply_user_id)
logger.info('[WX] sendImage, receiver={}'.format(reply_user_id))
except Exception as e:
logger.exception(e)
def _do_send_group(self, query, msg):
if not query:
return
context = dict()
group_name = msg['User']['NickName']
group_id = msg['User']['UserName']
group_chat_in_one_session = conf().get('group_chat_in_one_session', [])
if ('ALL_GROUP' in group_chat_in_one_session or \
group_name in group_chat_in_one_session or \
self.check_contain(group_name, group_chat_in_one_session)):
context['session_id'] = group_id
else:
context['session_id'] = msg['ActualUserName']
reply_text = super().build_reply_content(query, context)
if reply_text:
reply_text = '@' + msg['ActualNickName'] + ' ' + reply_text.strip()
self.send(conf().get("group_chat_reply_prefix", "") + reply_text, group_id)
def check_prefix(self, content, prefix_list):
for prefix in prefix_list:
if content.startswith(prefix):
return prefix
return None
def check_contain(self, content, keyword_list):
if not keyword_list:
return None
for ky in keyword_list:
if content.find(ky) != -1:
return True
return None
總結(jié)：此刻我們已經(jīng)把ChatGPT和itchat成功結(jié)合，實(shí)現(xiàn)了微信機(jī)器狗自動(dòng)會(huì)話(huà)功能，同時(shí)也增加了自動(dòng)發(fā)圖片、視頻功能。但是在小編登錄過(guò)程中，發(fā)現(xiàn)只要登錄出現(xiàn)異地操作，二維碼會(huì)重復(fù)彈出的問(wèn)題，截圖如下所示：
為了解決二維碼重復(fù)彈出問(wèn)題及程序異常退出問(wèn)題，我們接下來(lái)對(duì)itchat官方源碼進(jìn)行重寫(xiě)。
四、源碼重寫(xiě)
- 將itchat源碼login模塊進(jìn)行重寫(xiě)，重寫(xiě)后代碼如下：
import os
import time
import re
import io
import threading
import json
import xml.dom.minidom
import random
import traceback
import logging
try:
from httplib import BadStatusLine
except ImportError:
from http.client import BadStatusLine
import requests
from pyqrcode import QRCode
from .. import config, utils
from ..returnvalues import ReturnValue
from ..storage.templates import wrap_user_dict
from .contact import update_local_chatrooms, update_local_friends
from .messages import produce_msg
logger = logging.getLogger('itchat')
def load_login(core):
core.login = login
core.get_QRuuid = get_QRuuid
core.get_QR = get_QR
core.check_login = check_login
core.web_init = web_init
core.show_mobile_login = show_mobile_login
core.start_receiving = start_receiving
core.get_msg = get_msg
core.logout = logout
def login(self, enableCmdQR=False, picDir=None, qrCallback=None,
loginCallback=None, exitCallback=None):
if self.alive or self.isLogging:
logger.warning('itchat has already logged in.')
return
self.isLogging = True
while self.isLogging:
uuid = push_login(self)
if uuid:
qrStorage = io.BytesIO()
else:
logger.info('Getting uuid of QR code.')
while not self.get_QRuuid():
time.sleep(1)
logger.info('Downloading QR code.')
qrStorage = self.get_QR(enableCmdQR=enableCmdQR,
picDir=picDir, qrCallback=qrCallback)
logger.info('Please scan the QR code to log in.')
isLoggedIn = False
while not isLoggedIn:
status = self.check_login()
if hasattr(qrCallback, '__call__'):
qrCallback(uuid=self.uuid, status=status,
qrcode=qrStorage.getvalue())
if status == '200':
isLoggedIn = True
elif status == '201':
if isLoggedIn is not None:
logger.info('Please press confirm on your phone.')
isLoggedIn = None
logger.info('wait 10 seconds.')
time.sleep(10)
elif status != '408':
break
if isLoggedIn:
break
elif self.isLogging:
logger.info('Log in time out, reloading QR code.')
else:
return # log in process is stopped by user
logger.info('Loading the contact, this may take a little while.')
self.web_init()
self.show_mobile_login()
self.get_contact(True)
if hasattr(loginCallback, '__call__'):
r = loginCallback()
else:
utils.clear_screen()
if os.path.exists(picDir or config.DEFAULT_QR):
os.remove(picDir or config.DEFAULT_QR)
logger.info('Login successfully as %s' % self.storageClass.nickName)
self.start_receiving(exitCallback)
self.isLogging = False
總結(jié)：在觸發(fā)狀態(tài)碼為201的時(shí)候，進(jìn)行10秒等待，否則會(huì)重復(fù)彈出二維碼，讓你無(wú)法登錄。
- 將itchat源碼core模塊進(jìn)行重寫(xiě)，重寫(xiě)后代碼如下：
class Core(object):
def __init__(self):
''' init is the only method defined in core.py
alive is value showing whether core is running
- you should call logout method to change it
- after logout, a core object can login again
storageClass only uses basic python types
- so for advanced uses, inherit it yourself
receivingRetryCount is for receiving loop retry
- it's 5 now, but actually even 1 is enough
- failing is failing
'''
self.alive, self.isLogging = False, False
self.storageClass = storage.Storage(self)
self.memberList = self.storageClass.memberList
self.mpList = self.storageClass.mpList
self.chatroomList = self.storageClass.chatroomList
self.msgList = self.storageClass.msgList
self.loginInfo = {}
self.s = requests.Session()
self.uuid = None
self.functionDict = {'FriendChat': {}, 'GroupChat': {}, 'MpChat': {}}
self.useHotReload, self.hotReloadDir = False, 'itchat.pkl'
self.receivingRetryCount = 1000
def login(self, enableCmdQR=False, picDir=None, qrCallback=None,
loginCallback=None, exitCallback=None):
''' log in like web wechat does
for log in
- a QR code will be downloaded and opened
- then scanning status is logged, it paused for you confirm
- finally it logged in and show your nickName
for options
- enableCmdQR: show qrcode in command line
- integers can be used to fit strange char length
- picDir: place for storing qrcode
- qrCallback: method that should accept uuid, status, qrcode
- loginCallback: callback after successfully logged in
- if not set, screen is cleared and qrcode is deleted
- exitCallback: callback after logged out
- it contains calling of logout
for usage
..code::python
import itchat
itchat.login()
it is defined in components/login.py
and of course every single move in login can be called outside
- you may scan source code to see how
- and modified according to your own demand
'''
raise NotImplementedError()
總結(jié)： 修改該模塊主要因?yàn)楫惓４螖?shù)過(guò)多，itchat微信就會(huì)自動(dòng)退出。重寫(xiě)后將異常次數(shù)拉滿(mǎn)，這樣就能避免因?yàn)榫W(wǎng)絡(luò)不穩(wěn)定問(wèn)題造成微信自動(dòng)退出。
- 整個(gè)項(xiàng)目啟動(dòng)代碼配置如下：
{
"zhishuyun_api_key": "",
"proxy": "127.0.0.1:7890",
"single_chat_prefix": ["機(jī)器狗", "@機(jī)器狗"],
"single_chat_reply_prefix": "[機(jī)器狗] ",
"group_chat_prefix": ["@機(jī)器狗", "@機(jī)器狗"],
"group_name_white_list": ["機(jī)器狗家族", "AI回答"],
"image_create_prefix": ["畫(huà)", "看", "找"],
"conversation_max_tokens": 1000,
"character_desc": "我是機(jī)器狗, 一個(gè)由OpenAI訓(xùn)練的大型語(yǔ)言模型, 旨在回答并解決人們的任何問(wèn)題，并且可以使用多種語(yǔ)言與人交流。",
"expires_in_seconds": 1200
}
- 然后將proxy代理部署到linux服務(wù)器，部署截圖如下：
- 最后將代碼運(yùn)行起來(lái)并進(jìn)行手機(jī)掃碼登錄，登錄后log截圖如下：
注：此刻我們已經(jīng)完全脫機(jī)，不需要手機(jī)保持在線(xiàn)；可以實(shí)現(xiàn)24小時(shí)*N天超長(zhǎng)待機(jī)。
五、效果展示
- 代碼成功部署服務(wù)器后，我們?cè)谖⑿湃毫陌l(fā)送指定內(nèi)容，查看機(jī)器狗返回內(nèi)容如下：
機(jī)器狗文本信息截圖：
文本圖片魔改版截圖：
機(jī)器狗圖片信息截圖：
機(jī)器狗視頻信息截圖：
總結(jié)：圖片、視頻功能是小編通過(guò)itchat模塊自定義擴(kuò)展的，并不是ChatGPT真實(shí)返回的，供大家觀賞使用，切勿用于不法用途。
粉絲福利：公眾號(hào)后臺(tái)回復(fù) 「機(jī)器狗」 即可獲取機(jī)器狗完整代碼
今天分享到這里就結(jié)束了，歡迎大家關(guān)注下期內(nèi)容，我們不見(jiàn)不散??????
作者簡(jiǎn)介
我是TheWeiJun，有著執(zhí)著的追求，信奉終身成長(zhǎng)，不定義自己，熱愛(ài)技術(shù)但不拘泥于技術(shù)，愛(ài)好分享，喜歡讀書(shū)和樂(lè)于結(jié)交朋友，歡迎加我微信與我交朋友。 分享日常學(xué)習(xí)中關(guān)于爬蟲(chóng)、逆向和分析的一些思路，文中若有錯(cuò)誤的地方，歡迎大家多多交流指正??
評(píng)論
圖片
表情