最近公司需要整理一些记录。为了方便给AI 提供资料。因故需要从导出的消息记录.TXT中提取对饮的消息对象的完整消息。
所以写了以下代码。分享给大家。
import os
import re
from collections import defaultdict
def list_available_contacts(file_path):
"""
列出消息记录文件中所有可用的联系人
Args:
file_path: 消息记录文件的路径
Returns:
list: 联系人名称列表
"""
contacts = []
contact_pattern = re.compile(r'^消息对象:(.+)$')
try:
with open(file_path, 'r', encoding='utf-8') as file:
for line in file:
line = line.strip()
contact_match = contact_pattern.match(line)
if contact_match:
contacts.append(contact_match.group(1))
except Exception as e:
print(f"读取文件时出错: {e}")
return contacts
def extract_messages(file_path):
"""
从QQ消息记录文件中提取所有消息对象的往来消息
Args:
file_path: 消息记录文件的路径
Returns:
dict: 以消息对象为键,消息列表为值的字典
"""
# 初始化变量
messages_by_contact = defaultdict(list)
current_contact = None
message_buffer = []
in_message_section = False
# 正则表达式模式 - 增强版本以处理特殊格式的联系人名称
contact_pattern = re.compile(r'^消息对象:(.+)$')
date_pattern = re.compile(r'^(\d{4}-\d{2}-\d{2} \d{1,2}:\d{2}:\d{2}) (.+)$')
try:
with open(file_path, 'r', encoding='utf-8') as file:
for line in file:
line = line.strip()
# 跳过空行
if not line:
continue
# 检查是否是消息对象行
contact_match = contact_pattern.match(line)
if contact_match:
# 保存之前的消息对象的消息
if current_contact and message_buffer:
messages_by_contact[current_contact].extend(message_buffer)
message_buffer = []
# 设置新的当前消息对象
current_contact = contact_match.group(1)
in_message_section = False
continue
# 检查是否是分隔线
if line.startswith('====='):
if line == '================================================================':
# 如果是分隔线,标记为可能进入消息部分
in_message_section = True
continue
# 如果已经确定了消息对象,开始收集消息
if current_contact and in_message_section:
# 检查是否是消息行
date_match = date_pattern.match(line)
if date_match:
# 这是一条新消息的开始
timestamp = date_match.group(1)
sender = date_match.group(2)
message_content = ""
message_buffer.append(f"{timestamp} {sender}")
else:
# 这是消息内容的一部分
if message_buffer: # 确保有消息头
message_buffer.append(line)
# 保存最后一个消息对象的消息
if current_contact and message_buffer:
messages_by_contact[current_contact].extend(message_buffer)
except Exception as e:
print(f"处理文件时出错: {e}")
return messages_by_contact
def save_messages_to_files(messages_by_contact, output_dir):
"""
将每个消息对象的消息保存到单独的文件中
Args:
messages_by_contact: 以消息对象为键,消息列表为值的字典
output_dir: 输出目录
"""
# 确保输出目录存在
os.makedirs(output_dir, exist_ok=True)
for contact, messages in messages_by_contact.items():
# 创建安全的文件名
safe_contact_name = "".join(c if c.isalnum() or c in [' ', '_', '-'] else '_' for c in contact)
file_path = os.path.join(output_dir, f"{safe_contact_name}.txt")
try:
with open(file_path, 'w', encoding='utf-8') as file:
file.write(f"消息对象: {contact}\n")
file.write("="*50 + "\n\n")
for message in messages:
file.write(f"{message}\n")
except Exception as e:
print(f"保存 {contact} 的消息时出错: {e}")
def main():
import argparse
# 创建命令行参数解析器
parser = argparse.ArgumentParser(description='从QQ消息记录中提取消息')
parser.add_argument('--input', '-i', default=r"e:\GIT\qq-msg\全部消息记录(1).txt",
help='输入文件路径')
parser.add_argument('--output', '-o', default=r"e:\GIT\qq-msg\extracted_messages",
help='输出目录路径')
parser.add_argument('--contact', '-c',
help='指定要提取的联系人名称,不指定则提取所有联系人')
parser.add_argument('--list', '-l', action='store_true',
help='列出所有可用的联系人')
# 解析命令行参数
args = parser.parse_args()
input_file = args.input
output_dir = args.output
specific_contact = args.contact
# 如果用户请求列出所有联系人
if args.list:
print(f"正在从 {input_file} 中获取所有联系人...")
contacts = list_available_contacts(input_file)
print(f"找到 {len(contacts)} 个联系人:")
for contact in sorted(contacts):
print(f" - {contact}")
return
print(f"开始从 {input_file} 提取消息...")
# 提取消息
messages_by_contact = extract_messages(input_file)
# 如果指定了特定联系人
if specific_contact:
if specific_contact in messages_by_contact:
# 只保留指定联系人的消息
messages_to_save = {specific_contact: messages_by_contact[specific_contact]}
print(f"成功找到并提取联系人 '{specific_contact}' 的消息。")
else:
print(f"未找到联系人 '{specific_contact}'。可用的联系人有:")
for contact in sorted(messages_by_contact.keys()):
print(f" - {contact}")
return
else:
messages_to_save = messages_by_contact
# 保存消息到文件
save_messages_to_files(messages_to_save, output_dir)
print(f"消息提取完成!共提取了 {len(messages_to_save)} 个联系人的消息。")
print(f"结果保存在: {output_dir}")
if __name__ == "__main__":
main()
然后用AI 写了一个用法。参考
QQ消息记录提取工具-参考方法-感谢deepseek
这是一个用于从QQ导出的消息记录文本文件中提取和整理消息的Python工具。该工具可以按照消息对象(联系人)分类提取所有往来消息,并将结果保存到单独的文件中。
功能特点
- 从QQ导出的消息记录文本文件中提取消息
- 按照消息对象(联系人)分类整理消息
- 支持提取所有联系人的消息或指定特定联系人的消息
- 提供命令行参数支持,方便灵活使用
- 保留消息的时间戳、发送者和内容信息
- 支持列出所有可用的联系人
使用方法
基本用法
# 提取所有联系人的消息
python main.py
# 提取特定联系人的消息
python main.py --contact "联系人名称"
# 列出所有可用的联系人
python main.py --list
# 指定输入文件和输出目录
python main.py --input "路径/到/消息记录.txt" --output "输出目录"
命令行参数
--input
,-i
: 指定输入文件路径(默认为e:\GIT\qq-msg\全部消息记录(1).txt
)--output
,-o
: 指定输出目录路径(默认为e:\GIT\qq-msg\extracted_messages
)--contact
,-c
: 指定要提取的联系人名称,不指定则提取所有联系人--list
,-l
: 列出所有可用的联系人
示例
- 提取"澳洲-预付货"的消息记录:
python main.py --contact "澳洲-预付货"
- 使用专用脚本提取"澳洲-预付货"的消息记录:
python extract_australia.py
- 列出所有可用的联系人:
python main.py --list
输出结果
- 每个联系人的消息将保存在一个单独的文本文件中
- 文件名为联系人名称(特殊字符会被替换为下划线)
- 文件内容包括所有与该联系人的往来消息,按时间顺序排列
文件结构
main.py
: 主程序文件,包含所有核心功能extract_australia.py
: 专门用于提取"澳洲-预付货"消息的辅助脚本extracted_messages/
: 默认输出目录,包含提取的消息文件
系统要求
- Python 3.6 或更高版本
- 无需额外的第三方库
注意事项
- 输入文件必须是QQ导出的消息记录文本文件
- 文件编码应为UTF-8,以确保正确处理中文字符
- 大型消息记录文件可能需要较长处理时间