.4062073974432528:8691debe780be897f6dceb83e75a446a_69ec673fa3c75e7cbf955f49.69ec6d1fa3c75e7cbf956009.69ec6d1ff347046809cab3ce:Trae CN.T(2026/4/25 15:28:31)

config.yaml

@@ -1,5 +1,7 @@
 # 本地数据库路径
 db_path: ./DB/SQL.db
+# 历史记录和收藏数据库路径
+history_db_path: ./DB/history.db
 
 # 表结构信息
 TABLE: # 

history_db.py

@@ -0,0 +1,263 @@
+"""
+@Author: Liushu
+@Date: 2023/04/30
+历史记录和收藏功能的SQLite数据库模块
+"""
+import sqlite3
+import os
+import json
+from datetime import datetime
+from utility.utils import config_dict
+
+class HistoryDB:
+    def __init__(self, db_path=None):
+        if db_path is None:
+            db_path = config_dict.get('history_db_path', os.path.join(os.path.dirname(config_dict['db_path']), 'history.db'))
+        self.db_path = db_path
+        self._init_db()
+
+    def _get_connection(self):
+        return sqlite3.connect(self.db_path)
+
+    def _init_db(self):
+        conn = self._get_connection()
+        cursor = conn.cursor()
+
+        cursor.execute('''
+        CREATE TABLE IF NOT EXISTS query_history (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            natural_language TEXT NOT NULL,
+            generated_sql TEXT,
+            execution_time TEXT,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+            is_favorite BOOLEAN DEFAULT 0
+        )
+        ''')
+
+        cursor.execute('''
+        CREATE INDEX IF NOT EXISTS idx_created_at ON query_history(created_at)
+        ''')
+
+        cursor.execute('''
+        CREATE INDEX IF NOT EXISTS idx_is_favorite ON query_history(is_favorite)
+        ''')
+
+        conn.commit()
+        conn.close()
+
+    def add_history(self, natural_language, generated_sql=None, execution_time=None):
+        conn = self._get_connection()
+        cursor = conn.cursor()
+
+        cursor.execute('''
+        INSERT INTO query_history (natural_language, generated_sql, execution_time, created_at)
+        VALUES (?, ?, ?, ?)
+        ''', (natural_language, generated_sql, execution_time, datetime.now().strftime('%Y-%m-%d %H:%M:%S')))
+
+        history_id = cursor.lastrowid
+        conn.commit()
+        conn.close()
+
+        return history_id
+
+    def get_history(self, history_id):
+        conn = self._get_connection()
+        cursor = conn.cursor()
+
+        cursor.execute('''
+        SELECT id, natural_language, generated_sql, execution_time, created_at, is_favorite
+        FROM query_history
+        WHERE id = ?
+        ''', (history_id,))
+
+        result = cursor.fetchone()
+        conn.close()
+
+        if result:
+            return {
+                'id': result[0],
+                'natural_language': result[1],
+                'generated_sql': result[2],
+                'execution_time': result[3],
+                'created_at': result[4],
+                'is_favorite': bool(result[5])
+            }
+        return None
+
+    def get_all_history(self, limit=100, offset=0):
+        conn = self._get_connection()
+        cursor = conn.cursor()
+
+        cursor.execute('''
+        SELECT id, natural_language, generated_sql, execution_time, created_at, is_favorite
+        FROM query_history
+        ORDER BY created_at DESC
+        LIMIT ? OFFSET ?
+        ''', (limit, offset))
+
+        results = cursor.fetchall()
+        conn.close()
+
+        history_list = []
+        for result in results:
+            history_list.append({
+                'id': result[0],
+                'natural_language': result[1],
+                'generated_sql': result[2],
+                'execution_time': result[3],
+                'created_at': result[4],
+                'is_favorite': bool(result[5])
+            })
+
+        return history_list
+
+    def get_favorites(self, limit=100, offset=0):
+        conn = self._get_connection()
+        cursor = conn.cursor()
+
+        cursor.execute('''
+        SELECT id, natural_language, generated_sql, execution_time, created_at, is_favorite
+        FROM query_history
+        WHERE is_favorite = 1
+        ORDER BY created_at DESC
+        LIMIT ? OFFSET ?
+        ''', (limit, offset))
+
+        results = cursor.fetchall()
+        conn.close()
+
+        favorites_list = []
+        for result in results:
+            favorites_list.append({
+                'id': result[0],
+                'natural_language': result[1],
+                'generated_sql': result[2],
+                'execution_time': result[3],
+                'created_at': result[4],
+                'is_favorite': bool(result[5])
+            })
+
+        return favorites_list
+
+    def toggle_favorite(self, history_id):
+        conn = self._get_connection()
+        cursor = conn.cursor()
+
+        cursor.execute('''
+        SELECT is_favorite FROM query_history WHERE id = ?
+        ''', (history_id,))
+
+        result = cursor.fetchone()
+        if result:
+            new_favorite = 0 if result[0] else 1
+            cursor.execute('''
+            UPDATE query_history SET is_favorite = ? WHERE id = ?
+            ''', (new_favorite, history_id))
+            conn.commit()
+            conn.close()
+            return bool(new_favorite)
+
+        conn.close()
+        return None
+
+    def set_favorite(self, history_id, is_favorite):
+        conn = self._get_connection()
+        cursor = conn.cursor()
+
+        cursor.execute('''
+        UPDATE query_history SET is_favorite = ? WHERE id = ?
+        ''', (1 if is_favorite else 0, history_id))
+
+        rows_affected = cursor.rowcount
+        conn.commit()
+        conn.close()
+
+        return rows_affected > 0
+
+    def delete_history(self, history_id):
+        conn = self._get_connection()
+        cursor = conn.cursor()
+
+        cursor.execute('''
+        DELETE FROM query_history WHERE id = ?
+        ''', (history_id,))
+
+        rows_affected = cursor.rowcount
+        conn.commit()
+        conn.close()
+
+        return rows_affected > 0
+
+    def clear_all_history(self):
+        conn = self._get_connection()
+        cursor = conn.cursor()
+
+        cursor.execute('''
+        DELETE FROM query_history
+        ''')
+
+        rows_affected = cursor.rowcount
+        conn.commit()
+        conn.close()
+
+        return rows_affected
+
+    def update_history(self, history_id, generated_sql=None, execution_time=None):
+        conn = self._get_connection()
+        cursor = conn.cursor()
+
+        updates = []
+        params = []
+
+        if generated_sql is not None:
+            updates.append('generated_sql = ?')
+            params.append(generated_sql)
+
+        if execution_time is not None:
+            updates.append('execution_time = ?')
+            params.append(execution_time)
+
+        if updates:
+            params.append(history_id)
+            query = f'UPDATE query_history SET {", ".join(updates)} WHERE id = ?'
+            cursor.execute(query, params)
+
+            rows_affected = cursor.rowcount
+            conn.commit()
+            conn.close()
+
+            return rows_affected > 0
+
+        conn.close()
+        return False
+
+    def search_history(self, keyword, limit=50):
+        conn = self._get_connection()
+        cursor = conn.cursor()
+
+        cursor.execute('''
+        SELECT id, natural_language, generated_sql, execution_time, created_at, is_favorite
+        FROM query_history
+        WHERE natural_language LIKE ? OR generated_sql LIKE ?
+        ORDER BY created_at DESC
+        LIMIT ?
+        ''', (f'%{keyword}%', f'%{keyword}%', limit))
+
+        results = cursor.fetchall()
+        conn.close()
+
+        history_list = []
+        for result in results:
+            history_list.append({
+                'id': result[0],
+                'natural_language': result[1],
+                'generated_sql': result[2],
+                'execution_time': result[3],
+                'created_at': result[4],
+                'is_favorite': bool(result[5])
+            })
+
+        return history_list
+
+
+history_db = HistoryDB()