写了个脚本，读取数据库并转换为JSON格式

[html5syt]

这份代码大部分借助于AI，十分粗略，不过可以用，只写了群聊的SQL查询语句，理论上私聊也可以。

需要解密并去除密码后的数据库

实测在普通笔记本上导出9000多条聊天记录只需要5~10s。

我计划配合Flet写个渲染框架，复现当时的聊天记录。

这是代码：

import base64
import json
import time
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Union
import blackboxprotobuf
import threading
import sqlite3


def query_db(db_path, query):
    """
    执行SQLite查询并将结果转换为字典列表，自动尝试用Protobuf解析字节类型字段

    :param db_path: SQLite数据库文件路径
    :param query: SQL查询语句
    :return: 包含结果字典的列表，键为列名
    """
    conn = sqlite3.connect(db_path)
    try:
        cursor = conn.cursor()
        cursor.execute(query)

        # 获取列名信息
        columns = [col[0] for col in cursor.description] if cursor.description else []
        results = []

        # 遍历所有结果行
        for row in cursor.fetchall():
            processed_row = {}
            for idx, value in enumerate(row):
                col_name = columns[idx]

                # 处理字节类型字段
                if isinstance(value, bytes):
                    try:
                        # 尝试Protobuf解码
                        decoded, _ = blackboxprotobuf.decode_message(value)
                        processed_row[col_name] = decoded
                    except Exception as e:
                        # 解析失败保留原始值
                        print(e)
                        processed_row[col_name] = value
                else:
                    processed_row[col_name] = value

            results.append(processed_row)

        return results
    finally:
        conn.close()


def write_list_to_json(list, json_file_name):
    """
    将list写入到json文件
    :param list:
    :param json_file_name: 写入的json文件名字
    :param json_file_save_path: json文件存储路径
    :return:
    """
    # os.chdir(json_file_save_path)
    with open(json_file_name, "w") as f:
        json.dump(list, f)


def recursive_process(
    data: Union[dict, list], lock: threading.Lock
) -> Union[dict, list]:
    """
    递归地处理字典或列表中的字节字符串。
    """
    # 复制结构，避免修改原数据
    new_data = data.copy() if isinstance(data, dict) else list(data)

    # 处理字典
    if isinstance(data, dict):
        for key, value in data.items():
            if isinstance(value, bytes):
                # 替换字节字符串
                new_data[key] = process_byte_string(value, lock)
            elif isinstance(value, (dict, list)):
                # 递归处理子结构
                new_data[key] = recursive_process(value, lock)

    # 处理列表
    elif isinstance(data, list):
        for i, value in enumerate(data):
            if isinstance(value, bytes):
                # 替换字节字符串
                new_data[i] = process_byte_string(value, lock)
            elif isinstance(value, (dict, list)):
                # 递归处理子结构
                new_data[i] = recursive_process(value, lock)

    return new_data


def process_byte_string(byte_str: bytes, lock: threading.Lock) -> str:
    """
    处理字节字符串，按照要求转换。
    """
    try:
        # 1. 尝试 UTF-8 解码
        try:
            utf8_result = byte_str.decode("utf-8")
            return utf8_result
        except UnicodeDecodeError:
            # 如果 UTF-8 解码失败，尝试 ProtoBuf 解码
            try:
                decoded_message, _ = blackboxprotobuf.decode_message(byte_str)
                return f"ProtoBuf decoded: {json.dumps(decoded_message)}"
            except:
                # ProtoBuf 解码失败，转为 Base64 编码
                base64_result = base64.b64encode(byte_str).decode("utf-8")
                return base64_result
    finally:
        pass


def process_data_multithread(data: list, num_threads: int = 5) -> list:
    """
    使用多线程递归处理数据。
    """
    lock = threading.Lock()
    results = []
    with ThreadPoolExecutor(max_workers=num_threads) as executor:
        # 将数据拆分为多个子任务
        futures = []
        for chunk in data:
            future = executor.submit(recursive_process, chunk, lock)
            futures.append(future)

        # 收集结果
        for future in futures:
            results.append(future.result())

    return results


# 测试函数
# processed_data = process_data_multithread(input_data)
# print(json.dumps(processed_data, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    db_path = input("输入解密后的nt_msg相对于运行目录的路径：")
    group_list = input("请输入需要导出的群号，多个群号用空格分隔：").split()
    start_time_in = input("请输入开始时间（格式：2021-09-01 00:00:00）：")
    time_arr = time.strptime(start_time_in, "%Y-%m-%d %H:%M:%S")
    start_time = time.mktime(time_arr)  # 时间戳
    end_time_in = input("请输入结束时间（格式：2021-09-01 00:00:00）：")
    time_arr = time.strptime(end_time_in, "%Y-%m-%d %H:%M:%S")
    end_time = time.mktime(time_arr)  # 时间戳
    # # 测试用例
    # db_path = "nt_db/nt_msg.body.db"
    # group_list = []
    results = query_db(
        db_path,
        f"""SELECT *
FROM group_msg_table
WHERE "40050" <> 0
AND "40050" >= {int(start_time)}
AND "40050" < {int(end_time)}
AND "40021" IN ('{"','".join(map(str, group_list))}')
ORDER BY "40050" ;""",
    )

    results1 = process_data_multithread(results)

    with open(f"{'+'.join(map(str, group_list))}--S{start_time_in.replace(':', ';')}--E{end_time_in.replace(':', ';')}.json", "w") as f:
        json.dump(results1, f, ensure_ascii=False, indent=4)

顺便附上提示词：
我有一个sqlite3数据库，没有密码加密，请写一个Python函数，输入数据库路径和一条查询语句（字符串），返回字典编码的查询结果，键为列名。其中如果结果中值类型为byte，则尝试调用blackprotobuf库解析值并替换原来的值（相当于解密值）。

在一个Python程序中有一个列表，其中嵌套了许多字典和列表，它的一部分如下所示：

<字典数据>

请写一个Python函数，以多线程递归遍历检查字典的每一个值（检查到底，不能漏下），若检测为Byte字符串，则按照如下步骤转换并替换原值：

1. UTF-8解码，如错误转2
2. ProtoBuf解码，使用blackboxprotobuf库，如错误转3
3. 将原Byte字符串Base64编码
   （至少遍历检查3次）

[kuyunjian]

我用ai写了个py脚本，把你导出的json转换成可读txt

import json
import time


def parse_chat_type(chat_type):
    chat_types = {1: '私聊', 2: '群聊', 4: '频道', 103: '公众号', 102: '企业客服', 100: '临时会话'}
    return chat_types.get(chat_type, '未知类型')


def parse_msg_type(msg_type):
    msg_types = {0: '无消息', 1: '空白消息', 2: '文本消息', 3: '群文件', 5: '系统消息', 6: '语音消息',
                 7: '视频文件', 8: '合并转发消息', 9: '回复消息', 10: '红包', 11: '应用消息'}
    return msg_types.get(msg_type, '未知消息类型')


def parse_sub_msg_type(sub_msg_type):
    sub_msg_types = {0: '非常规text消息', 1: '普通文本消息', 2: '图片消息', 3: '群公告', 4: '撤回消息提醒',
                     8: '表情包', 11: '精华消息', 12: '拍一拍消息', 16: '群文件docx', 32: '平台文本消息',
                     33: '回复消息', 64: '群文件xlsx', 161: '存在链接', 512: '群文件zip',
                     2048: '群文件exe', 4096: '表情消息'}
    return sub_msg_types.get(sub_msg_type, '未知子消息类型')


def extract_content(msg):
    """
    尝试从 msg 中提取消息内容字段。原先假定结构为：
    msg["40800"]["40800"]["45101"]
    如果 msg["40800"] 为列表，则取列表的第一个元素再继续获取。
    """
    content_field = msg.get("40800", {})
    # 如果 content_field 是列表，取第一个元素
    if isinstance(content_field, list):
        if content_field:
            content_field = content_field[0]
        else:
            return '无内容'
    # 继续检查是否为字典类型
    if isinstance(content_field, dict):
        inner_field = content_field.get("40800", {})
        if isinstance(inner_field, dict):
            return inner_field.get("45101", '无内容')
    return '无内容'


def convert_timestamp(ts):
    """
    将时间戳转换为可读的时间字符串，格式为 YYYY-MM-DD HH:MM:SS
    """
    try:
        return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts))
    except Exception as e:
        return "无效时间"


def process_json(input_file, output_file):
    with open(input_file, 'r', encoding='utf-8') as f:
        data = json.load(f)

    chat_type = parse_chat_type(data[0].get("40010", 0))

    with open(output_file, 'w', encoding='utf-8') as f:
        f.write(f"聊天类型: {chat_type}\n")
        f.write("============================\n")

        total_messages = len(data)
        start_time = time.time()

        for index, msg in enumerate(data):
            msg_seq = msg.get("40003", '未知')
            msg_type = parse_msg_type(msg.get("40011", 0))
            sub_msg_type = parse_sub_msg_type(msg.get("40012", 0))
            sender = msg.get("40093", '未知发送者')
            content = extract_content(msg)

            # 提取并转换时间戳
            timestamp = msg.get("40050", None)
            readable_time = convert_timestamp(timestamp) if timestamp is not None else '无时间'

            f.write(f"时间: {readable_time}\n")
            f.write(f"唯一序号id: {msg_seq}\n")
            f.write(f"发送者: {sender}\n")
            f.write(f"消息类型: {msg_type} ({sub_msg_type})\n")
            f.write(f"消息内容: {content}\n")
            f.write("----------------------------\n")

            if time.time() - start_time >= 5:
                print(f"已处理 {index + 1}/{total_messages} 条消息...")
                start_time = time.time()

    print("处理完成，输出文件:", output_file)


if __name__ == "__main__":
    input_file = r"1.json"  # 替换为你的 json 文件路径
    output_file = "output.txt"
    process_json(input_file, output_file)

[kuyunjian]

我另外有1GB的数据文件，group这个表有十万行，预计一个群聊需要导出几小时，希望后面有大佬可以优化一下遍历结构或者分页或者采用cpugpu加速等技术

[html5syt]

我有个想法
可以分开群聊导出
毕竟数据库文件在自己手里
多复制几份一起解析也不是不可以

或者找找有没有办法使用已知的protobuf定义来更快解析（我当时使用blackboxprotobuf因为它门槛低，只需要把blob丢给它就能得到所有可用数据

[kuyunjian]

我有个想法 可以分开群聊导出 毕竟数据库文件在自己手里 多复制几份一起解析也不是不可以

或者找找有没有办法使用已知的protobuf定义来更快解析（我当时使用blackboxprotobuf因为它门槛低，只需要把blob丢给它就能得到所有可用数据

弄了一下，大概好像是解密方式限制了速度，和表的大小无关，我把一个群聊的聊天记录单独拿出来，并且用gpt针对写了这个表的代码，并加了一个计数器，只有5000余行的数据，总是在一些特定的行卡好一会儿

---

[kuyunjian]

我是真计算机小白，只会用gpt，所以我给这段代码添加了个超时，等我跑完我看看这些超时的数据和普通数据有什么区别。或者大佬们可以写一下其它的解密方法。（不然我就要学对应的教程了（））

---

[html5syt]

上次哪里看到过有人放出protobuf定义表的来着，谁会原版protobuf解析（需要定义表）的可以试试，速度应该会大幅度提升

[html5syt]

只有5000余行的数据，总是在一些特定的行卡好一会儿

这种很可能是遇到了嵌套N层的聊天记录，尤其是3层每层100条的那种，解密速度会特别慢（我写的是要求递归处理到底，为的是完整解析，实际上也可以写一个忽略或者是只解析1层protobuf然后下面几层保留原来的blob不变之类的）

[kuyunjian]

上次哪里看到过有人放出protobuf定义表的来着，谁会原版protobuf解析（需要定义表）的可以试试，速度应该会大幅度提升

issue#51这个大佬写得数据库解析私聊信息特别快，可能可以参考一下，我先看看重复导出是否还是数据库损坏差不多的地方database disk image is malformed数据库损坏的原因貌似找到了

[Tealina28]

syntax = "proto3";

message GroupSections { repeated GroupSection sections = 40800; }

message GroupSection {
  uint64 id = 45001;
  uint32 type = 45002;
  // 1：文本段，2：图片段，3：文件消息，6：表情段，7：引用段，
  // 8：提示消息（中间灰色），10：应用消息
  // 21：通话段
  // 26：动态消息

  // 引用
  string senderUid = 40020;
  string interlocutorUid = 40021;

  uint32 senderNum = 47403;
  uint32 quotedTimestamp = 47404;
//  uint32 interlocutorNum = 47411;

  GroupSection quotedSection = 47423;

  // 文字
  string text = 45101;

  // 文件
  string fileName = 45402; // imageFileName2
  uint64 fileSize = 45405;

  uint64 fileTimestamp = 45505; // ?

  // 图片
//  string imageFileName = ;

  string imageUrlLow = 45802;
  string imageUrlHigh = 45803;
  string imageUrlOrigin = 45804;

  string imageFilePath = 45812;

  string imageText = 45815;

  // 表情消息
  // 1: QQ 系统表情，2: emoji 表情
  // https://bot.q.qq.com/wiki/develop/api/openapi/emoji/model.html
  uint32 emojiId = 47601;
  string emojiText = 47602;

  // 应用消息
  string applicationMessage = 47901;

  // 通话消息
  string callStatus = 48153;
  string callText = 48157;

  // 动态消息
  GroupFeedMessage feedTitle = 48175;
  GroupFeedMessage feedContent = 48176;

  string feedUrl = 48180;
  string feedLogoUrl = 48181;
  uint32 feedPublisherNum = 48182;

  string feedJumpInfo = 48183;
  string feedPublisherUid = 48188;

  // 提示消息
  string noticeInfo = 48214;
  string noticeInfo2 = 48271; // ?
}

message GroupFeedMessage { string text = 48178; }

不是很完整（和正确），但应该可以用。
其实我在写群聊导出，但是被一些问题卡住了，也许明天能写好吧。

[Tealina28]

我刚刚往我的仓库推了个commit，你把最新版拉下来应该就可以用。不想要私聊的话可以在main注释掉，还是觉得慢的话可以在group里的线程池那里改线程数。为了快点发出来借助了一点llm，可能有问题，反馈或者提pr都行。

[html5syt]

我这周在学校里（高中牲一枚）反思了一下这个方法，其实JSON的导出和完全解密只是使其他程序解析更加方便（并且提升人类可读性），但是会以降低解析速度作为代价。
实际上如果只需要部分信息（如文字信息）那么JSON里消息段完全可以按照需求选择解析，或者解析超时的保持原样（base64编码）之后当需要的时候再另行解析。

我现在在做的解析程序考虑直接使用SQL命令读数据库，然后用户看到哪里解析到哪里。

（个人观点，如错请纠正🙏）

[kuyunjian]

我这周在学校里（高中牲一枚）反思了一下这个方法，其实JSON的导出和完全解密只是使其他程序解析更加方便（并且提升人类可读性），但是会以降低解析速度作为代价。 实际上如果只需要部分信息（如文字信息）那么JSON里消息段完全可以按照需求选择解析，或者解析超时的保持原样（base64编码）之后当需要的时候再另行解析。

我现在在做的解析程序考虑直接使用SQL命令读数据库，然后用户看到哪里解析到哪里。

（个人观点，如错请纠正🙏）

1. 你们有遇到我最新issue里面的数据库损坏的相同情况吗，我想把重心放在那上面
2. 我想既然另一个 db 文件group_msg_fts.db里面有所有的文本记录，又有和nt_msg.db匹配的唯一消息id，我觉得对应起来解密会不会更快
3. 我用gpt把我想说的话整理了一下,这是我最近尝试的方法：

### **聊天记录导出问题笔记**

1. **备份工具的使用**  
   - 使用 **QQ 官方备份工具** 将聊天记录备份到 **另一个小米手机** 上后，再进行导出，有可能成功导出包括图片在内的媒体文件。  
   - 媒体文件一般存储在 **媒体文件夹** 中，可以通过导出后的数据进行查找。

2. **大数据量导出失败的问题**  
   - 在遇到 **超大文本或大内存的数据库**（例如 **几GB的nt_msg.db**）时，导出到另一台手机时可能会出现导出失败的情况。  
   - 以 **5GB的nt_msg.db** 为例：  
     - 在原手机上，**nt_msg.db** 已经损坏。  
     - 但是原手机上的 **group_msg_fts.db** 并未损坏，仍可读取部分数据。  
   - 即使在表面上选择了“**导出全部聊天记录**”，但在导出过程中，涉及异常大的群聊时会导致失败。

3. **异常群聊的导出问题**  
   - 使用 **QQ 官方聊天记录导出工具**，针对 **异常大的群聊** 导出时会出现以下问题：  
     - **无法导出**：导出工具可能会卡在“**连接两个手机**”的界面，随后自动退出。  
     - 即便选择导出特定时间范围内的聊天记录，若涉及超大群聊，导出仍可能失败。  
   - 具体表现：  
     - 即使勾选了 **很小一部分时间范围** 内的数据，但只要群聊数据量过大，依然无法导出。  
     - 导出异常大的群聊时可能涉及预先解析全部聊天记录的过程，这一过程无法顺利完成，导致导出失败。  

4. **其他群聊的导出情况**  
   - **非异常大的群聊**：可以单独勾选并成功导出。  
   - **导出环境要求**：需要在 **原手机** 上进行导出操作，不能依靠转移到新手机后再导出。  

[html5syt]

3. 我用gpt把我想说的话整理了一下,这是我最近尝试的方法：

聊天记录导出问题笔记

1. 备份工具的使用

   • 使用 QQ 官方备份工具 将聊天记录备份到 另一个小米手机 上后，再进行导出，有可能成功导出包括图片在内的媒体文件。
   • 媒体文件一般存储在 媒体文件夹 中，可以通过导出后的数据进行查找。

2. 大数据量导出失败的问题

   • 在遇到 超大文本或大内存的数据库（例如 几GB的nt_msg.db）时，导出到另一台手机时可能会出现导出失败的情况。
   • 以 5GB的nt_msg.db 为例：
     • 在原手机上，nt_msg.db 已经损坏。
     • 但是原手机上的 group_msg_fts.db 并未损坏，仍可读取部分数据。
   • 即使在表面上选择了“导出全部聊天记录”，但在导出过程中，涉及异常大的群聊时会导致失败。

3. 异常群聊的导出问题

   • 使用 QQ 官方聊天记录导出工具，针对 异常大的群聊 导出时会出现以下问题：
     • 无法导出：导出工具可能会卡在“连接两个手机”的界面，随后自动退出。
     • 即便选择导出特定时间范围内的聊天记录，若涉及超大群聊，导出仍可能失败。
   • 具体表现：
     • 即使勾选了 很小一部分时间范围 内的数据，但只要群聊数据量过大，依然无法导出。
     • 导出异常大的群聊时可能涉及预先解析全部聊天记录的过程，这一过程无法顺利完成，导致导出失败。

4. 其他群聊的导出情况

   • 非异常大的群聊：可以单独勾选并成功导出。
   • 导出环境要求：需要在 原手机 上进行导出操作，不能依靠转移到新手机后再导出。

[html5syt]

2. 我想既然另一个 db 文件group_msg_fts.db里面有所有的文本记录，又有和nt_msg.db匹配的唯一消息id，我觉得对应起来解密会不会更快

要看消息类型，如果你的数据库里大多数都是文本消息那肯定更快，但是遇到了一堆聊天记录之类的就体现不出来效果了