解锁并提取Linux客户端微信数据库 (vibe coded)
1# -*- coding: utf-8 -*-#
2# -------------------------------------------------------------------------------
3# Name: exportCSV.py
4# Description:
5# Author: xaoyaoo
6# Date: 2024/04/20
7# -------------------------------------------------------------------------------
8import csv
9import json
10import os
11from wxdump_linux.db import DBHandler
12
13
14def export_csv(wxid, outpath, db_config, my_wxid="我", page_size=5000):
15 if not os.path.exists(outpath):
16 outpath = os.path.join(os.getcwd(), "export" + os.sep + wxid)
17 if not os.path.exists(outpath):
18 os.makedirs(outpath)
19
20 db = DBHandler(db_config, my_wxid)
21
22 count = db.get_msgs_count(wxid)
23 chatCount = count.get(wxid, 0)
24 if chatCount == 0:
25 return False, "没有聊天记录"
26
27 if page_size > chatCount:
28 page_size = chatCount + 1
29
30 users = {}
31 for i in range(0, chatCount, page_size):
32 start_index = i
33 data, users_t = db.get_msgs(wxid, start_index, page_size)
34 print(users, users_t)
35 users.update(users_t)
36
37 if len(data) == 0:
38 return False, "没有聊天记录"
39
40 save_path = os.path.join(outpath, f"{wxid}_{i}_{i + page_size}.csv")
41
42 with open(save_path, "w", encoding="utf-8", newline='') as f:
43 csv_writer = csv.writer(f, quoting=csv.QUOTE_MINIMAL)
44
45 csv_writer.writerow(["id", "MsgSvrID", "type_name", "is_sender", "talker", "room_name", "msg", "src",
46 "CreateTime"])
47 for row in data:
48 id = row.get("id", "")
49 MsgSvrID = row.get("MsgSvrID", "")
50 type_name = row.get("type_name", "")
51 is_sender = row.get("is_sender", "")
52 talker = row.get("talker", "")
53 room_name = row.get("room_name", "")
54 msg = row.get("msg", "")
55 src = row.get("src", "")
56 CreateTime = row.get("CreateTime", "")
57 csv_writer.writerow([id, MsgSvrID, type_name, is_sender, talker, room_name, msg, src, CreateTime])
58 with open(os.path.join(outpath, "users.json"), "w", encoding="utf-8") as f:
59 json.dump(users, f, ensure_ascii=False, indent=4)
60 return True, f"导出成功: {outpath}"
61
62
63if __name__ == '__main__':
64 pass