ChatGPT聊天记录归档实战：基于Python的自动化解决方案

1次阅读

没有评论

共计 2438 个字符，预计需要花费 7 分钟才能阅读完成。

在日常使用 ChatGPT 时，我发现随着对话数量的增加，历史记录管理变得越来越困难。未归档的聊天记录会导致几个明显问题：

重要信息难以回溯：有价值的对话被淹没在海量记录中
知识无法沉淀：碎片化对话难以形成结构化知识库
资源浪费：重复提问相同问题导致 API 调用次数增加

在考虑存储方案时，我对比了两种主流方式：

直接存储 JSON 文件
优点：实现简单，无需额外依赖
缺点：查询效率低，难以做复杂分析
数据库存储
优点：支持高效查询，便于后续分析
缺点：需要额外维护数据库

综合考虑后，我选择了 SQLite 作为存储方案，因为：

零配置，单文件管理
支持标准 SQL 语法
适合中小规模数据量

OpenAI 提供了 Conversations API 来获取聊天历史。以下是核心代码示例：

import openai
from typing import List, Dict

def fetch_conversations(api_key: str, limit: int = 100) -> List[Dict]:
    """
    获取 ChatGPT 对话历史

    :param api_key: OpenAI API 密钥
    :param limit: 单次请求最大记录数
    :return: 对话记录列表
    """
    openai.api_key = api_key

    try:
        conversations = []
        after_id = None

        while True:
            params = {
                'limit': limit,
                'after': after_id
            }

            response = openai.Conversation.list(**params)
            conversations.extend(response['data'])

            if not response['has_more']:
                break

            after_id = response['data'][-1]['id']

        return conversations
    except Exception as e:
        print(f"获取对话历史出错: {str(e)}")
        return []

OpenAI API 有严格的速率限制，我们需要合理处理：

分页参数：使用 after 参数实现分页获取
速率限制：添加适当的延迟

import time

# 在循环中添加延迟
time.sleep(0.5)  # 每次请求间隔 0.5 秒

我设计了以下 SQLite 表结构来存储对话：

CREATE TABLE IF NOT EXISTS conversations (
    id TEXT PRIMARY KEY,
    created_at INTEGER,
    title TEXT,
    model TEXT,
    messages TEXT  -- JSON 格式存储对话内容
);

CREATE INDEX IF NOT EXISTS idx_created_at ON conversations(created_at);

为了避免每次都全量同步，我记录了最后同步时间：

def get_last_sync_time(db_conn) -> int:
    """获取上次同步时间戳"""
    cursor = db_conn.cursor()
    cursor.execute("SELECT MAX(created_at) FROM conversations")
    result = cursor.fetchone()
    return result[0] if result[0] else 0

使用 asyncio 提高 IO 密集型任务的效率：

import aiohttp
import asyncio

async def async_fetch_conversations(session, api_key, after=None):
    """异步获取对话历史"""
    url = "https://api.openai.com/v1/conversations"
    headers = {"Authorization": f"Bearer {api_key}"}
    params = {"limit": 100}

    if after:
        params["after"] = after

    async with session.get(url, headers=headers, params=params) as response:
        return await response.json()

实现 token 自动刷新机制：

def refresh_token(old_token):
    # 实现 token 刷新逻辑
    pass

在存储前过滤敏感信息：

import re

def sanitize_text(text):
    """过滤敏感信息"""
    # 移除邮箱
    text = re.sub(r'[\w\.-]+@[\w\.-]+\.\w+', '[REDACTED]', text)
    # 移除电话号码
    text = re.sub(r'\+?\d[\d-]{8,}\d', '[REDACTED]', text)
    return text

建议实现定期自动备份：

import shutil
import datetime

def backup_database(db_path):
    """数据库备份"""
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_path = f"{db_path}.bak_{timestamp}"
    shutil.copy2(db_path, backup_path)

未来可以考虑使用 NLP 技术实现语义检索：