Linux服务器微信机器人
鸣谢
xiaozhao为本项目的测试工作给予了很大帮助，感谢
这里是他的博客地址：https://aaaaaaamua.github.io
简述
因为实验室群需要统计最后发言时间，于是写了一个微信机器人，代码和注意事项已经存放在github上，这里仅作个人记录,并备份主程序和部分程序代码
项目地址：https://github.com/hustler0000/ALinuxSimpleWxbot/
帮助/功能文档
机器人会默默记下群成员的最后发言时间，@机器人发消息可以触发指令，机器人每天会在群内推送新闻消息，每个小时都会自动微信发送存活信息到我微信
改昵称必看（重要）
roomname 群昵称 新老群成员可以使用这个修改在数据库里的群昵称，方便机器人称呼和统计最后发言时间，此功能只有在自己修改了群昵称后需要操作一次
username 群昵称 此指令可以修改在数据库里的微信昵称，如果机器人没有提示请不要操作
以上两条指令，如无必要请不要操作，并且不要同时修改群昵称和微信昵称，不然机器人可能会坏掉
常规指令
help 显示本帮助文档
last 显示本人最后发言时间
search 某人 输出某人的最后发言时间
all 输出一个文件，里面是所有人的最后发言时间
check 检查机器人存活状态
feedback 反馈内容 发送反馈
注意事项
当有群成员使用发送反馈时，机器人会向你的主微信账号发送反馈内容，当有新群员加入时，机器人也会提醒你引导成员将自己的群昵称添加进数据库内，当然你也可以手动操作数据库来添加。
为了方便地手动操作数据库，还有一个SqliteOperate.py文件，这个python程序提供了简单管理sqlite数据库的条件，运行该程序，并输入相应数据库语句来对你的数据库进行操作。
用post请求访问服务器的3001端口下的/healthz?token=你的token 路径可以检查机器人docker程序是否掉线
用get请求访问服务器8080端口下的/check 路径可以检查机器人python程序是否掉线
原理
十分简单，因为是二次开发，就是docker收发微信信息，收到的信息通过api转发到python跑的api服务中并且处理
存在问题
1.程序可能存在sql注入漏洞，可能会导致机器人功能异常
2.目前机器人只能部署到一个群内，部署到多个群需要开启多个docker服务和python脚本，十分麻烦且浪费资源
3.微信定时两天掉线，详细解决方案以及讨论请查看https://github.com/danni-cool/wechatbot-webhook 内的issue。
4.只能用手机先登录机器人的微信账号，然后再扫码登录服务器docker中的微信，而且手机上的机器人微信还不能退出，否则机器人会掉线，即你需要有一台设备一直登录着机器人微信，并且这台设备能随时扫码登录服务器上的微信
源代码
最初版本的主程序代码
from fastapi import FastAPI, Form
import uvicorn
import json
import datetime
import sqlite3 as sl
import os
flag=1
app = FastAPI()
@app.post("/receive_msg")
async def print_json(source: str=Form(), content: str = Form(), isMentioned: str=Form()):
a={"success": True}
source=json.loads(source)
s=source.get("from")
ss=s.get("payload")
name=ss.get("name")
time=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
cont = content.split("\u2005")
if(len(cont)>1):
con = cont[1]
concon = con.split(" ")
con2 = concon[0]
sql1="select * from POST where username='%s'" % (name)
conn = sl.connect('menbers.db')
cursor = conn.cursor()
cursor.execute(sql1)
data=cursor.fetchall()
conn.commit()
conn.close()
if(data==""):
flag=0
sql1="replace into POST(username,last_post_time) values('%s','%s')" % (name, time)
cursor.execute(sql1)
conn.commit()
a = {"success": True, "data": {"type": "text", "content": "我还不认识你，可能因为你是新成员，或者改了微信昵称，不过我已经把你加到我的小本本里了，如果你是新人，请使用：\nroomname 群昵称\n这条指令，在我的小本本上记下你的群昵称，欢迎你加入到HackTB实验室这个大家庭！\n\n如果你是修改了微信昵称，请使用：\nusername 你的群昵称\n在我的小本本上记下你的新微信昵称\n请不要同时修改微信昵称和群昵称，不然我真的记不过来呀[流泪]望见谅\n\n@我并输入help指令可以呼出我的帮助菜单哦！"}}
cmd = 'curl --location "http://localhost:3001/webhook/msg" --header "Content-Type: application/json" --data \'{"to": "Doom.","type": "text","content":"实验室有新成员或者有人改了微信昵称，新人或者昵称为' + name + '，请迅速检查处理!"}\''
os.system(cmd)
conn.close()
else:
flag=1
if(isMentioned!="1"):
sql="update POST set last_post_time='%s' where username='%s'" % (time,name)
conn = sl.connect('menbers.db')
cursor = conn.cursor()
cursor.execute(sql)
conn.commit()
conn.close()
if(isMentioned=="1" and con2=="roomname"):
sql="update POST set roomname='%s' where username='%s'" % (concon[1],name)
conn = sl.connect('menbers.db')
cursor = conn.cursor()
cursor.execute(sql)
conn.commit()
conn.close()
if(isMentioned=="1" and con2=="username"):
sql="update POST set username='%s' where roomname=%'s'" % (name,concon[1])
conn = sl.connect('menbers.db')
cursor = conn.cursor()
cursor.execute(sql)
conn.commit()
sql = "delete from POST where username='%s' and roomname is null"
cursor.execute(sql)
conn.commit()
conn.close()
if(isMentioned=="1" and flag==1):
a={"success": True,"data": {"type": "text","content":"@"+name+"\u2005真闲！不过如果你真的无聊的话，可以试试sql注入我，帮助主人找漏洞哦！先在此谢过了！"}}
if(con=="help"):
a={"success": True,"data": {"type": "text","content": "我是Scr1ptKidB0t，我会默默记下大家的最后发言时间\n@我发消息可以触发指令，大家@我的时候要我回复了才能继续@哦：\n\n改昵称必看（重要）：\nroomname 群昵称 新老群成员可以使用这个修改在小本本里的群昵称，方便我称呼和统计最后发言时间，此功能只有在自己修改了群昵称后需要操作一次哦\nusername 群昵称 此指令可以修改在小本本里的微信昵称，如果机器人没有提示请不要操作\n以上两条指令，如无必要请不要操作，并且不要同时修改群昵称和微信昵称，不然机器人可能会坏掉，谢谢\n\nhelp 显示本帮助文档\nlast 显示本人最后发言时间\nsearch 某人 输出某人的最后发言时间\nall 输出一个文件，里面是所有人的最后发言时间\nfeedback 反馈内容 发送反馈\n\n项目地址：https://github.com/danni-cool/wechatbot-webhook\n使用指南：https://www.bilibili.com/read/cv28706223/\n\n希望大家珍惜我，不要@我刷屏，不要连续@我，玩坏了掉线了及时告知我的主人，希望能和大家一起进步呀"}}
if(con=="last"):
sql="select roomname,last_post_time from POST where username='%s'" % (name)
conn = sl.connect('menbers.db')
cursor = conn.cursor()
cursor.execute(sql)
data = cursor.fetchall()
conn.commit()
conn.close()
data=str(data)
d=data.split(",")
d0=d[0]
d00=d0[3:-1]
d1=d[1]
d11=d1[2:-3]
a={"success": True,"data": {"type": "text","content": "你好"+d00+"\n你最后的发言时间是："+d11+"\n记得要多发言，营造活跃的群内气氛哦！"}}
if(con2=="search"):
menb=concon[1]
sql="select roomname,last_post_time from POST where username='%s' or roomname='%s'" % (menb,menb)
conn = sl.connect('menbers.db')
cursor = conn.cursor()
cursor.execute(sql)
data = cursor.fetchall()
conn.commit()
conn.close()
data=str(data)
d=data.split(",")
d0=d[0]
d00=d0[3:-1]
d1=d[1]
d11=d1[2:-3]
a={"success": True,"data": {"type": "text","content": "你好"+name+"\n"+d00+"最后的发言时间是："+d11+"\n记得要多发言，营造活跃的群内气氛哦！"}}
if(con=="all"):
sql="select id,roomname,last_post_time from POST"
conn = sl.connect('menbers.db')
cursor = conn.cursor()
cursor.execute(sql)
data = cursor.fetchall()
conn.commit()
conn.close()
for item in data:
item=str(item)
ii=item.split(",")
id=ii[0]
rn=ii[1]
roomname=rn[2:-1]
t=ii[2]
time=t[2:-2]
if(id[1:]=="1"):
with open("all_menbers.txt","w") as f:
f.write(roomname+" "+time+"\n")
f.close()
else:
with open("all_menbers.txt","a") as f:
f.write(roomname+" "+time+"\n")
f.close()
os.system("curl --location --request POST 'http://localhost:3001/webhook/msg' --form 'to=HackTB实验室' --form content=@'/root/ScriptKidB0T/all_menbers.txt' --form 'isRoom=1'")
a={"success": True,"data": {"type": "text","content":"以上是大家的发言时间记录，请大家踊跃发言，一起成长呀"}}
if(con2=="feedback"):
sql="select roomname from POST where username='%s'" % (name)
conn = sl.connect('menbers.db')
cursor = conn.cursor()
cursor.execute(sql)
data = cursor.fetchall()
conn.commit()
conn.close()
data=str(data)
d=data.split(",")
d0=d[0]
d00=d0[3:-1]
a={"success": True,"data": {"type": "text","content":"成员"+d00+"，你的反馈我已经告诉主人了，感谢你的支持呀！"}}
fb=concon[1]
cmd='curl --location "http://localhost:3001/webhook/msg" --header "Content-Type: application/json" --data \'{"to": "Doom.","type": "text","content":"实验室的'+d00+'发送了反馈，内容为'+fb+'，请迅速处理!"}\''
os.system(cmd)
return a
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8080)最初版本的根本不敢直接放到github上好吗
修改后版本
from apscheduler.schedulers.background import BackgroundScheduler
from fastapi import FastAPI, Form
import uvicorn
import json
import datetime
import sqlite3 as sl
import os
import requests
from bs4 import BeautifulSoup
import feedparser
app = FastAPI()
scheduler = BackgroundScheduler()
def getrss(strings,url):
d=feedparser.parse(url)
strings=strings+d.entries[0].title+"\n"
strings=strings+"链接："+d.entries[0].link+"\n"
strings = strings +"来源："+ d.feed.title + "\n"
return strings
def roomdailynews():
newurl="https://www.ddosi.org/category/%E6%B8%97%E9%80%8F%E6%B5%8B%E8%AF%95/"
header = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.132 Safari/537.36'
}
response = requests.get(url=newurl, headers=header)
response.raise_for_status()
response = response.content.decode('utf-8')
response = BeautifulSoup(response, 'lxml')
response = response.find_all(attrs={"class": "entry-title"})
newlink = response[0].a.attrs['href']
newtitle= response[0].text
todaytime = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S %A')
news=""
count=1
rssfile=open("rss.txt")
rsslist=rssfile.readlines()
for line in rsslist:
if(count==1):
news=news+"早上好！现在是"+todaytime+"\n\n每日快讯:\n"
if(count==3):
news=news+"\n威胁情报：\n"
if(count==7):
news=news+"\n漏洞报告：\n"
if(count==9):
news=news+"\n最新CVE报告：\nhttps://cassandra.cerias.purdue.edu/CVE_changes/today.html\n部分链接需梯子"+"\n\n安全文章：\n"+newtitle+"\n"+newlink+"\n来源：https://www.ddosi.org/category/%E6%B8%97%E9%80%8F%E6%B5%8B%E8%AF%95/\n"
news=getrss(news,line)
count=count+1
news=news+"以上就是今早的新内容，一起来学习呀！"
url1 = 'http://127.0.0.1:3001/webhook/msg'
headers1 = {"content-type":"application/json"}
payload1 = {"to":"HackTB实验室","isRoom": True ,"type": "text","content":news}
requests.post(url1,json=payload1,headers=headers1)
def wxautocheck():
reply = 'curl --location "http://localhost:3001/webhook/msg" --header "Content-Type: application/json" --data \'{"to": "Doom.","type": "text","content":"I am still alive!"}\''
os.system(reply)
@app.on_event("startup")
async def app_start():
scheduler.add_job(roomdailynews, 'interval', days=1)
scheduler.add_job(wxautocheck, 'interval', hours=1)
scheduler.start()
@app.get("/check")
async def check():
a="success!"
return a
@app.post("/receive_msg")
async def print_json(source: str=Form(), content: str = Form(), isMentioned: str=Form()):
a={"success": True}
source=json.loads(source)
source=source.get("from")
source=source.get("payload")
name=source.get("name")
time=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
args = content.split("\u2005")
if(len(args)>1):
args = args[1]
args = args.split(" ",maxsplit=1)
sql="select * from POST where username='%s'" % (name)
conn = sl.connect('menbers.db')
cursor = conn.cursor()
cursor.execute(sql)
data=cursor.fetchall()
if(data==""):
sql="replace into POST(username,last_post_time) values('%s','%s')" % (name, time)
cursor.execute(sql)
a = {"success": True, "data": {"type": "text", "content": "我还不认识你，可能因为你是新成员，或者改了微信昵称，不过我已经把你加到我的小本本里了，如果你是新人，请使用：\nroomname 群昵称\n这条指令，在我的小本本上记下你的群昵称，欢迎你加入到HackTB实验室这个大家庭！\n\n如果你是修改了微信昵称，请使用：\nusername 你的群昵称\n在我的小本本上记下你的新微信昵称\n请不要同时修改微信昵称和群昵称，不然我真的记不过来呀[流泪]望见谅\n\n@我并输入help可以呼出我的帮助菜单哦！"}}
cmd = 'curl --location "http://localhost:3001/webhook/msg" --header "Content-Type: application/json" --data \'{"to": "Doom.","type": "text","content":"实验室有新成员或者有人改了微信昵称，新人或者昵称为' + name + '，请迅速检查处理!"}\''
os.system(cmd)
elif(isMentioned!="1"):
sql="update POST set last_post_time='%s' where username='%s'" % (time,name)
cursor.execute(sql)
else:
a={"success": True,"data": {"type": "text","content":"@"+name+"\u2005真闲！不过如果你真的无聊的话，可以试试sql注入我，帮助主人找漏洞哦！先在此谢过了！"}}
if(args[0]=="roomname"):
sql="update POST set roomname='%s' where username='%s'" % (args[1],name)
cursor.execute(sql)
sql="select roomname from POST where username='%s'" % (name)
cursor.execute(sql)
data = cursor.fetchall()
data = str(data)
data = data.split(",")
dname = data[0]
dname = dname[3:-1]
a = {"success": True, "data": {"type": "text", "content": "成功修改小本本上的群昵称为"+dname}}
if(args[0]=="username"):
sql="update POST set username='%s' where roomname='%s'" % (name,args[1])
cursor.execute(sql)
sql = "delete from POST where username='%s' and roomname is null" % (name)
cursor.execute(sql)
sql="select username from POST where roomname='%s'" % (args[1])
cursor.execute(sql)
data = cursor.fetchall()
data = str(data)
data = data.split(",")
dname = data[0]
dname = dname[3:-1]
a = {"success": True, "data": {"type": "text", "content": "成功修改小本本上的微信名称为" + dname}}
if(args[0]=="check"):
a = {"success": True, "data": {"type": "text", "content":"I am still alive!"}}
if(args[0]=="help"):
a={"success": True,"data": {"type": "text","content": "我是Scr1ptKidB0t，我会默默记下大家的最后发言时间\n@我发消息可以触发指令，大家@我的时候要我回复了才能继续@哦：\n\n改昵称必看（重要）：\nroomname 群昵称 新老群成员可以使用这个修改在小本本里的群昵称，方便我称呼和统计最后发言时间，此功能只有在自己修改了群昵称后需要操作一次哦\nusername 群昵称 此指令可以修改在小本本里的微信昵称，如果机器人没有提示请不要操作\n以上两条指令，如无必要请不要操作，并且不要同时修改群昵称和微信昵称，不然机器人可能会坏掉，谢谢\n\nhelp 显示本帮助文档\nlast 显示本人最后发言时间\nsearch 某人 输出某人的最后发言时间\nall 输出一个文件，里面是所有人的最后发言时间\ncheck 检查机器人存活状态\nfeedback 反馈内容 发送反馈\n\n项目地址：https://github.com/hustler0000/SimpleWxbot\n\n希望大家珍惜我，不要@我刷屏，不要连续@我，玩坏了掉线了及时告知我的主人，希望能和大家一起进步呀"}}
if(args[0]=="last"):
sql="select roomname,last_post_time from POST where username='%s'" % (name)
cursor.execute(sql)
data = cursor.fetchall()
data=str(data)
data=data.split(",")
dname=data[0]
dname=dname[3:-1]
dtime=data[1]
dtime=dtime[2:-3]
a={"success": True,"data": {"type": "text","content": "你好"+dname+"\n你最后的发言时间是："+dtime+"\n记得要多发言，营造活跃的群内气氛哦！"}}
if(args[0]=="search"):
menb=args[1]
sql="select roomname,last_post_time from POST where username='%s' or roomname='%s'" % (menb,menb)
cursor.execute(sql)
data = cursor.fetchall()
data=str(data)
data=data.split(",")
dname=data[0]
dname=dname[3:-1]
dtime=data[1]
dtime=dtime[2:-3]
a={"success": True,"data": {"type": "text","content": "你好"+name+"\n"+dname+"最后的发言时间是："+dtime+"\n记得要多发言，营造活跃的群内气氛哦！"}}
if(args[0]=="all"):
os.system("rm -rf all_menbers.txt")
sql="select id,roomname,last_post_time from POST"
cursor.execute(sql)
data = cursor.fetchall()
with open("all_menbers.txt", "a") as f:
for item in data:
item=str(item)
item=item.split(",")
roomname=item[1]
roomname=roomname[2:-1]
time=item[2]
time=time[2:-2]
f.write(roomname+" "+time+"\n")
f.close()
os.system("curl --location --request POST 'http://localhost:3001/webhook/msg' --form 'to=HackTB实验室' --form content=@'/root/ScriptKidB0T/all_menbers.txt' --form 'isRoom=1'")
a={"success": True,"data": {"type": "text","content":"以上是大家的发言时间记录，请大家踊跃发言，一起成长呀"}}
if(args[0]=="feedback"):
sql="select roomname from POST where username='%s'" % (name)
cursor.execute(sql)
data = cursor.fetchall()
data=str(data)
data=data.split(",")
dname=data[0]
dname=dname[3:-1]
a={"success": True,"data": {"type": "text","content":"成员"+dname+"，你的反馈我已经告诉主人了，感谢你的支持呀！"}}
fb=args[1]
cmd='curl --location "http://localhost:3001/webhook/msg" --header "Content-Type: application/json" --data \'{"to": "Doom.","type": "text","content":"实验室的'+dname+'发送了反馈，内容为'+fb+'，请迅速处理!"}\''
os.system(cmd)
conn.commit()
conn.close()
return a
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8080)
github版本
sqlite3数据库操作程序
import sqlite3 as sl
while(1):
sql = input(">>>")
con = sl.connect('menbers.db')
cursor = con.cursor()
cursor.execute(sql)
data = cursor.fetchall()
print(type(data))
print(data)
con.commit()
con.close()创建初始数据库程序
from pywinauto.application import Application
import pywinauto
import time
import psutil
import pandas as pd
import numpy as np
import sqlite3 as sl
import csv
def get_wechat_pid():
pids = psutil.pids()
for pid in pids:
p = psutil.Process(pid)
if p.name() == 'WeChat.exe':
return pid
return None
def get_name_list(pid):
print('>>> WeChat.exe pid: {}'.format(pid))
print('>>> 请打开【微信=>目标群聊=>聊天成员=>查看更多】，尤其是【查看更多】，否则查找不全！')
for i in range(20):
print('\r({:2d} 秒)'.format(20 - i), end='')
time.sleep(1)
app = Application(backend='uia').connect(process=pid)
win_main_Dialog = app.window(class_name='WeChatMainWndForPC')
chat_list = win_main_Dialog.child_window(control_type='List', title='聊天成员')
name_list = []
all_members = []
for i in chat_list.items():
p = i.descendants()
if p and len(p) > 5:
if p[5].texts() and p[5].texts()[0].strip() != '' and (p[5].texts()[0].strip() != '添加' and p[5].texts()[0].strip() != '移出'):
name_list.append(p[5].texts()[0].strip())
all_members.append([p[5].texts()[0].strip(), p[3].texts()[0].strip()])
pd.DataFrame(np.array(all_members)).to_csv('all_members.csv', header=['群昵称', '微信昵称'])
print('\r>>> 群成员共 {} 人，结果已保存至all_members.csv'.format(len(name_list)))
return name_list
def match():
pid = get_wechat_pid()
if pid == None:
print('>>> 找不到WeChat.exe，请先打开WeChat.exe再运行此脚本！')
return
try:
member_list = get_name_list(pid)
except pywinauto.findwindows.ElementNotFoundError as e:
print('\r>>> 未找到【聊天成员】窗口，程序终止！')
print('>>> 若已开启【聊天成员】窗口但仍报错，请重启微信（原因：可能存在多个WeChat进程）')
return
if __name__ == '__main__':
match()
sql = 'create table if not exists POST(id integer primary key autoincrement,username text,roomname text ,last_post_time text)'
con = sl.connect('menbers.db')
cursor = con.cursor()
cursor.execute(sql)
con.commit()
con.close()
print("init ok")
with open("all_members.csv", mode="r", encoding="utf-8-sig") as mb:
reader = csv.reader(mb)
for row in reader:
print(row[2] + row[1])
if (row[1] != "群昵称"):
sql = "replace into POST(username,roomname) values('%s','%s')" % (row[2], row[1])
con = sl.connect('menbers.db')
cursor = con.cursor()
cursor.execute(sql)
con.commit()
con.close()
print(row[2] + row[1] + "ok")
mb.close()RSS链接
https://www.freebuf.com/feed
https://www.sec-wiki.com/news/rss
https://wechat2rss.xlab.app/feed/de09ec267e5c4545e0a759cc62c3da7866ea49e0.xml
https://wechat2rss.xlab.app/feed/b93962f981247c0091dad08df5b7a6864ab888e9.xml
https://wechat2rss.xlab.app/feed/7874947663d806190d77bdca6f8f6855f65a1b20.xml
https://wechat2rss.xlab.app/feed/034265b14906a59ef7cf1fcbd56699b54a696094.xml
https://www.seebug.org/rss/new/
https://wechat2rss.xlab.app/feed/981c000a01bbdc1f128d260cc91c15d3a6afb530.xml
https://wechat2rss.xlab.app/feed/90c827b8290310a96ef80a13df9dbcc06ab69892.xml
https://wechat2rss.xlab.app/feed/6ce082e908ac0894ff00b2d9d8e186181cd810bd.xml
https://paper.seebug.org/rss
https://xz.aliyun.com/feed
https://wechat2rss.xlab.app/feed/837190f74457627e0a5567700c573fe8afd7d3fe.xml
https://www.secpulse.com/feed
未来走向
1.考虑多群部署，添加私聊检测机器人是否活着的功能
2.考虑添加GPT或者谷歌AI，制作一个AI助理