ChatGPT聊天记录导出实战：从API调用到数据持久化的完整解决方案

1次阅读

没有评论

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

在直接调用 ChatGPT API 保存聊天记录时，开发者常遇到几个典型问题：

数据分页问题：当对话历史较长时，API 返回的数据会被分页处理，需要手动管理游标（cursor）来获取完整记录。
速率限制：OpenAI API 有严格的请求速率限制，短时间内频繁调用会导致请求失败。
非结构化数据：API 返回的原始数据是嵌套的 JSON 结构，包含大量元数据，直接存储效率低下且难以查询。
上下文断裂：在多轮对话中，如果处理不当，可能导致对话上下文的逻辑断裂。

针对聊天记录的存储，常见的方案有：

JSON 文件：简单易用，适合小规模数据，但查询效率低，不支持复杂操作。
SQLite：轻量级数据库，支持 SQL 查询，适合中等规模数据，无需额外服务。
云存储（如 AWS S3、Google Cloud Storage）：适合大规模数据和高可用性需求，但成本较高。

对于大多数开发者，SQLite 是一个平衡的选择，因为它无需额外配置，且能提供足够的查询能力。

使用 aiohttp 库实现异步 API 调用，提高请求效率：

import aiohttp
import asyncio

async def fetch_chat_history(session, api_key, cursor=None):
    url = "https://api.openai.com/v1/chat/history"
    params = {"cursor": cursor} if cursor else {}
    headers = {"Authorization": f"Bearer {api_key}"}
    async with session.get(url, params=params, headers=headers) as response:
        if response.status == 200:
            return await response.json()
        else:
            raise Exception(f"API request failed: {response.status}")

ChatGPT 的对话通常是树形结构，递归解析可以完整保留上下文：

def parse_conversation(messages, parent_id=None):
    conversation = []
    for msg in messages:
        if msg["parent_id"] == parent_id:
            node = {"id": msg["id"],
                "role": msg["role"],
                "content": msg["content"],
                "timestamp": msg["timestamp"],
                "children": parse_conversation(messages, msg["id"])
            }
            conversation.append(node)
    return conversation

统一时间戳格式，便于后续处理：

from datetime import datetime

def standardize_timestamp(ts):
    return datetime.strptime(ts, "%Y-%m-%dT%H:%M:%S.%fZ").strftime("%Y-%m-%d %H:%M:%S")

import logging
from tenacity import retry, stop_after_attempt, wait_exponential

class ChatAPI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.session = aiohttp.ClientSession()

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
    async def get_history(self, cursor=None):
        try:
            return await fetch_chat_history(self.session, self.api_key, cursor)
        except Exception as e:
            logging.error(f"Failed to fetch history: {e}")
            raise

def export_to_markdown(conversation, filename):
    with open(filename, "w", encoding="utf-8") as f:
        for msg in conversation:
            f.write(f"**{msg['role']}**: {msg['content']}\n\n")

实现令牌桶算法控制请求速率：

import time

class RateLimiter:
    def __init__(self, rate, per):
        self.rate = rate
        self.per = per
        self.tokens = rate
        self.last_check = time.time()

    async def wait(self):
        now = time.time()
        elapsed = now - self.last_check
        self.last_check = now
        self.tokens += elapsed * (self.rate / self.per)
        if self.tokens > self.rate:
            self.tokens = self.rate
        if self.tokens < 1:
            delay = (1 - self.tokens) * (self.per / self.rate)
            await asyncio.sleep(delay)
        else:
            self.tokens -= 1

使用 SQLite 实现本地缓存，防止数据丢失：

import sqlite3

def init_db():
    conn = sqlite3.connect("chat_history.db")
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS messages (
            id TEXT PRIMARY KEY,
            role TEXT,
            content TEXT,
            timestamp TEXT,
            parent_id TEXT
        )
    """)
    conn.commit()
    conn.close()