提交代码调整。
This commit is contained in:
@@ -13,6 +13,7 @@ from db.connection import DBConnectionManager
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ContactsDBOperator(BaseDBOperator):
|
||||
"""微信联系人数据库操作类"""
|
||||
|
||||
@@ -20,9 +21,7 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
"""初始化联系人数据库操作类"""
|
||||
super().__init__(db_manager or DBConnectionManager.get_instance())
|
||||
self.logger = logging.getLogger("ContactsDBOperator")
|
||||
|
||||
# 确保数据库表存在
|
||||
# self._ensure_table_exists()
|
||||
|
||||
|
||||
def _ensure_table_exists(self):
|
||||
"""确保联系人表存在"""
|
||||
@@ -57,7 +56,7 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
UNIQUE KEY `idx_user_name` (`user_name`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='微信联系人信息表';
|
||||
""")
|
||||
|
||||
|
||||
# 创建群成员表 - 增加了更多字段以支持详细信息
|
||||
self.execute_update("""
|
||||
CREATE TABLE IF NOT EXISTS t_chatroom_member (
|
||||
@@ -89,12 +88,12 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
UNIQUE KEY `idx_chatroom_member` (`chatroom_id`, `wxid`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='微信群成员信息表';
|
||||
""")
|
||||
|
||||
|
||||
self.logger.info("成功创建或确认微信联系人表和群成员表存在")
|
||||
except Exception as e:
|
||||
self.logger.error(f"创建微信联系人表或群成员表失败: {e}")
|
||||
raise
|
||||
|
||||
|
||||
def save_contacts(self, contacts_data: List[Dict], contact_type: str) -> bool:
|
||||
"""保存联系人信息到数据库
|
||||
|
||||
@@ -108,7 +107,7 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
if not contacts_data:
|
||||
self.logger.warning(f"没有{contact_type}类型的联系人数据需要保存")
|
||||
return True
|
||||
|
||||
|
||||
try:
|
||||
for contact in contacts_data:
|
||||
# 将驼峰命名转换为下划线命名
|
||||
@@ -132,33 +131,34 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
'description': contact.get('description', ''),
|
||||
'card_img_url': contact.get('cardImgUrl', ''),
|
||||
'label_list': contact.get('labelList', ''),
|
||||
'phone_num_list': json.dumps(contact.get('phoneNumList', [])) if contact.get('phoneNumList') else '',
|
||||
'phone_num_list': json.dumps(contact.get('phoneNumList', [])) if contact.get(
|
||||
'phoneNumList') else '',
|
||||
'type': contact_type
|
||||
}
|
||||
|
||||
|
||||
# 构建SQL语句
|
||||
fields = ', '.join(data.keys())
|
||||
placeholders = ', '.join(['%s'] * len(data))
|
||||
values = tuple(data.values())
|
||||
|
||||
|
||||
# 使用INSERT ... ON DUPLICATE KEY UPDATE语法
|
||||
update_clause = ', '.join([f"{k}=VALUES({k})" for k in data.keys() if k != 'user_name'])
|
||||
|
||||
|
||||
sql = f"""
|
||||
INSERT INTO t_wechat_contacts ({fields})
|
||||
VALUES ({placeholders})
|
||||
ON DUPLICATE KEY UPDATE {update_clause}
|
||||
"""
|
||||
|
||||
|
||||
self.execute_update(sql, values)
|
||||
|
||||
|
||||
self.logger.info(f"成功保存{len(contacts_data)}个{contact_type}类型的联系人")
|
||||
return True
|
||||
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"保存{contact_type}类型的联系人失败: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def save_simple_contacts(self, contact_list: List[str], contact_type: str) -> bool:
|
||||
"""保存简单联系人列表(只有user_name)到数据库
|
||||
|
||||
@@ -172,7 +172,7 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
if not contact_list:
|
||||
self.logger.warning(f"没有{contact_type}类型的联系人数据需要保存")
|
||||
return True
|
||||
|
||||
|
||||
try:
|
||||
for user_name in contact_list:
|
||||
# 构建SQL语句
|
||||
@@ -181,16 +181,16 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
VALUES (%s, %s)
|
||||
ON DUPLICATE KEY UPDATE type = VALUES(type), update_time = CURRENT_TIMESTAMP
|
||||
"""
|
||||
|
||||
|
||||
self.execute_update(sql, (user_name, contact_type))
|
||||
|
||||
|
||||
self.logger.info(f"成功保存{len(contact_list)}个{contact_type}类型的简单联系人")
|
||||
return True
|
||||
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"保存{contact_type}类型的简单联系人失败: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def get_contacts_by_type(self, contact_type: str) -> List[Dict]:
|
||||
"""根据类型获取联系人列表
|
||||
|
||||
@@ -206,13 +206,13 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
WHERE type = %s
|
||||
ORDER BY nick_name
|
||||
"""
|
||||
|
||||
|
||||
results = self.execute_query(sql, (contact_type,))
|
||||
return results
|
||||
except Exception as e:
|
||||
self.logger.error(f"获取{contact_type}类型的联系人失败: {e}")
|
||||
return []
|
||||
|
||||
|
||||
def get_contact_by_user_name(self, user_name: str) -> Optional[Dict]:
|
||||
"""根据user_name获取联系人信息
|
||||
|
||||
@@ -228,13 +228,13 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
WHERE user_name = %s
|
||||
LIMIT 1
|
||||
"""
|
||||
|
||||
|
||||
result = self.execute_query(sql, (user_name,), fetch_one=True)
|
||||
return result
|
||||
except Exception as e:
|
||||
self.logger.error(f"获取联系人{user_name}失败: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def get_display_name(self, user_name: str) -> str:
|
||||
"""获取联系人的显示名称(优先使用备注,其次是昵称,最后是微信ID)
|
||||
|
||||
@@ -247,9 +247,9 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
contact = self.get_contact_by_user_name(user_name)
|
||||
if not contact:
|
||||
return user_name
|
||||
|
||||
|
||||
return contact.get('remark') or contact.get('nick_name') or user_name
|
||||
|
||||
|
||||
def get_all_contacts_name_map(self) -> Dict[str, str]:
|
||||
"""获取所有联系人的ID到显示名称的映射
|
||||
|
||||
@@ -260,9 +260,9 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
sql = """
|
||||
SELECT user_name, remark, nick_name FROM t_wechat_contacts
|
||||
"""
|
||||
|
||||
|
||||
results = self.execute_query(sql)
|
||||
|
||||
|
||||
name_map = {}
|
||||
for result in results:
|
||||
user_name = result.get('user_name')
|
||||
@@ -270,7 +270,7 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
nick_name = result.get('nick_name')
|
||||
display_name = remark or nick_name or user_name
|
||||
name_map[user_name] = display_name
|
||||
|
||||
|
||||
return name_map
|
||||
except Exception as e:
|
||||
self.logger.error(f"获取所有联系人名称映射失败: {e}")
|
||||
@@ -289,7 +289,7 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
if not member_details or not chatroom_id:
|
||||
self.logger.warning(f"没有群聊{chatroom_id}的成员详细信息需要保存")
|
||||
return False
|
||||
|
||||
|
||||
try:
|
||||
# 获取现有的群成员信息,以便更新而不是替换
|
||||
existing_members_sql = """
|
||||
@@ -297,26 +297,26 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
WHERE chatroom_id = %s
|
||||
"""
|
||||
existing_members_result = self.execute_query(existing_members_sql, (chatroom_id,))
|
||||
existing_members = {row.get('wxid'): (row.get('is_owner'), row.get('is_admin'))
|
||||
for row in existing_members_result}
|
||||
|
||||
existing_members = {row.get('wxid'): (row.get('is_owner'), row.get('is_admin'))
|
||||
for row in existing_members_result}
|
||||
|
||||
for member in member_details:
|
||||
wxid = member.get('userName', '')
|
||||
if not wxid:
|
||||
continue
|
||||
|
||||
|
||||
# 保留现有的群主和管理员标识
|
||||
is_owner, is_admin = 0, 0
|
||||
if wxid in existing_members:
|
||||
is_owner, is_admin = existing_members[wxid]
|
||||
|
||||
|
||||
# 处理电话号码列表
|
||||
phone_num_list = member.get('phoneNumList', [])
|
||||
if phone_num_list:
|
||||
phone_num_str = json.dumps(phone_num_list)
|
||||
else:
|
||||
phone_num_str = ''
|
||||
|
||||
|
||||
# 构建数据
|
||||
data = {
|
||||
'chatroom_id': chatroom_id,
|
||||
@@ -343,26 +343,26 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
'remark_py_initial': member.get('remarkPyInitial', ''),
|
||||
'remark_quan_pin': member.get('remarkQuanPin', '')
|
||||
}
|
||||
|
||||
|
||||
# 构建SQL语句 - 使用REPLACE INTO确保更新现有记录
|
||||
fields = ', '.join(data.keys())
|
||||
placeholders = ', '.join(['%s'] * len(data))
|
||||
values = tuple(data.values())
|
||||
|
||||
|
||||
sql = f"""
|
||||
REPLACE INTO t_chatroom_member ({fields})
|
||||
VALUES ({placeholders})
|
||||
"""
|
||||
|
||||
|
||||
self.execute_update(sql, values)
|
||||
|
||||
|
||||
self.logger.info(f"成功保存群聊{chatroom_id}的{len(member_details)}个成员详细信息")
|
||||
return True
|
||||
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"保存群聊{chatroom_id}的成员详细信息失败: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def process_chatroom_member_detail_response(self, chatroom_id: str, response: Dict) -> bool:
|
||||
"""处理获取群成员详情的API响应
|
||||
|
||||
@@ -377,14 +377,14 @@ class ContactsDBOperator(BaseDBOperator):
|
||||
if response.get('ret') != 200:
|
||||
self.logger.error(f"获取群聊{chatroom_id}成员详情失败: {response.get('msg')}")
|
||||
return False
|
||||
|
||||
|
||||
data = response.get('data', [])
|
||||
if not data:
|
||||
self.logger.warning(f"群聊{chatroom_id}成员详情数据为空")
|
||||
return False
|
||||
|
||||
|
||||
return self.save_chatroom_member_detail(chatroom_id, data)
|
||||
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"处理群聊{chatroom_id}成员详情数据失败: {e}")
|
||||
return False
|
||||
return False
|
||||
|
||||
@@ -22,7 +22,8 @@ class MessageStorageDB(BaseDBOperator):
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||||
"""
|
||||
params = (
|
||||
msg.roomid, now_time, msg.sender, msg.content, msg.msg_type, msg.content, msg.msg_id, msg.msg_source, "")
|
||||
msg.roomid, now_time, msg.sender, msg.content, msg.msg_type, msg.content.xml_content, msg.msg_id,
|
||||
msg.msg_source, "")
|
||||
result = self.execute_update(sql, params)
|
||||
return result
|
||||
|
||||
|
||||
1
gewechat/call_back_message/Gewechat.apifox.json
Normal file
1
gewechat/call_back_message/Gewechat.apifox.json
Normal file
File diff suppressed because one or more lines are too long
4616
gewechat/call_back_message/Gewechat.md
Normal file
4616
gewechat/call_back_message/Gewechat.md
Normal file
File diff suppressed because one or more lines are too long
18
gewechat/client/fetchContactsList.py
Normal file
18
gewechat/client/fetchContactsList.py
Normal file
@@ -0,0 +1,18 @@
|
||||
import requests
|
||||
import json
|
||||
|
||||
url = "/contacts/fetchContactsList"
|
||||
|
||||
base_url="http://192.168.2.240:2531/v2/api"
|
||||
payload = json.dumps({
|
||||
"appId": "wx_3BC6eSHGE5xEm_hH3__7c"
|
||||
})
|
||||
|
||||
headers = {
|
||||
'X-GEWE-TOKEN': 'cb43f52db27e4a56bb6ec7da54373582',
|
||||
'Content-Type': 'application/json'
|
||||
}
|
||||
|
||||
response = requests.request("POST", base_url+url, headers=headers, data=payload)
|
||||
|
||||
print(response.text)
|
||||
@@ -121,6 +121,3 @@ class MessageUtil:
|
||||
members = {member["wxid"]: member["nickName"] for member in data["data"]["memberList"]}
|
||||
return members
|
||||
|
||||
def download_file_from_url(self, url: str, target_dir: str) -> str:
|
||||
# 根据获取的文件地址,从server 下载 :http://{服务ip}:2532/download/{接口返回的文件路径}
|
||||
return ""
|
||||
|
||||
103
robot.py
103
robot.py
@@ -34,6 +34,8 @@ from xiuren.xiuren_pdf import generate_pdf_from_images
|
||||
from db.connection import DBConnectionManager
|
||||
from message_util import MessageUtil
|
||||
|
||||
from db.contacts_db import ContactsDBOperator
|
||||
|
||||
|
||||
class Robot(Job):
|
||||
"""个性化自己的机器人
|
||||
@@ -104,6 +106,7 @@ class Robot(Job):
|
||||
# 消息存档模块初始化,自动完成入库动作
|
||||
self.message_storage = MessageStorage(self.client)
|
||||
|
||||
self.contacts_db = ContactsDBOperator(self.db_manager)
|
||||
@staticmethod
|
||||
def value_check(args: dict) -> bool:
|
||||
if args:
|
||||
@@ -404,67 +407,92 @@ class Robot(Job):
|
||||
self.LOG.error(f"xiu_ren_pdf_send error:{e}")
|
||||
|
||||
def get_all_contacts(self) -> dict:
|
||||
"""获取所有联系人信息并保存到数据库"""
|
||||
from db.contacts_db import ContactsDBOperator
|
||||
|
||||
"""获取所有联系人信息并返回字典格式 {wxid: nickname}"""
|
||||
try:
|
||||
contacts_dict = {}
|
||||
# 获取所有联系人的wxid列表
|
||||
contacts_wxids = self.client.fetch_contacts_list(self.app_id)
|
||||
self.LOG.info(f"contacts_wxids: {contacts_wxids}")
|
||||
if not contacts_wxids:
|
||||
self.LOG.warning("获取联系人列表为空")
|
||||
# 获取所有联系人列表
|
||||
response = self.client.fetch_contacts_list(self.app_id)
|
||||
self.LOG.info(f"获取联系人列表响应: {response}")
|
||||
|
||||
if not response or response.get("ret") != 200:
|
||||
self.LOG.warning(f"获取联系人列表失败: {response}")
|
||||
return contacts_dict
|
||||
|
||||
# 初始化联系人数据库操作类
|
||||
contacts_db = ContactsDBOperator()
|
||||
|
||||
|
||||
# 从响应中提取联系人数据
|
||||
contact_data = response.get("data", {})
|
||||
|
||||
# 处理好友列表
|
||||
friends = contact_data.get("friends", [])
|
||||
for wxid in friends:
|
||||
contacts_dict[wxid] = wxid # 默认使用wxid作为昵称
|
||||
|
||||
# 处理群聊列表
|
||||
chatrooms = contact_data.get("chatrooms", [])
|
||||
for chatroom_id in chatrooms:
|
||||
contacts_dict[chatroom_id] = chatroom_id
|
||||
# 如果是群聊,则获取群成员信息
|
||||
self.update_chatroom_member_details(chatroom_id)
|
||||
|
||||
# 处理公众号列表
|
||||
ghs = contact_data.get("ghs", [])
|
||||
for gh_id in ghs:
|
||||
contacts_dict[gh_id] = gh_id
|
||||
|
||||
# 获取联系人详细信息(昵称等)
|
||||
self.update_contact_details(contacts_dict)
|
||||
|
||||
self.LOG.info(f"成功获取并保存{len(contacts_dict)}个联系人信息")
|
||||
return contacts_dict
|
||||
|
||||
except Exception as e:
|
||||
self.LOG.error(f"获取联系人信息失败: {e}")
|
||||
return {}
|
||||
|
||||
def update_contact_details(self, contacts_dict):
|
||||
"""更新联系人详细信息(昵称等)"""
|
||||
try:
|
||||
# 将wxid列表分批处理,每批50个
|
||||
batch_size = 50
|
||||
for i in range(0, len(contacts_wxids), batch_size):
|
||||
batch_wxids = contacts_wxids[i:i + batch_size]
|
||||
|
||||
wxids = list(contacts_dict.keys())
|
||||
|
||||
for i in range(0, len(wxids), batch_size):
|
||||
batch_wxids = wxids[i:i + batch_size]
|
||||
|
||||
# 批量获取联系人详细信息
|
||||
contact_info = self.client.get_detail_info(self.app_id, batch_wxids)
|
||||
|
||||
|
||||
# 处理返回的数据
|
||||
if contact_info and contact_info.get("ret") == 200 and "data" in contact_info:
|
||||
contact_data = contact_info.get("data", [])
|
||||
|
||||
|
||||
if contact_data:
|
||||
for contact in contact_data:
|
||||
user_name = contact.get("userName")
|
||||
if not user_name:
|
||||
if not user_name or user_name not in contacts_dict:
|
||||
continue
|
||||
|
||||
|
||||
# 更新昵称
|
||||
contacts_dict[user_name] = contact.get("nickName") or user_name
|
||||
|
||||
try:
|
||||
# 判断联系人类型
|
||||
contact_type = "friends" # 默认为好友类型
|
||||
if user_name.endswith("@chatroom"):
|
||||
contact_type = "chatrooms"
|
||||
# 如果是群聊,则获取群成员信息
|
||||
self.update_chatroom_member_details(user_name)
|
||||
elif user_name.startswith("gh_"):
|
||||
contact_type = "ghs"
|
||||
|
||||
|
||||
# 保存到数据库
|
||||
contacts_db.save_contacts([contact], contact_type)
|
||||
|
||||
# 添加到返回字典
|
||||
contacts_dict[user_name] = contact.get("nickName") or user_name
|
||||
|
||||
self.contacts_db.save_contacts([contact], contact_type)
|
||||
|
||||
except Exception as e:
|
||||
self.LOG.error(f"处理联系人 {user_name} 失败: {e}")
|
||||
continue
|
||||
else:
|
||||
self.LOG.error(f"获取联系人详情失败: {contact_info}")
|
||||
|
||||
self.LOG.info(f"成功获取并保存{len(contacts_dict)}个联系人信息")
|
||||
return contacts_dict
|
||||
|
||||
else:
|
||||
self.LOG.error(f"获取联系人详情失败: {contact_info}")
|
||||
|
||||
except Exception as e:
|
||||
self.LOG.error(f"获取联系人信息失败: {e}")
|
||||
return {}
|
||||
self.LOG.error(f"更新联系人详细信息失败: {e}")
|
||||
|
||||
def update_chatroom_member_details(self, chatroom_id):
|
||||
"""更新群成员详细信息"""
|
||||
@@ -485,10 +513,7 @@ class Robot(Job):
|
||||
member_wxids # 直接传递列表,不需要转换为集合
|
||||
)
|
||||
|
||||
# 使用ContactsDBOperator处理响应
|
||||
from db.contacts_db import ContactsDBOperator
|
||||
contacts_db = ContactsDBOperator()
|
||||
success = contacts_db.process_chatroom_member_detail_response(chatroom_id, details_response)
|
||||
success = self.contacts_db.process_chatroom_member_detail_response(chatroom_id, details_response)
|
||||
|
||||
if success:
|
||||
self.LOG.info(f"成功更新群聊{chatroom_id}的成员详细信息")
|
||||
|
||||
@@ -234,7 +234,6 @@ class ContactManager:
|
||||
# "province": cnt.get("province", ""),
|
||||
# "city": cnt.get("city", ""),
|
||||
# "gender": gender}
|
||||
self._friends = self.message_util.get_friends()
|
||||
self._logger.info(f"联系人信息已刷新,共 {len(new_contacts)} 个联系人")
|
||||
self._classify_contacts()
|
||||
|
||||
|
||||
@@ -134,19 +134,18 @@ class MessageStorage:
|
||||
# 解析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.msg_id} -> {download_path}")
|
||||
|
||||
if file_url:
|
||||
logger.info(f"记录gewe服务端图片路径成功: {msg.msg_id} -> {file_url}")
|
||||
# 后续如果需要使用,则去服务器端提取图片
|
||||
# 直接使用下载后的路径更新数据库
|
||||
self.message_db.update_message_image_path(msg.msg_id, download_path)
|
||||
self.message_db.update_message_image_path(msg.msg_id, file_url)
|
||||
|
||||
return {
|
||||
'success': True,
|
||||
'message_id': msg.msg_id,
|
||||
'roomid': msg.roomid,
|
||||
'sender': msg.sender,
|
||||
'file_path': download_path
|
||||
'file_path': file_url
|
||||
}
|
||||
else:
|
||||
return {
|
||||
@@ -162,7 +161,7 @@ class MessageStorage:
|
||||
'message_id': msg.msg_id,
|
||||
'roomid': msg.roomid,
|
||||
'sender': msg.sender,
|
||||
'error': "WCF实例不存在或消息ID无效"
|
||||
'error': "实例不存在或消息ID无效"
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"图片处理出错: {msg.msg_id}, 错误: {e}")
|
||||
@@ -174,10 +173,6 @@ class MessageStorage:
|
||||
'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:
|
||||
|
||||
Reference in New Issue
Block a user