cyonjan/PyWxDump
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加联系人标签显示

Browse Source
This commit is contained in:
xaoyaoo 2024-06-12 00:35:03 +08:00
parent 856d2ae123
commit 728e6ba062
4 changed files with 95 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
pywxdump/api/api.py
View File

@ -22,7 +22,7 @@ from pywxdump.api.utils import read_session, get_session_wxids, save_session, er
from pywxdump import read_info, VERSION_LIST, batch_decrypt, BiasAddr, merge_db, decrypt_merge, merge_real_time_db
from pywxdump.dbpreprocess import wxid2userinfo, ParsingMSG, get_user_list, get_recent_user_list, ParsingMediaMSG, \
download_file, export_csv, export_json
download_file, export_csv, export_json, ParsingMicroMsg
from pywxdump.dbpreprocess.utils import dat2img
# app = Flask(__name__, static_folder='../ui/web/dist', static_url_path='/')
@ -178,7 +178,18 @@ def recent_user_list():
merge_path = read_session(g.sf, my_wxid, "merge_path")
user_list = get_recent_user_list(merge_path, merge_path, limit=200)
return ReJson(0, user_list)
@api.route('/api/user_labels_dict', methods=["GET", 'POST'])
@error9999
def user_labels_dict():
"""
获取标签字典
:return:
"""
my_wxid = read_session(g.sf, "test", "last")
if not my_wxid: return ReJson(1001, body="my_wxid is required")
merge_path = read_session(g.sf, my_wxid, "merge_path")
user_labels_dict = ParsingMicroMsg(merge_path).labels_dict()
return ReJson(0, user_labels_dict)
@api.route('/api/user_list', methods=["GET", 'POST'])
@error9999
@ -225,7 +236,7 @@ def wxid2user():
@error9999
def mywxid():
"""
获取联系人列表
获取我的微信id
:return:
"""
my_wxid = read_session(g.sf, "test", "last")

5
pywxdump/dbpreprocess/__init__.py
View File

@ -36,8 +36,9 @@ def get_user_list(MicroMsg_db_path, OpenIMContact_db_path=None, word=None):
parsing_openimcontact = ParsingOpenIMContact(OpenIMContact_db_path)
users += parsing_openimcontact.user_list(word=word)
# 去重
users = [dict(t) for t in {tuple(d.items()) for d in users}]
return users
# print(users)
unique_users = [dict(t) for t in {tuple(d.items()) for d in users}]
return unique_users
def get_recent_user_list(MicroMsg_db_path, OpenIMContact_db_path=None, limit=200):

135
pywxdump/dbpreprocess/parsingMicroMsg.py
View File

@ -28,6 +28,61 @@ class ParsingMicroMsg(DatabaseBase):
except Exception as e:
return None
def get_ExtraBuf(self, ExtraBuf: bytes):
"""
读取ExtraBuf（联系人表）
:param ExtraBuf:
:return:
"""
if not ExtraBuf:
return None
try:
buf_dict = {
'DDF32683': '0', '74752C06': '性别[1男2女]', '88E28FCE': '2', '761A1D2D': '3', '0263A0CB': '4',
'0451FF12': '5',
'228C66A8': '6', '46CF10C4': '个性签名', 'A4D9024A': '国', 'E2EAA8D1': '省', '1D025BBF': '市',
'4D6C4570': '11',
'F917BCC0': '公司名称', '759378AD': '手机号', '4335DFDD': '14', 'DE4CDAEB': '15', 'A72BC20A': '16',
'069FED52': '17',
'9B0F4299': '18', '3D641E22': '19', '1249822C': '20', '4EB96D85': '企微属性', 'B4F73ACB': '22',
'0959EB92': '23',
'3CF4A315': '24', 'C9477AC60201E44CD0E8': '26', 'B7ACF0F5': '28', '57A7B5A8': '29',
'81AE19B4': '朋友圈背景',
'695F3170': '31', 'FB083DD9': '32', '0240E37F': '33', '315D02A3': '34', '7DEC0BC3': '35',
'0E719F13': '备注图片',
'16791C90': '37'
}
rdata = {}
for buf_name in buf_dict:
rdata_name = buf_dict[buf_name]
buf_name = bytes.fromhex(buf_name)
offset = ExtraBuf.find(buf_name)
if offset == -1:
rdata[rdata_name] = ""
continue
offset += len(buf_name)
type_id = ExtraBuf[offset: offset + 1]
offset += 1
if type_id == b"\x04":
rdata[rdata_name] = int.from_bytes(ExtraBuf[offset: offset + 4], "little")
elif type_id == b"\x18":
length = int.from_bytes(ExtraBuf[offset: offset + 4], "little")
rdata[rdata_name] = ExtraBuf[offset + 4: offset + 4 + length].decode("utf-16").rstrip("\x00")
elif type_id == b"\x17":
length = int.from_bytes(ExtraBuf[offset: offset + 4], "little")
rdata[rdata_name] = ExtraBuf[offset + 4: offset + 4 + length].decode("utf-8").rstrip("\x00")
elif type_id == b"\x05":
rdata[rdata_name] = f"0x{ExtraBuf[offset: offset + 8].hex()}"
return rdata
except Exception as e:
print(f'解析错误:\n{e}')
return None
def ChatRoom_RoomData(self, RoomData):
# 读取群聊数据,主要为 wxid，以及对应昵称
if RoomData is None or not isinstance(RoomData, bytes):
@ -54,7 +109,7 @@ class ParsingMicroMsg(DatabaseBase):
wxid = "','".join(wxid)
wxid = f"'{wxid}'"
# 获取username是wx_id的用户
sql = ("SELECT A.UserName, A.NickName, A.Remark,A.Alias,A.Reserved6,B.bigHeadImgUrl "
sql = ("SELECT A.UserName, A.NickName, A.Remark,A.Alias,A.Reserved6,B.bigHeadImgUrl,A.LabelIDList "
"FROM Contact A,ContactHeadImgUrl B "
f"WHERE A.UserName = B.usrName AND A.UserName in ({wxid}) "
"ORDER BY NickName ASC;")
@ -64,9 +119,10 @@ class ParsingMicroMsg(DatabaseBase):
users = {}
for row in result:
# 获取wxid,昵称，备注，描述，头像
username, nickname, remark, Alias, describe, headImgUrl = row
username, nickname, remark, Alias, describe, headImgUrl, LabelIDList = row
LabelIDList = LabelIDList.split(",") if LabelIDList else []
users[username] = {"wxid": username, "nickname": nickname, "remark": remark, "account": Alias,
"describe": describe, "headImgUrl": headImgUrl}
"describe": describe, "headImgUrl": headImgUrl, "LabelIDList": tuple(LabelIDList)}
return users
def user_list(self, word=None):
@ -77,7 +133,7 @@ class ParsingMicroMsg(DatabaseBase):
"""
users = []
sql = (
"SELECT A.UserName, A.NickName, A.Remark,A.Alias,A.Reserved6,B.bigHeadImgUrl "
"SELECT A.UserName, A.NickName, A.Remark,A.Alias,A.Reserved6,B.bigHeadImgUrl,A.LabelIDList "
"FROM Contact A left join ContactHeadImgUrl B on A.UserName==B.usrName "
"ORDER BY A.NickName DESC;")
if word:
@ -95,11 +151,12 @@ class ParsingMicroMsg(DatabaseBase):
if not result:
return []
for row in result:
# 获取wxid,昵称，备注，描述，头像
username, nickname, remark, Alias, describe, headImgUrl = row
# 获取wxid,昵称，备注，描述，头像,标签
username, nickname, remark, Alias, describe, headImgUrl, LabelIDList = row
LabelIDList = LabelIDList.split(",") if LabelIDList else []
users.append(
{"wxid": username, "nickname": nickname, "remark": remark, "account": Alias,
"describe": describe, "headImgUrl": headImgUrl})
"describe": describe, "headImgUrl": headImgUrl if headImgUrl else "", "LabelIDList": tuple(LabelIDList)})
return users
def recent_chat_wxid(self):
@ -167,58 +224,18 @@ class ParsingMicroMsg(DatabaseBase):
"Announcement": Announcement, "AnnouncementEditor": AnnouncementEditor, "wxid2remark": wxid2remark})
return rooms
def get_ExtraBuf(self, ExtraBuf: bytes):
def labels_dict(self, id_is_key=True):
"""
读取ExtraBuf（联系人表）
:param ExtraBuf:
读取标签列表
:param label_list:
:return:
"""
if not ExtraBuf:
return None
try:
buf_dict = {
'DDF32683': '0', '74752C06': '性别[1男2女]', '88E28FCE': '2', '761A1D2D': '3', '0263A0CB': '4',
'0451FF12': '5',
'228C66A8': '6', '46CF10C4': '个性签名', 'A4D9024A': '国', 'E2EAA8D1': '省', '1D025BBF': '市',
'4D6C4570': '11',
'F917BCC0': '公司名称', '759378AD': '手机号', '4335DFDD': '14', 'DE4CDAEB': '15', 'A72BC20A': '16',
'069FED52': '17',
'9B0F4299': '18', '3D641E22': '19', '1249822C': '20', '4EB96D85': '企微属性', 'B4F73ACB': '22',
'0959EB92': '23',
'3CF4A315': '24', 'C9477AC60201E44CD0E8': '26', 'B7ACF0F5': '28', '57A7B5A8': '29',
'81AE19B4': '朋友圈背景',
'695F3170': '31', 'FB083DD9': '32', '0240E37F': '33', '315D02A3': '34', '7DEC0BC3': '35',
'0E719F13': '备注图片',
'16791C90': '37'
}
rdata = {}
for buf_name in buf_dict:
rdata_name = buf_dict[buf_name]
buf_name = bytes.fromhex(buf_name)
offset = ExtraBuf.find(buf_name)
if offset == -1:
rdata[rdata_name] = ""
continue
offset += len(buf_name)
type_id = ExtraBuf[offset: offset + 1]
offset += 1
if type_id == b"\x04":
rdata[rdata_name] = int.from_bytes(ExtraBuf[offset: offset + 4], "little")
elif type_id == b"\x18":
length = int.from_bytes(ExtraBuf[offset: offset + 4], "little")
rdata[rdata_name] = ExtraBuf[offset + 4: offset + 4 + length].decode("utf-16").rstrip("\x00")
elif type_id == b"\x17":
length = int.from_bytes(ExtraBuf[offset: offset + 4], "little")
rdata[rdata_name] = ExtraBuf[offset + 4: offset + 4 + length].decode("utf-8").rstrip("\x00")
elif type_id == b"\x05":
rdata[rdata_name] = f"0x{ExtraBuf[offset: offset + 8].hex()}"
return rdata
except Exception as e:
print(f'解析错误:\n{e}')
return None
sql = "SELECT LabelId, LabelName FROM ContactLabel ORDER BY LabelName ASC;"
result = self.execute_sql(sql)
if not result:
return []
if id_is_key:
labels = {row[0]: row[1] for row in result}
else:
labels = {row[1]: row[0] for row in result}
return labels

4
pywxdump/dbpreprocess/parsingOpenIMContact.py
View File

@ -42,7 +42,7 @@ class ParsingOpenIMContact(DatabaseBase):
# 获取用户名、昵称、备注和聊天记录数量
username, nickname, remark, headImgUrl = row
users[username] = {"wxid": username, "nickname": nickname, "remark": remark, "account": "", "describe": "",
"headImgUrl": headImgUrl}
"headImgUrl": headImgUrl, "LabelIDList": ()}
return users
def user_list(self, word=None):
@ -70,5 +70,5 @@ class ParsingOpenIMContact(DatabaseBase):
username, nickname, remark, headImgUrl = row
users.append(
{"wxid": username, "nickname": nickname, "remark": remark, "account": "", "describe": "",
"headImgUrl": headImgUrl})
"headImgUrl": headImgUrl, "LabelIDList": ()})
return users