ChatGPT导出实战指南：从API调用到数据处理的完整解决方案

1次阅读

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

在开发过程中，许多开发者在使用 ChatGPT API 导出对话记录时遇到了不少挑战。这些挑战不仅影响了开发效率，还可能导致数据丢失或格式混乱。以下是一些常见的痛点：

分页处理：ChatGPT API 返回的数据通常分页显示，需要开发者手动处理分页逻辑，增加了代码复杂度。
速率限制：API 调用有严格的速率限制，频繁请求可能导致被封禁或返回错误。
数据格式不一致：返回的数据可能包含 Markdown、HTML 或纯文本，需要统一格式化处理。
错误重试机制：网络波动或 API 临时故障可能导致请求失败，缺乏有效的重试机制会导致数据不完整。

为了简化 API 调用，我们可以封装一个 Python 类，处理认证、分页和错误重试。以下是一个完整的代码示例：

import requests
import time

class ChatGPTExporter:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.openai.com/v1/chat/completions"
        self.headers = {"Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def fetch_conversations(self, max_retries=3, delay=1):
        conversations = []
        params = {"limit": 100, "offset": 0}
        retries = 0

        while True:
            try:
                response = requests.get(
                    self.base_url,
                    headers=self.headers,
                    params=params
                )
                response.raise_for_status()
                data = response.json()
                conversations.extend(data["data"])

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

                params["offset"] += params["limit"]
                retries = 0

            except requests.exceptions.RequestException as e:
                if retries >= max_retries:
                    raise
                retries += 1
                time.sleep(delay * retries)

        return conversations

ChatGPT 返回的数据可能包含多种格式，我们需要将其统一为易于处理的格式。以下是一个处理 Markdown 和 HTML 的示例：

import markdown
from bs4 import BeautifulSoup

def clean_text(text, format_type="markdown"):
    if format_type == "markdown":
        html = markdown.markdown(text)
        soup = BeautifulSoup(html, "html.parser")
        return soup.get_text()
    elif format_type == "html":
        soup = BeautifulSoup(text, "html.parser")
        return soup.get_text()
    else:
        return text

根据不同的需求，我们可以选择不同的存储方案。以下是几种常见方案的对比：

JSON：适合结构化数据，易于读写，但文件体积较大。
CSV：适合表格数据，体积较小，但不支持嵌套结构。
数据库：适合大规模数据，支持复杂查询，但需要额外的维护成本。

以下是一个将数据保存为 JSON 和 CSV 的示例：

import json
import csv

def save_to_json(data, filename):
    with open(filename, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

def save_to_csv(data, filename):
    with open(filename, "w", encoding="utf-8", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["id", "content"])
        for item in data:
            writer.writerow([item["id"], item["content"]])

通过批量请求减少 API 调用次数，可以有效降低速率限制的影响。以下是一个批量处理的示例：

def fetch_batch_conversations(self, batch_size=10):
    conversations = []
    params = {"limit": batch_size, "offset": 0}

    while True:
        batch = self.fetch_conversations(params=params)
        conversations.extend(batch)

        if len(batch) < batch_size:
            break

        params["offset"] += batch_size

    return conversations

使用异步请求可以显著提高数据获取的效率。以下是使用 aiohttp 库的示例：

import aiohttp
import asyncio

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

async def fetch_all_conversations(api_key):
    async with aiohttp.ClientSession() as session:
        tasks = []
        for offset in range(0, 1000, 100):
            params = {"limit": 100, "offset": offset}
            task = fetch_async_conversations(
                session,
                "https://api.openai.com/v1/chat/completions",
                {"Authorization": f"Bearer {api_key}"},
                params
            )
            tasks.append(task)

        return await asyncio.gather(*tasks)

为了避免重复请求相同的数据，可以使用缓存策略。以下是使用 functools.lru_cache 的示例：

from functools import lru_cache

@lru_cache(maxsize=100)
def get_conversation(conversation_id):
    response = requests.get(f"https://api.openai.com/v1/chat/completions/{conversation_id}",
        headers={"Authorization": f"Bearer {self.api_key}"}
    )
    return response.json()

在使用 ChatGPT API 时，开发者可能会遇到以下常见问题：