Python Tkinter 集成 DocsGPT 本地 API 开发实践

Python Tkinter 集成 DocsGPT 本地 API 开发实践

本文介绍如何使用 Python 和 Tkinter 构建一个集成代码编辑器与 GPT 助手的本地 IDE，支持调用本地 DocsGPT API 及 OpenAI 官方 API。

项目结构

gpt_code_ide/
├── main.py          # 主程序入口
├── database.py      # SQLite 数据库操作模块
├── gpt_client.py    # GPT API 客户端封装
└── requirements.txt # 依赖包列表

1. 依赖配置 (requirements.txt)

tkinter
sqlite3
requests
pygments
idlelib

2. 数据库模块 (database.py)

负责管理 SQLite 数据库，存储提问历史、代码片段及用户设置。

import sqlite3
import json
from datetime import datetime
from typing import List, Dict, Optional

class PyAIDatabase:
    def __init__(self, db_name: str = "pyai.db"):
        self.db_name = db_name
        self.init_database()

    def init_database(self):
        with sqlite3.connect(self.db_name) as conn:
            cursor = conn.cursor()
            # 创建历史记录表
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS query_history (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    question TEXT NOT NULL,
                    answer TEXT NOT NULL,
                    code_context TEXT,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)
            # 创建代码片段表
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS code_snippets (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    title TEXT NOT NULL,
                    code TEXT NOT NULL,
                    language TEXT DEFAULT 'python',
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)
            # 创建配置表
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS settings (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )
            """)
            # 插入默认配置
            cursor.execute("""
                INSERT OR IGNORE INTO settings (key, value) 
                VALUES ('api_key', ''), ('api_url', 'https://api.openai.com/v1/chat/completions'), ('theme', 'light'), ('font_size', '12')
            """)
            conn.commit()

    def save_query(self, question: str, answer: str, code_context: Optional[str] = None) -> int:
        with sqlite3.connect(self.db_name) as conn:
            cursor = conn.cursor()
            cursor.execute("""INSERT INTO query_history (question, answer, code_context) VALUES (?, ?, ?)""", (question, answer, code_context))
            conn.commit()
            return cursor.lastrowid

    def get_query_history(self, limit: int = 50, offset: int = 0) -> List[Dict]:
        with sqlite3.connect(self.db_name) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute("""SELECT id, question, answer, code_context, created_at, updated_at FROM query_history ORDER BY created_at DESC LIMIT ? OFFSET ?""", (limit, offset))
            rows = cursor.fetchall()
            return [dict(row) for row in rows]

    def delete_query(self, query_id: int) -> bool:
        with sqlite3.connect(self.db_name) as conn:
            cursor = conn.cursor()
            cursor.execute("DELETE FROM query_history WHERE id = ?", (query_id,))
            conn.commit()
            return cursor.rowcount > 0

    def clear_history(self) -> bool:
        with sqlite3.connect(self.db_name) as conn:
            cursor = conn.cursor()
            cursor.execute("DELETE FROM query_history")
            conn.commit()
            return cursor.rowcount > 0

    def search_history(self, keyword: str) -> List[Dict]:
        with sqlite3.connect(self.db_name) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute("""SELECT id, question, answer, created_at FROM query_history WHERE question LIKE ? OR answer LIKE ? ORDER BY created_at DESC""", (f"%{keyword}%", f"%{keyword}%"))
            return [dict(row) for row in cursor.fetchall()]

# 全局数据库实例
db = PyAIDatabase()

3. GPT API 客户端 (gpt_client.py)

支持 OpenAI 和 DocsGPT 两种 API 类型，处理同步/异步调用及流式响应。

import requests
import json
from typing import Optional, Dict, Any
import threading
from database import db

class GPTClient:
    def __init__(self, api_key: str = None, api_url: str = None):
        self.api_key = api_key or db.get_setting('api_key') or ""
        self.api_url = api_url or db.get_setting('api_url') or "https://api.openai.com/v1/chat/completions"
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        })

    def set_api_key(self, api_key: str):
        self.api_key = api_key
        self.session.headers.update({"Authorization": f"Bearer {api_key}"})
        db.update_setting('api_key', api_key)

    def set_api_url(self, api_url: str):
        self.api_url = api_url
        db.update_setting('api_url', api_url)

    def ask_question(self, question: str, code_context: Optional[str] = None, model: str = "gpt-3.5-turbo", max_tokens: int = 1000, temperature: float = 0.7, callback=None) -> Optional[str]:
        if not self.api_key:
            return "错误：请先设置 API 密钥"
        
        messages = []
        if code_context:
            messages.append({
                "role": "system",
                "content": f"以下是相关代码上下文：\n```python\n{code_context}\n```\n请基于此代码上下文回答问题。"
            })
        messages.append({"role": "user", "content": question})

        data = {
            "model": model,
            "messages": messages,
            "max_tokens": max_tokens,
            "temperature": temperature,
            "stream": False
        }

        try:
            if callback:
                thread = threading.Thread(target=self._async_ask, args=(data, callback))
                thread.daemon = True
                thread.start()
                return "正在处理..."
            else:
                return self._sync_ask(data)
        except Exception as e:
            return f"请求错误：{str(e)}"

    def _async_ask(self, data: Dict, callback):
        try:
            response = self.session.post(self.api_url, json=data, timeout=30)
            if response.status_code == 200:
                result = response.json()
                answer = result['choices'][0]['message']['content']
                callback(answer)
            else:
                error_msg = f"API 错误：{response.status_code}\n{response.text}"
                callback(error_msg)
        except requests.exceptions.Timeout:
            callback("错误：请求超时，请检查网络连接")
        except requests.exceptions.ConnectionError:
            callback("错误：无法连接到 API 服务器")
        except Exception as e:
            callback(f"未知错误：{str(e)}")

    def _sync_ask(self, data: Dict) -> str:
        try:
            response = self.session.post(self.api_url, json=data, timeout=30)
            if response.status_code == 200:
                result = response.json()
                return result['choices'][0]['message']['content']
            else:
                return f"API 错误：{response.status_code}\n{response.text}"
        except requests.exceptions.Timeout:
            return "错误：请求超时，请检查网络连接"
        except requests.exceptions.ConnectionError:
            return "错误：无法连接到 API 服务器"
        except Exception as e:
            return f"未知错误：{str(e)}"

gpt_client = GPTClient()

4. 主 GUI 程序 (main.py)

使用 Tkinter 构建左右分栏界面，左侧为代码编辑器，右侧为 GPT 助手。

(注：此处省略大量 GUI 布局代码，核心逻辑包括主题切换、文件操作、代码运行及问答交互)

主要功能模块：

• CodeEditor: 代码编辑区，支持语法高亮、文件保存、查找替换。
• GPTAssistant: 问答区，支持 API 配置、问题输入、答案显示及历史记录查询。
• PythonIDE: 主窗口，整合菜单、工具栏及事件绑定。

使用说明

步骤	操作
安装依赖	pip install requests autopep8
启动服务	在 DocsGPT 目录中运行 python app.py 或 npm start
配置 API	在程序右侧输入框填写 API 密钥，选择 API 类型为 docsgpt 或 openai
测试连接	点击'测试连接'验证本地 DocsGPT 服务 (http://localhost:3001/api/answer)
运行代码	编写 Python 代码后点击'运行代码'按钮执行
提问助手	在右侧输入问题，勾选'包含代码上下文'后获取答案

功能特点

• 双 API 支持: 兼容 OpenAI 官方 API 和本地 DocsGPT API。
• 历史记录: 基于 SQLite 持久化存储问答记录，支持搜索。
• 界面优化: 支持浅色/深色主题切换，海蓝色主题按钮。
• 代码执行: 内置临时文件机制安全运行 Python 代码并捕获输出。
• 兼容性: 自动适应不同的响应格式，提供详细的错误信息。