liuwei_x_x/abot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重大版本调整：gewechat兼容。

Browse Source
This commit is contained in:
liuwei
2025-04-22 11:17:03 +08:00
parent 41def62467
commit a62bb61901
48 changed files with 2855 additions and 1420 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
utils/wechat/contact_manager.py
View File

@@ -5,7 +5,9 @@
import logging
from typing import Dict, Optional, List, Tuple
from wcferry import Wcf
from gewechat_client import GewechatClient
from utils.json_converter import json_to_object
class ContactManager:
@@ -50,7 +52,7 @@ class ContactManager:
cls._instance = ContactManager()
return cls._instance
def set_contacts(self, contacts: Dict[str, str], head_imgs: Dict[str, str], wcf: Wcf) -> None:
def set_contacts(self, contacts: Dict[str, str]) -> None:
"""设置联系人字典
Args:
@@ -67,13 +69,12 @@ class ContactManager:
"gender": gender}
"""
self._contacts = contacts
self._friends = wcf.get_friends()
self._head_images = head_imgs
self._friends = contacts
self._logger.info(f"联系人信息已更新，共 {len(contacts)} 个联系人")
# 分类联系人
self._classify_contacts(wcf)
self._classify_contacts()
def _classify_contacts(self, wcf: Wcf) -> None:
def _classify_contacts(self) -> None:
"""将联系人分类为群组、个人联系人、公共好友和公众号"""
self._group_contacts = {}
self._personal_contacts = {}
@@ -90,8 +91,6 @@ class ContactManager:
# 判断是否为群组（wxid以@chatroom结尾）
elif wxid.endswith('@chatroom'):
self._group_contacts[wxid] = nickname
# 如果是群，这处理群列表内容
self._group_contacts_friends[wxid] = wcf.get_chatroom_members(wxid)
# # 其他为普通好友和群成员
# else:
@@ -216,7 +215,7 @@ class ContactManager:
self._personal_contacts[wxid] = nickname
self._logger.debug(f"已更新联系人: {wxid} -> {nickname}")
def refresh_contacts(self, new_contacts: Dict[str, str], head_imgs: Dict[str, str], wcf: Wcf) -> None:
def refresh_contacts(self, new_contacts: Dict[str, str]) -> None:
"""刷新联系人信息
Args:
@@ -235,11 +234,9 @@ class ContactManager:
# "province": cnt.get("province", ""),
# "city": cnt.get("city", ""),
# "gender": gender}
self._friends = wcf.get_friends()
self._head_images = head_imgs
self._friends = self.message_util.get_friends()
self._logger.info(f"联系人信息已刷新，共 {len(new_contacts)} 个联系人")
# 重新分类联系人
self._classify_contacts(wcf)
self._classify_contacts()
def get_contact_statistics(self) -> Tuple[int, int, int, int, int]:
"""获取联系人统计信息

74
utils/wechat/message_to_db.py
View File

@@ -3,12 +3,15 @@ import xml.etree.ElementTree as ET
import logging
import concurrent.futures # 添加线程池支持
import os
from wcferry import WxMsg, Wcf
from gewechat_client import GewechatClient
from db.connection import DBConnectionManager
from db.message_storage import MessageStorageDB
# 导入积分系统
from db.points_db import PointsDBOperator, PointSource
from gewechat.call_back_message.message import WxMessage
# 配置日志
logging.basicConfig(
level=logging.INFO,
@@ -19,11 +22,11 @@ logger = logging.getLogger("MessageStorage")
class MessageStorage:
def __init__(self, wcf: Wcf = None):
def __init__(self, client: GewechatClient = None):
# 获取数据库连接管理器的单例
self.db_manager = DBConnectionManager.get_instance()
self.message_db = MessageStorageDB(self.db_manager)
self.points_db = PointsDBOperator(self.db_manager)
# 初始化本地缓存字典，使用 group_id 作为键
self.local_membercounts = {}
@@ -34,7 +37,7 @@ class MessageStorage:
self.pending_tasks = []
# 保存WCF实例，用于图片处理
self.wcf = wcf
self.client = client
# 图片处理相关初始化
self.image_executor = concurrent.futures.ThreadPoolExecutor(max_workers=3) # 专用于图片处理的线程池
@@ -46,7 +49,7 @@ class MessageStorage:
os.makedirs(self.image_dir, exist_ok=True)
logger.info(f"图片存储目录: {self.image_dir}")
def process_message(self, message: WxMsg):
def process_message(self, message: WxMessage):
# 示例message字符串
current_date = datetime.now().strftime('%Y-%m-%d')
# 生成Redis key
@@ -59,7 +62,7 @@ class MessageStorage:
redis_conn.expire(key, 86400 * 2)
# 或者使用字符串：r.incr(key) # 如果只存储一个整数值，字符串类型可能更简单
def archive_message(self, msg: WxMsg):
def archive_message(self, msg: WxMessage):
"""异步存档消息，防止堵塞主线程"""
# 提交任务到线程池
future = self.executor.submit(self._archive_message_task, msg)
@@ -70,7 +73,7 @@ class MessageStorage:
# 清理已完成的任务
self._cleanup_completed_tasks()
def _archive_message_task(self, msg: WxMsg):
def _archive_message_task(self, msg: WxMessage):
"""实际执行消息存档的任务函数"""
try:
# 使用 MessageStorageDB 类存档消息
@@ -80,7 +83,7 @@ class MessageStorage:
'roomid': msg.roomid,
'sender': msg.sender,
'content': msg.content, # 添加消息内容
'message_id': msg.id # 添加消息ID
'message_id': msg.msg_id # 添加消息ID
}
except Exception as e:
logger.error(f"存档消息出错: {e}")
@@ -89,13 +92,13 @@ class MessageStorage:
'roomid': msg.roomid,
'sender': msg.sender,
'content': msg.content, # 添加消息内容
'message_id': msg.id, # 添加消息ID
'message_id': msg.msg_id, # 添加消息ID
'error': str(e)
}
def process_image(self, msg: WxMsg):
def process_image(self, msg: WxMessage):
"""异步处理图片消息，与消息存档分离"""
if msg.type != 3 or not self.wcf: # 不是图片消息或没有WCF实例
if msg.msg_type != 3 or not self.client: # 不是图片消息或没有WCF实例
return False
# 提交任务到图片处理线程池
@@ -108,11 +111,11 @@ class MessageStorage:
self._cleanup_completed_tasks()
return True
def _process_image_task(self, msg: WxMsg):
def _process_image_task(self, msg: WxMessage):
"""实际执行图片处理的任务函数"""
try:
# 使用wcf下载图片，确保图片存在
if self.wcf and msg.id:
if self.client and msg.msg_id:
# 创建按群ID或个人wxid分割的目录
target_dir = os.path.join(self.image_dir, msg.roomid if msg.roomid else msg.sender)
# 确保目标目录存在
@@ -120,16 +123,27 @@ class MessageStorage:
os.makedirs(target_dir, exist_ok=True)
# 尝试使用wcf下载图片到分组后的目录
download_path = self.wcf.download_image(msg.id, msg.extra, target_dir)
json = self.client.download_image(msg.msg_id, msg.content.xml_content, 1)
# {
# "ret": 200,
# "msg": "操作成功",
# "data": {
# "fileUrl": "/download/20240720/wx_BTVoJ_o_r6DpxNCNiycFE/0ca5b675-8e2c-4dc1-b288-3c44a40086ec4"
# }
# }
# 解析JSON
if json and json.get('data') and json['data'].get('fileUrl'):
file_url = json['data']['fileUrl']
download_path = self.download_file_from_url(file_url, target_dir)
if download_path:
logger.info(f"使用wcf下载图片成功: {msg.id} -> {download_path}")
logger.info(f"使用wcf下载图片成功: {msg.msg_id} -> {download_path}")
# 直接使用下载后的路径更新数据库
self.message_db.update_message_image_path(msg.id, download_path)
self.message_db.update_message_image_path(msg.msg_id, download_path)
return {
'success': True,
'message_id': msg.id,
'message_id': msg.msg_id,
'roomid': msg.roomid,
'sender': msg.sender,
'file_path': download_path
@@ -137,7 +151,7 @@ class MessageStorage:
else:
return {
'success': False,
'message_id': msg.id,
'message_id': msg.msg_id,
'roomid': msg.roomid,
'sender': msg.sender,
'error': "图片下载失败"
@@ -145,21 +159,25 @@ class MessageStorage:
else:
return {
'success': False,
'message_id': msg.id,
'message_id': msg.msg_id,
'roomid': msg.roomid,
'sender': msg.sender,
'error': "WCF实例不存在或消息ID无效"
}
except Exception as e:
logger.error(f"图片处理出错: {msg.id}, 错误: {e}")
logger.error(f"图片处理出错: {msg.msg_id}, 错误: {e}")
return {
'success': False,
'message_id': msg.id,
'message_id': msg.msg_id,
'roomid': msg.roomid,
'sender': msg.sender,
'error': str(e)
}
def download_file_from_url(self, url: str, target_dir: str) -> str:
# TODO 根据获取的文件地址，从server 下载 ：http://{服务ip}:2532/download/{接口返回的文件路径}
return ""
def _process_image_callback(self, future):
"""处理异步图片处理任务完成后的回调"""
try:
@@ -256,13 +274,13 @@ class MessageStorage:
# 格式化输出字符串，添加emoji和美化格式
ranking_str = f"🏆 {yesterday} 发言排行榜 🏆\n"
# 为不同名次添加不同的奖杯和样式，并发放积分
for rank, result in enumerate(results, start=1):
username = result['wx_id']
speech_count = result['speech_count']
display_name = allContacts.get(username, username)
# 根据排名发放不同数量的积分
reward_points = 0
if rank == 1:
@@ -280,14 +298,14 @@ class MessageStorage:
else:
reward_points = 5
ranking_str += f"👍 {rank}.{display_name}: {speech_count}次 +{reward_points}积分\n"
# 发放积分奖励
if reward_points > 0:
success, _ = self.points_db.add_points(
username,
groupId,
reward_points,
PointSource.OTHER,
username,
groupId,
reward_points,
PointSource.OTHER,
f"{yesterday}发言排行第{rank}名奖励"
)
if not success: