ChatGPT对话导出实战：如何高效备份与分析对话数据

1次阅读

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

在通过 API 管理 ChatGPT 对话数据时，开发者常遇到三类典型问题：

API 分页限制 ：官方 API 返回结果存在分页机制，单次请求最多返回 100 条记录，需处理has_more 标识和分页令牌
流式响应解析：部分响应采用流式传输（chunked encoding），需要特殊处理未完整加载的数据包
合规性要求：GDPR 等法规要求对话数据可审计、可删除，需实现完整的数据生命周期管理

方案类型	优点	缺点
官方导出工具	无需开发，一键导出	仅支持基础格式，无增量导出功能
第三方 SDK	封装完善，支持高级功能	存在版本兼容风险
自定义 Python 脚本	完全可控，可定制存储和分析逻辑	需自行处理所有异常情况

import requests
def fetch_conversations(api_key, after=None):
    headers = {'Authorization': f'Bearer {api_key}'}
    params = {'limit': 100}
    if after:
        params['after'] = after

    try:
        response = requests.get(
            'https://api.openai.com/v1/conversations',
            headers=headers,
            params=params
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f'Request failed: {e}')
        return None

def process_stream_response(response):
    buffer = ''
    for chunk in response.iter_content(chunk_size=1024):
        if chunk:
            buffer += chunk.decode('utf-8')
            while '\n' in buffer:
                line, buffer = buffer.split('\n', 1)
                if line.strip():
                    yield json.loads(line)

关键实现逻辑：

每次导出后记录最后一条消息的 ID
下次请求时使用 after 参数从断点继续
使用本地 SQLite 数据库存储导出状态

import sqlite3
import time
from datetime import datetime

class ChatGPTExporter:
    def __init__(self, api_key):
        self.api_key = api_key
        self.conn = sqlite3.connect('chatgpt_export.db')
        self._init_db()

    def _init_db(self):
        self.conn.execute('''CREATE TABLE IF NOT EXISTS exports
                            (id TEXT PRIMARY KEY, created INTEGER, content TEXT)''')
        self.conn.execute('''CREATE TABLE IF NOT EXISTS checkpoint
                            (last_id TEXT)''')
        self.conn.commit()

    def export_to_csv(self, filepath):
        last_id = self._get_checkpoint()
        with open(filepath, 'a', encoding='utf-8') as f:
            while True:
                data = fetch_conversations(self.api_key, after=last_id)
                if not data or not data.get('data'):
                    break

                for conv in data['data']:
                    f.write(f'{conv["id"]},{conv["created"]},"{conv["content"]}"\n')
                    self._save_checkpoint(conv['id'])

                if not data.get('has_more'):
                    break

                time.sleep(0.5)  # 速率控制

    def _get_checkpoint(self):
        cursor = self.conn.execute('SELECT last_id FROM checkpoint LIMIT 1')
        return cursor.fetchone()[0] if cursor else None

    def _save_checkpoint(self, last_id):
        self.conn.execute('INSERT OR REPLACE INTO checkpoint VALUES (?)', (last_id,))
        self.conn.commit()