如何实现Cherry Studio聊天窗口内容的上下文导出：技术方案与避坑指南

1次阅读

没有评论

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

在开发基于 Cherry Studio 的聊天应用时，导出包含完整上下文的聊天记录是一个常见但复杂的任务。业务上通常需要这些数据用于审计、分析或备份。技术上的主要挑战包括：

消息碎片化：长时间对话可能分布在多个 API 响应中
上下文丢失：直接按时间排序会导致对话逻辑断裂
性能瓶颈：海量历史数据的导出可能超时

常见的技术方案各有优劣：

定期快照
优点：恢复点明确，操作简单
缺点：无法获取快照间的增量内容
增量同步
优点：资源消耗小
缺点：需要维护复杂的状态同步机制
全量导出
优点：数据完整
缺点：对服务器压力大

推荐采用分页全量导出 + 本地缓存的混合方案，平衡完整性与性能。

import requests
from datetime import datetime, timedelta

def fetch_messages(session_id, start_time, page_size=100):
    headers = {'Authorization': 'Bearer YOUR_API_TOKEN'}
    params = {
        'session_id': session_id,
        'start_time': start_time.isoformat(),
        'limit': page_size
    }

    try:
        response = requests.get(
            'https://api.cherrystudio.com/v1/messages',
            headers=headers,
            params=params,
            timeout=30
        )
        response.raise_for_status()
        return response.json()['messages']
    except requests.exceptions.RequestException as e:
        print(f"API 调用失败: {e}")
        return None

def reconstruct_threads(messages):
    """
    时间复杂度：O(n)
    空间复杂度：O(n)
    """
    threads = {}
    for msg in sorted(messages, key=lambda x: x['timestamp']):
        # 处理回复关系
        if msg.get('reply_to'):
            parent = threads.get(msg['reply_to'])
            if parent:
                parent.setdefault('replies', []).append(msg)
                continue

        # 新建会话线程
        threads[msg['id']] = msg

    return list(threads.values())

{
  "version": "1.0",
  "export_time": "2023-08-20T14:30:00Z",
  "threads": [
    {
      "root_message": {
        "id": "msg_001",
        "timestamp": "2023-08-20T09:00:00Z",
        "sender": "user1",
        "content": "你好，有个问题想咨询"
      },
      "replies": [
        {
          "id": "msg_002",
          "timestamp": "2023-08-20T09:02:00Z",
          "sender": "user2",
          "content": "请说"
        }
      ]
    }
  ]
}

采用基于时间的游标分页
每页默认获取 100 条消息
记录最后一条消息的时间戳作为下一页起点

使用 SQLite 存储已获取的消息
建立 (session_id, message_id) 的复合索引
定时清理超过 30 天的缓存

class ExportResume:
    def __init__(self, session_id):
        self.checkpoint_file = f"./{session_id}_checkpoint.json"

    def save_checkpoint(self, last_timestamp):
        with open(self.checkpoint_file, 'w') as f:
            json.dump({'last_timestamp': last_timestamp}, f)

    def load_checkpoint(self):
        try:
            with open(self.checkpoint_file) as f:
                return datetime.fromisoformat(json.load(f)['last_timestamp'])
        except FileNotFoundError:
            return None

每次 API 调用前验证 access_token
实现细粒度的 RBAC 控制
记录所有导出操作的审计日志

def sanitize_content(content):
    patterns = [(r'\b\d{16}\b', 'CREDIT_CARD'),  # 信用卡号
        (r'\b\d{3}-\d{2}-\d{4}\b', 'SSN')  # 社保号
    ]

    for pattern, replacement in patterns:
        content = re.sub(pattern, replacement, content)
    return content