dzming.li
解锁并提取Linux客户端微信数据库 (vibe coded)
1# -*- coding: utf-8 -*-#
2# -------------------------------------------------------------------------------
3# Name: __init__.py
4# Description: db
5# Author: xaoyaoo
6# Date: 2024/04/15
7# -------------------------------------------------------------------------------
8from .utils import download_file, dat2img
9
10from .db_favorite import FavoriteHandler
11from .db_msg import MsgHandler
12from .db_micro import MicroHandler
13from .db_media import MediaHandler
14from .db_open_im_contact import OpenIMContactHandler
15from .db_public_msg import PublicMsgHandler
16from .db_open_im_media import OpenIMMediaHandler
17from .db_sns import SnsHandler
18
19
20class DBHandler(MicroHandler, MediaHandler, OpenIMContactHandler, PublicMsgHandler, OpenIMMediaHandler,
21 FavoriteHandler, SnsHandler):
22 _class_name = "DBHandler"
23
24 def __init__(self, db_config, my_wxid, *args, **kwargs):
25 self.config = db_config
26 self.my_wxid = my_wxid
27
28 super().__init__(self.config)
29 # 加速查询索引
30 self.Micro_add_index()
31 self.Msg_add_index()
32 self.PublicMsg_add_index()
33 self.Media_add_index()
34
35 def get_user(self, word=None, wxids=None, labels=None):
36 """
37 获取联系人列表
38 :param word: 搜索关键字
39 :param wxids: wxid列表
40 :param labels: 标签列表
41 :return: 联系人dict {wxid: {}}
42 """
43 users = self.get_user_list(word=word, wxids=wxids, label_ids=labels)
44 users.update(self.get_im_user_list(word=word, wxids=wxids))
45 return users
46
47 def get_msgs(self, wxids: list or str = "", start_index=0, page_size=500, msg_type: str = "",
48 msg_sub_type: str = "", start_createtime=None, end_createtime=None):
49 """
50 获取聊天记录列表
51 :param wxids:[ wxid]
52 :param start_index: 起始索引
53 :param page_size: 页大小
54 :param msg_type: 消息类型
55 :param msg_sub_type: 消息子类型
56 :param start_createtime: 开始时间
57 :param end_createtime: 结束时间
58 :return: 聊天记录列表 {"id": _id, "MsgSvrID": str(MsgSvrID), "type_name": type_name, "is_sender": IsSender,
59 "talker": talker, "room_name": StrTalker, "msg": msg, "src": src, "extra": {},
60 "CreateTime": CreateTime, }
61 """
62 msgs0, wxid_list0 = self.get_msg_list(wxids=wxids, start_index=start_index, page_size=page_size,
63 msg_type=msg_type,
64 msg_sub_type=msg_sub_type, start_createtime=start_createtime,
65 end_createtime=end_createtime, my_talker=self.my_wxid)
66 msgs1, wxid_list1 = self.get_plc_msg_list(wxids=wxids, start_index=start_index, page_size=page_size,
67 msg_type=msg_type,
68 msg_sub_type=msg_sub_type, start_createtime=start_createtime,
69 end_createtime=end_createtime, my_talker=self.my_wxid)
70 msgs = msgs0 + msgs1
71 wxid_list = wxid_list0 + wxid_list1
72
73 users = self.get_user(wxids=wxid_list)
74 return msgs, users
75
76 def get_msgs_count(self, wxids: list = ""):
77 chat_count = self.get_m_msg_count(wxids)
78 chat_count1 = self.get_plc_msg_count(wxids)
79 # 合并两个字典，相同key，则将value相加
80 count = {k: chat_count.get(k, 0) + chat_count1.get(k, 0) for k in
81 list(set(list(chat_count.keys()) + list(chat_count1.keys())))}
82 return count
83
84
85__all__ = ["DBHandler", "FavoriteHandler", "MsgHandler", "MicroHandler", "MediaHandler",
86 "OpenIMContactHandler", "PublicMsgHandler", "OpenIMMediaHandler", "SnsHandler"]