LegendWechatBot 项目进程 Week4(2025-03-24 ~ 2025-03-31)
系统功能更新
消息处理协程锁
由于在先前的版本中, 为了避免风控检测,
在处理消息前加了协程锁LegendSemaphore,
导致消息处理在有的时候会抽风堵塞, 因此在本次更新中,
移除了每条消息前的锁, 改为了在发送文字消息前的随机延迟等待
1
time.sleep(random.randint(1, 3) / random.randint(2, 10))
那协程锁白写了吗?当然不是, 改成AI消息处理限制, 防止api爆炸
消息处理逻辑优化
由于积分系统和黑白名单系统的加入,
因此在消息处理前增添了判断逻辑以及自动踢人 1
2
3
4
5
6
7
8
9
10
11
12# utils/LegendBot.py
if (
(msg.from_group() and self.DB.get_chatroom_whitelist(to) and self.DB.get_black(msg.sender) <= config.RobotConfig['black']) # 群聊且满足条件
or (not msg.from_group() and self.DB.get_black(msg.sender) <= config.RobotConfig['black']) # 私聊且满足条件
or msg.sender in config.admin # 来自管理员
):
...
# 自动踢人
elif self.DB.get_black(msg.sender) > config.RobotConfig['black'] and msg.from_group():
self.bot.sendMsg('你坏事做尽， 被移除群聊， 欢迎找kanwuqing面议解封, 有偿解封所得分发给群友作精神补偿', to, at)
self.bot.del_chatroom_members(msg.roomid, msg.sender)
数据库积分相关方法更新
添加了积分相关的处理逻辑, 实现了积分的增删改查,
以及更新用户签到状态 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68# database/LegendBotDB.py
# 查看积分
def get_points(self, wxid: str) -> int:
return self._execute_in_queue(self._get_points, wxid)
def _get_points(self, wxid: str) -> int:
session = self.DBSession()
try:
user = session.query(User).filter_by(wxid=wxid).first()
return user.points if user else 0
finally:
session.close()
# 增减积分
def add_points(self, wxid: str, num: int) -> bool:
"""Thread-safe point addition"""
return self._execute_in_queue(self._add_points, wxid, num)
def _add_points(self, wxid: str, num: int) -> bool:
"""Thread-safe point addition"""
session = self.DBSession()
try:
# Use UPDATE with atomic operation
result = session.execute(
update(User)
.where(User.wxid == wxid)
.values(points=User.points + num)
)
if result.rowcount == 0:
# User doesn't exist, create new
user = User(wxid=wxid, points=num)
session.add(user)
logger.info(f"数据库: 用户{wxid}积分增加{num}")
session.commit()
return True
except SQLAlchemyError as e:
session.rollback()
logger.error(f"数据库: 用户{wxid}积分增加失败, 错误: {e}")
return False
finally:
session.close()
# 设置积分
def set_points(self, wxid: str, num: int) -> bool:
"""Thread-safe point setting"""
return self._execute_in_queue(self._set_points, wxid, num)
def _set_points(self, wxid: str, num: int) -> bool:
"""Thread-safe point setting"""
session = self.DBSession()
try:
result = session.execute(
update(User)
.where(User.wxid == wxid)
.values(points=num)
)
if result.rowcount == 0:
user = User(wxid=wxid, points=num)
session.add(user)
logger.info(f"数据库: 用户{wxid}积分设置为{num}")
session.commit()
return True
except SQLAlchemyError as e:
session.rollback()
logger.error(f"数据库: 用户{wxid}积分设置失败, 错误: {e}")
return False
finally:
session.close()
1 | # database/LegendBotDB.py |
插件更新
api版本
VVQuest
由于VVQuest项目作者Daniel提供了api版本,
为提升插件运行速度, 因此决定保留本地及其部署方式作备用与参考,
并将插件更新为api版本
1 | # plugins/VVQuest/main.py |
肯德基疯狂星期四文案
每周都要吃肯德基的朋友有福了, 以后每周都有不同的理由让朋友v你50了
api来自pearAPI提供, 感谢
首先是肯德基文案获取 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
async def kfc(self, bot: LegendWechatBot, msg: WxMsg):
if msg.from_group():
to, at = msg.roomid, msg.sender
else:
to, at = msg.sender, None
if not self.enable:
return
if msg.content == 'kfc':
async with aiohttp.ClientSession() as session:
url = f"https://api.pearktrue.cn/api/kfc?type=json"
async with session.get(url) as resp:
if resp.status != 200:
logger.warning(f"天气查询失败: {resp.status}")
return
rsp1 = await resp.json()
bot.sendMsg(rsp1['text'].replace('\\n', '\n'), to, at)
每周四还有定时发送功能😄 1
2
3
4
5
6
7
8
9
10
11
12
13
14
async def send_kfc(self, bot: LegendWechatBot):
if not self.enable:
return
for group in LegendBotDB().get_chatroom_list():
async with aiohttp.ClientSession() as session:
url = f"https://api.pearktrue.cn/api/kfc?type=json"
async with session.get(url) as resp:
if resp.status != 200:
logger.warning(f"天气查询失败: {resp.status}")
return
rsp1 = await resp.json()
bot.sendMsg(rsp1['text'], group)
天气预报
api来自free-api提供, 感谢
天气预报共实现了如下功能 - 查询天气 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26# plugins/Weather/main.py
if msg.content.count(' ') == 1:
city, day = msg.content.split(' ')
else:
city, day = msg.content, 0
if day >= 3:
return
async with aiohttp.ClientSession() as session:
url = f"https://api.seniverse.com/v3/weather/daily.json?key={self.key}&location={city}&language=zh-Hans&unit=c"
async with session.get(url) as resp:
if resp.status != 200:
logger.warning(f"天气查询失败: {resp.status}")
return
rsp1 = await resp.json()
if 'status_code' in rsp1 and rsp1['status_code'] == "AP010006":
bot.sendMsg("城市名错误, 请重新输入", to, at)
return
rsp = rsp1["results"][0]["daily"][day]
upd = rsp1['results'][0]['last_update']
res = f"{city}{rsp['date']}天气, 更新于{upd}\n白天天气:{rsp['text_day']}, 夜间天气:{rsp['text_night']}\n最高温: {rsp['high']}, 最低温: {rsp['low']}\n降水概率: {rsp['precip']}%, 湿度: {rsp['humidity']}\n风力风向: {rsp['wind_direction']}风{rsp['rainfall']}级, 风速: {rsp['wind_speed']}"
bot.sendMsg(res, to, at)
LegendBotDB().add_points(msg.sender, -1)
预订天气预报
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31# plugins/Weather/main.py
if msg.content.startswith("预报 ") and msg.from_group():
city = msg.content[3:]
async with aiohttp.ClientSession() as session:
url = f"https://api.seniverse.com/v3/weather/daily.json?key={self.key}&location={city}&language=zh-Hans&unit=c"
async with session.get(url) as resp:
if resp.status != 200:
logger.warning(f"天气查询失败: {resp.status}")
return
rsp1 = await resp.json()
if 'status_code' in rsp1 and rsp1['status_code'] == "AP010006":
bot.sendMsg("城市名错误, 请重新输入", to, at)
return
if to in self.subs:
if city in self.subs[to]:
bot.sendMsg("该城市已订阅, 请勿重复订阅", to, at)
return
if len(self.subs[to]) > 5:
bot.sendMsg("该群聊已订阅5个城市, 请先取消订阅", to, at)
return
self.subs[to].append(city)
with open('plugins/Weather/subs.json', 'w', encoding='utf-8') as f:
json.dump(self.subs, f, ensure_ascii=False, indent=4)
bot.sendMsg("订阅成功", to, at)
return
self.subs[to] = [city]取消预订天气预报
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16# plugins/Weather/main.py
elif msg.content.startswith("td "):
city = msg.content[3:]
if to in self.subs:
if city in self.subs[to]:
self.subs[to].remove(city)
with open('plugins/Weather/subs.json', 'w', encoding='utf-8') as f:
json.dump(self.subs, f, ensure_ascii=False, indent=4)
bot.sendMsg("取消订阅成功", to, at)
return
else:
bot.sendMsg("该城市未订阅, 请先订阅", to, at)
return
else:
bot.sendMsg("该群聊未订阅任何城市, 请先订阅", to, at)
return
AI对话 (初版)
有了AI的接入, 机器人能真正意义上回答所有问题了
1 | """这句很重要""" |
签到查看运势
机器人的正常运营依赖于插件的正常使用, 插件的正常使用离不开积分的管控和激励, 每天的签到获取积分的同时, 还加入了运势来增加趣味性
运势获取功能
1 | # plugins/SignUp/main.py |
展开
在这里插一首歌吧
-
在两年前我就说过, 要让好多好多人看到我们一起构思的运势, 现在这个目标离我越来越近, 但你却离我愈发遥远了...
签到功能
1 | # plugins/SignUp/main.py |
todo list
- 添加AI群聊自定义角色功能
- 添加AI作图
- AI聊天上下文功能
- 完善文档
项目已开源至 Github ，欢迎star和fork 若你觉得对你的开发有帮助, 或是对你的生活提供了方便, 欢迎来 爱发电 赞助
如果想一起开发或贡献插件等, 欢迎在相关标准制定后按照标准提交PR, 或 联系作者
如果想一起开发或贡献插件等, 欢迎在相关标准制定后按照标准提交PR, 或 
如果想一起开发或贡献插件等, 欢迎在相关标准制定后按照标准提交PR, 或 
如果想一起开发或贡献插件等, 欢迎在相关标准制定后按照标准提交PR, 或