"""查询历史服务 — 记录和管理用户查询历史。 支持: - 自动记录每次查询（内容、耗时、结果数、数据库类型） - 查询历史分页查看 - 收藏常用查询 - 快捷查询（个人/团队共享） - 查询统计（高频表、查询趋势） 人性化设计: - 自动记录查询历史，用户无需手动保存 - 支持收藏常用查询，一键复用 - 快捷查询模板，减少重复输入 - 查询统计帮助用户了解自己的数据使用习惯 """ from typing import List, Optional, Dict, Any from dataclasses import dataclass, field from datetime import datetime, timedelta from collections import Counter @dataclass class QueryHistoryEntry: """查询历史条目。""" id: int = 0 user_id: int = 0 username: str = "" instance_id: int = 0 instance_name: str = "" db_type: str = "" # mysql / postgres / redis / mongo database_name: str = "" query_content: str = "" # SQL / Redis命令 / MongoDB查询 query_type: str = "" # SELECT / GET / find 等 execution_time_ms: int = 0 result_rows: int = 0 is_success: bool = True error_message: Optional[str] = None is_favorite: bool = False note: str = "" # 用户备注 created_at: str = "" @dataclass class QuickQuery: """快捷查询模板。""" id: int = 0 name: str = "" # 查询名称，如"查看活跃用户数" description: str = "" # 描述 query_content: str = "" # 查询内容 db_type: str = "" # 适用的数据库类型 database_pattern: str = "" # 适用的数据库名称模式 is_shared: bool = False # 是否团队共享 created_by: int = 0 created_by_name: str = "" usage_count: int = 0 # 使用次数 created_at: str = "" class QueryHistoryService: """查询历史服务。 内存存储实现，生产环境应替换为数据库存储。 """ _history: List[QueryHistoryEntry] = [] _favorites: set = set() # (user_id, history_id) _quick_queries: List[QuickQuery] = [] _history_counter: int = 0 _quick_counter: int = 0 _max_history_per_user: int = 500 # 每个用户最多保留500条 @classmethod async def record_query( cls, user_id: int, username: str, instance_id: int, instance_name: str, db_type: str, database_name: str, query_content: str, query_type: str, execution_time_ms: int, result_rows: int, is_success: bool = True, error_message: Optional[str] = None, ) -> QueryHistoryEntry: """记录一次查询。 每次查询自动调用，无需用户干预。 """ cls._history_counter += 1 entry = QueryHistoryEntry( id=cls._history_counter, user_id=user_id, username=username, instance_id=instance_id, instance_name=instance_name, db_type=db_type, database_name=database_name, query_content=query_content[:2000], # 限制长度 query_type=query_type, execution_time_ms=execution_time_ms, result_rows=result_rows, is_success=is_success, error_message=error_message, created_at=datetime.now().isoformat(), ) cls._history.append(entry) # 限制每个用户的历史记录数量 user_entries = [h for h in cls._history if h.user_id == user_id] if len(user_entries) > cls._max_history_per_user: # 删除最旧的 to_remove = len(user_entries) - cls._max_history_per_user oldest_ids = {h.id for h in sorted(user_entries, key=lambda x: x.created_at)[:to_remove]} cls._history = [h for h in cls._history if h.id not in oldest_ids] return entry @classmethod async def get_history( cls, user_id: int, db_type: Optional[str] = None, instance_id: Optional[int] = None, keyword: Optional[str] = None, page: int = 1, page_size: int = 20, ) -> Dict[str, Any]: """获取用户查询历史。""" entries = [h for h in cls._history if h.user_id == user_id] if db_type: entries = [h for h in entries if h.db_type == db_type] if instance_id: entries = [h for h in entries if h.instance_id == instance_id] if keyword: kw = keyword.lower() entries = [h for h in entries if kw in h.query_content.lower()] total = len(entries) entries.sort(key=lambda x: x.created_at, reverse=True) start = (page - 1) * page_size end = start + page_size page_entries = entries[start:end] return { "total": total, "page": page, "page_size": page_size, "items": [ { "id": e.id, "instance_name": e.instance_name, "db_type": e.db_type, "database_name": e.database_name, "query_content": e.query_content[:200], "query_type": e.query_type, "execution_time_ms": e.execution_time_ms, "result_rows": e.result_rows, "is_success": e.is_success, "is_favorite": (user_id, e.id) in cls._favorites, "note": e.note, "created_at": e.created_at, } for e in page_entries ], } @classmethod async def toggle_favorite(cls, user_id: int, history_id: int) -> bool: """切换收藏状态。""" key = (user_id, history_id) if key in cls._favorites: cls._favorites.remove(key) return False # 已取消收藏 else: cls._favorites.add(key) return True # 已收藏 @classmethod async def get_favorites( cls, user_id: int, page: int = 1, page_size: int = 20 ) -> Dict[str, Any]: """获取收藏的查询。""" favorite_ids = {hid for uid, hid in cls._favorites if uid == user_id} entries = [h for h in cls._history if h.id in favorite_ids] entries.sort(key=lambda x: x.created_at, reverse=True) total = len(entries) start = (page - 1) * page_size end = start + page_size return { "total": total, "page": page, "page_size": page_size, "items": [ { "id": e.id, "instance_name": e.instance_name, "db_type": e.db_type, "database_name": e.database_name, "query_content": e.query_content[:200], "query_type": e.query_type, "execution_time_ms": e.execution_time_ms, "result_rows": e.result_rows, "created_at": e.created_at, } for e in entries[start:end] ], } # ── 快捷查询 ── @classmethod async def create_quick_query( cls, name: str, description: str, query_content: str, db_type: str, database_pattern: str, is_shared: bool, created_by: int, created_by_name: str, ) -> QuickQuery: """创建快捷查询。""" cls._quick_counter += 1 qq = QuickQuery( id=cls._quick_counter, name=name, description=description, query_content=query_content, db_type=db_type, database_pattern=database_pattern, is_shared=is_shared, created_by=created_by, created_by_name=created_by_name, created_at=datetime.now().isoformat(), ) cls._quick_queries.append(qq) return qq @classmethod async def get_quick_queries( cls, user_id: int, db_type: Optional[str] = None, keyword: Optional[str] = None, ) -> List[Dict[str, Any]]: """获取快捷查询列表（个人+团队共享）。""" result = [] for qq in cls._quick_queries: # 条件：共享的 或 自己创建的 if qq.is_shared or qq.created_by == user_id: if db_type and qq.db_type != db_type: continue if keyword and keyword.lower() not in qq.name.lower(): continue result.append({ "id": qq.id, "name": qq.name, "description": qq.description, "query_content": qq.query_content[:500], "db_type": qq.db_type, "is_shared": qq.is_shared, "created_by_name": qq.created_by_name, "usage_count": qq.usage_count, "created_at": qq.created_at, }) return result @classmethod async def use_quick_query(cls, quick_query_id: int) -> Optional[str]: """使用快捷查询，返回查询内容并增加使用次数。""" for qq in cls._quick_queries: if qq.id == quick_query_id: qq.usage_count += 1 return qq.query_content return None @classmethod async def get_query_statistics(cls, user_id: int) -> Dict[str, Any]: """获取用户查询统计。 人性化：帮助用户了解自己的数据使用习惯。 """ entries = [h for h in cls._history if h.user_id == user_id] if not entries: return {"message": "暂无查询记录"} # 按数据库类型分布 db_type_dist = Counter(h.db_type for h in entries) # 高频查询表（从SQL中提取表名） tables = [] for h in entries: if h.db_type in ("mysql", "postgres"): # 简单提取表名 import re found = re.findall(r"FROM\s+(\w+)", h.query_content, re.IGNORECASE) tables.extend(found) table_dist = Counter(tables).most_common(10) # 查询趋势（最近7天） now = datetime.now() daily_counts = {} for i in range(7): day = (now - timedelta(days=i)).strftime("%Y-%m-%d") count = sum( 1 for h in entries if h.created_at.startswith(day) ) daily_counts[day] = count # 平均执行时间 avg_time = sum(h.execution_time_ms for h in entries) / len(entries) # 总查询次数 total_queries = len(entries) return { "total_queries": total_queries, "avg_execution_time_ms": round(avg_time, 2), "db_type_distribution": dict(db_type_dist), "top_tables": [{"table": t, "count": c} for t, c in table_dist], "daily_trend": daily_counts, "success_rate": round( sum(1 for h in entries if h.is_success) / total_queries * 100, 2 ), } # ── 预置快捷查询模板 ── @classmethod def seed_default_quick_queries(cls): """预置默认快捷查询模板。""" defaults = [ QuickQuery( id=-1, name="查看所有表", description="列出数据库中所有表", query_content="SHOW TABLES", db_type="mysql", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), QuickQuery( id=-2, name="查看表结构", description="查看指定表的列信息", query_content="DESCRIBE {table_name}", db_type="mysql", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), QuickQuery( id=-3, name="查看慢查询", description="MySQL慢查询日志", query_content="SELECT * FROM mysql.slow_log ORDER BY start_time DESC LIMIT 20", db_type="mysql", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), QuickQuery( id=-4, name="查看活跃连接", description="当前数据库连接情况", query_content="SHOW PROCESSLIST", db_type="mysql", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), QuickQuery( id=-5, name="查看表大小", description="各表占用空间统计", query_content="""SELECT table_name AS `表名`, ROUND(data_length / 1024 / 1024, 2) AS `数据大小(MB)`, ROUND(index_length / 1024 / 1024, 2) AS `索引大小(MB)`, table_rows AS `行数` FROM information_schema.tables WHERE table_schema = DATABASE() ORDER BY data_length DESC LIMIT 50""", db_type="mysql", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), QuickQuery( id=-6, name="Redis查看所有键", description="SCAN遍历所有键", query_content="SCAN 0 MATCH * COUNT 50", db_type="redis", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), QuickQuery( id=-7, name="Redis查看服务器信息", description="Redis INFO命令", query_content="INFO server", db_type="redis", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), QuickQuery( id=-8, name="PG查看所有表", description="PostgreSQL列出所有表", query_content="SELECT schemaname, tablename, n_tup_ins, n_tup_upd, n_tup_del FROM pg_stat_user_tables ORDER BY n_live_tup DESC LIMIT 50", db_type="postgres", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), QuickQuery( id=-9, name="MongoDB查找文档", description="查找集合中的文档", query_content='{"collection": "{collection_name}", "operation": "find", "limit": 20}', db_type="mongo", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), QuickQuery( id=-10, name="MongoDB统计文档数", description="统计集合文档数量", query_content='{"collection": "{collection_name}", "operation": "countDocuments"}', db_type="mongo", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), ] cls._quick_queries.extend(defaults) cls._quick_counter = max(cls._quick_counter, 10) # 初始化预置快捷查询 QueryHistoryService.seed_default_quick_queries()