""" 数据处理工具类，用于处理聊天记录和联系人信息 """ import os import sqlite3 from typing import List, Dict, Tuple from datetime import datetime import re class DataProcessor: def __init__(self): """初始化数据处理器""" # 获取当前文件所在目录的上两级目录作为项目根目录 project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # 使用os.path.join构建跨平台的路径 db_dir = os.path.join(project_root, "app", "Database", "Msg") self.micro_msg_db_path = os.path.join(db_dir, "MicroMsg.db") self.msg_db_path = os.path.join(db_dir, "MSG.db") self.misc_db_path = os.path.join(db_dir, "MISC.db") print(f"[DataProcessor] 数据库路径配置:") print(f"MicroMsg数据库: {self.micro_msg_db_path}") print(f"MSG数据库: {self.msg_db_path}") print(f"MISC数据库: {self.misc_db_path}") self.micro_msg_conn = None self.msg_conn = None self.misc_conn = None self.init_database() def check_misc_tables(self): """检查MISC数据库的表结构""" if not self.misc_conn: raise Exception("MISC 数据库未连接") try: cursor = self.misc_conn.cursor() # 获取所有表名 cursor.execute("SELECT name FROM sqlite_master WHERE type='table'") tables = cursor.fetchall() print("\nMISC数据库中的表:") for table in tables: print(f"\n表名: {table[0]}") cursor.execute(f"PRAGMA table_info({table[0]})") columns = cursor.fetchall() for col in columns: print(f" 列名: {col[1]}, 类型: {col[2]}") except sqlite3.Error as e: print(f"检查表结构失败：{str(e)}") finally: if cursor: cursor.close() def init_database(self): """初始化数据库连接""" if not os.path.exists(os.path.dirname(self.micro_msg_db_path)): print(f"[DataProcessor] 数据库目录不存在，尝试创建: {os.path.dirname(self.micro_msg_db_path)}") try: os.makedirs(os.path.dirname(self.micro_msg_db_path), exist_ok=True) except Exception as e: print(f"[DataProcessor] 创建数据库目录失败: {str(e)}") if os.path.exists(self.micro_msg_db_path): try: self.micro_msg_conn = sqlite3.connect(self.micro_msg_db_path, check_same_thread=False) print(f"[DataProcessor] 成功连接到 MicroMsg 数据库: {self.micro_msg_db_path}") except sqlite3.Error as e: print(f"[DataProcessor] 连接 MicroMsg 数据库失败: {str(e)}") self.micro_msg_conn = None else: print(f"[DataProcessor] MicroMsg 数据库文件不存在: {self.micro_msg_db_path}") if os.path.exists(self.msg_db_path): try: self.msg_conn = sqlite3.connect(self.msg_db_path, check_same_thread=False) print(f"[DataProcessor] 成功连接到 MSG 数据库: {self.msg_db_path}") except sqlite3.Error as e: print(f"[DataProcessor] 连接 MSG 数据库失败: {str(e)}") self.msg_conn = None else: print(f"[DataProcessor] MSG 数据库文件不存在: {self.msg_db_path}") if os.path.exists(self.misc_db_path): try: self.misc_conn = sqlite3.connect(self.misc_db_path, check_same_thread=False) print(f"[DataProcessor] 成功连接到 MISC 数据库: {self.misc_db_path}") # 检查MISC数据库的表结构 self.check_misc_tables() except sqlite3.Error as e: print(f"[DataProcessor] 连接 MISC 数据库失败: {str(e)}") self.misc_conn = None else: print(f"[DataProcessor] MISC 数据库文件不存在: {self.misc_db_path}") # 如果数据库未连接，尝试从其他位置查找 if not all([self.micro_msg_conn, self.msg_conn, self.misc_conn]): print("[DataProcessor] 尝试从其他位置查找数据库...") alt_db_paths = [ os.path.join(project_root, "app", "DataBase", "Msg"), # 注意大小写 os.path.join(project_root, "app", "database", "msg"), os.path.join(project_root, "app", "Database", "msg"), os.path.join(project_root, "app", "database", "Msg"), ] for db_dir in alt_db_paths: if os.path.exists(db_dir): print(f"[DataProcessor] 找到数据库目录: {db_dir}") self.micro_msg_db_path = os.path.join(db_dir, "MicroMsg.db") self.msg_db_path = os.path.join(db_dir, "MSG.db") self.misc_db_path = os.path.join(db_dir, "MISC.db") # 重新尝试连接 if not self.micro_msg_conn and os.path.exists(self.micro_msg_db_path): try: self.micro_msg_conn = sqlite3.connect(self.micro_msg_db_path, check_same_thread=False) print(f"[DataProcessor] 成功连接到 MicroMsg 数据库: {self.micro_msg_db_path}") except sqlite3.Error as e: print(f"[DataProcessor] 连接 MicroMsg 数据库失败: {str(e)}") if not self.msg_conn and os.path.exists(self.msg_db_path): try: self.msg_conn = sqlite3.connect(self.msg_db_path, check_same_thread=False) print(f"[DataProcessor] 成功连接到 MSG 数据库: {self.msg_db_path}") except sqlite3.Error as e: print(f"[DataProcessor] 连接 MSG 数据库失败: {str(e)}") if not self.misc_conn and os.path.exists(self.misc_db_path): try: self.misc_conn = sqlite3.connect(self.misc_db_path, check_same_thread=False) print(f"[DataProcessor] 成功连接到 MISC 数据库: {self.misc_db_path}") self.check_misc_tables() except sqlite3.Error as e: print(f"[DataProcessor] 连接 MISC 数据库失败: {str(e)}") if all([self.micro_msg_conn, self.msg_conn, self.misc_conn]): print("[DataProcessor] 所有数据库连接成功") break def check_contact_table(self): """检查Contact表的结构""" if not self.micro_msg_conn: raise Exception("MicroMsg 数据库未连接") try: cursor = self.micro_msg_conn.cursor() # 获取表结构 cursor.execute("PRAGMA table_info(Contact)") columns = cursor.fetchall() print("\nContact表结构:") for col in columns: print(f"列名: {col[1]}, 类型: {col[2]}") # 获取一条示例数据 cursor.execute("SELECT * FROM Contact LIMIT 1") result = cursor.fetchone() if result: print("\n示例数据:") for i, col in enumerate(columns): print(f"{col[1]}: {result[i]}") except sqlite3.Error as e: print(f"检查表结构失败：{str(e)}") finally: if cursor: cursor.close() def get_all_contacts(self) -> List[Dict]: """获取所有联系人列表 Returns: List[Dict]: 联系人列表 """ if not self.micro_msg_conn: raise Exception("MicroMsg 数据库未连接") try: cursor = self.micro_msg_conn.cursor() # 查询所有联系人 query = """ SELECT Contact.UserName, Contact.Alias, Contact.Type, Contact.Remark, Contact.NickName, Contact.PYInitial, Contact.RemarkPYInitial, ContactHeadImgUrl.smallHeadImgUrl, ContactHeadImgUrl.bigHeadImgUrl, Contact.ExTraBuf, COALESCE(ContactLabel.LabelName, 'None') AS labelName FROM Contact INNER JOIN ContactHeadImgUrl ON Contact.UserName = ContactHeadImgUrl.usrName LEFT JOIN ContactLabel ON Contact.LabelIDList = ContactLabel.LabelId WHERE (Type!=4 AND VerifyFlag=0) AND NickName != '' ORDER BY CASE WHEN RemarkPYInitial = '' THEN PYInitial ELSE RemarkPYInitial END ASC """ try: cursor.execute(query) results = cursor.fetchall() except sqlite3.OperationalError: # 处理ContactLabel表不存在的情况 query = """ SELECT Contact.UserName, Contact.Alias, Contact.Type, Contact.Remark, Contact.NickName, Contact.PYInitial, Contact.RemarkPYInitial, ContactHeadImgUrl.smallHeadImgUrl, ContactHeadImgUrl.bigHeadImgUrl, Contact.ExTraBuf, 'None' as labelName FROM Contact INNER JOIN ContactHeadImgUrl ON Contact.UserName = ContactHeadImgUrl.usrName WHERE (Type!=4 AND VerifyFlag=0) AND NickName != '' ORDER BY CASE WHEN RemarkPYInitial = '' THEN PYInitial ELSE RemarkPYInitial END ASC """ cursor.execute(query) results = cursor.fetchall() print(f"\n找到 {len(results)} 个联系人") contacts = [] for result in results: original_name = result[3] if result[3] else result[4] # 优先使用备注名 display_name = re.sub(r'[\\/:*?"<>|\s\.]', '_', original_name) # 用于文件名等需要处理特殊字符的场景 if original_name: # 确保名称不为空 contacts.append({ 'wxid': result[0], 'alias': result[1], 'type': result[2], 'name': display_name, 'original_name': original_name, 'nickname': result[4], 'py_initial': result[5], 'remark_py_initial': result[6], 'small_avatar_url': result[7], 'big_avatar_url': result[8], 'extra_buf': result[9], 'label_name': result[10] }) return contacts except sqlite3.Error as e: raise Exception(f"数据库查询失败：{str(e)}") finally: if cursor: cursor.close() def get_chat_history(self, contact_id: str, start_date: str, end_date: str) -> List[Dict]: """获取指定时间范围内的聊天记录 Args: contact_id: 联系人ID start_date: 开始日期（格式：YYYY-MM-DD） end_date: 结束日期（格式：YYYY-MM-DD） Returns: List[Dict]: 聊天记录列表 """ try: cursor = self.msg_conn.cursor() # 查询聊天记录 query = """ SELECT localId, TalkerId, Type, SubType, IsSender, CreateTime, Status, StrContent, strftime('%Y-%m-%d %H:%M:%S', CreateTime, 'unixepoch', 'localtime') as create_time FROM MSG WHERE StrTalker = ? AND CreateTime BETWEEN strftime('%s', ?) AND strftime('%s', ?) AND Type = 1 -- 只获取文本消息 ORDER BY CreateTime ASC """ cursor.execute(query, (contact_id, start_date, end_date)) results = cursor.fetchall() # 转换为字典列表 chat_history = [] for msg in results: chat_history.append({ 'local_id': msg[0], 'talker_id': msg[1], 'type': msg[2], 'sub_type': msg[3], 'is_sender': msg[4], 'timestamp': msg[5], 'status': msg[6], 'message': msg[7], 'create_time': msg[8] }) return chat_history except sqlite3.Error as e: raise Exception(f"数据库查询失败：{str(e)}") finally: if cursor: cursor.close() def analyze_chat_content(self, chat_history: List[Dict]) -> Dict: """分析聊天内容，提取关键信息 Args: chat_history: 聊天记录列表 Returns: Dict: 分析结果，包含关键词、聊天频率等信息 """ # 实现聊天内容分析逻辑 pass def get_contact_avatar(self, wxid: str) -> bytes: """获取联系人头像数据 Args: wxid: 联系人的微信ID Returns: bytes: 头像的二进制数据 """ try: print(f"[DataProcessor] 开始获取头像 - 联系人ID: {wxid}") cursor = self.misc_conn.cursor() # 从ContactHeadImg1表获取头像数据 query = "SELECT smallHeadBuf FROM ContactHeadImg1 WHERE usrName=?" cursor.execute(query, (wxid,)) result = cursor.fetchone() if result and result[0]: data = result[0] print(f"[DataProcessor] 成功获取头像数据 - 联系人ID: {wxid}, 数据大小: {len(data)} 字节") return data else: print(f"[DataProcessor] 未找到头像数据 - 联系人ID: {wxid}") return None except sqlite3.Error as e: print(f"[DataProcessor] 获取头像失败 - 联系人ID: {wxid}, 错误: {str(e)}") return None finally: if cursor: cursor.close() def close(self): """关闭数据库连接""" if self.micro_msg_conn: self.micro_msg_conn.close() if self.msg_conn: self.msg_conn.close() if self.misc_conn: self.misc_conn.close() def __del__(self): self.close()