"""查询历史服务 — 记录和管理用户查询历史（SQLite 持久化）。 支持: - 自动记录每次查询（内容、耗时、结果数、数据库类型） - 查询历史分页查看 - 收藏常用查询 - 快捷查询（个人/团队共享） - 查询统计（高频表、查询趋势） 人性化设计: - 自动记录查询历史，用户无需手动保存 - 支持收藏常用查询，一键复用 - 快捷查询模板，减少重复输入 - 查询统计帮助用户了解自己的数据使用习惯 """ import re from typing import List, Optional, Dict, Any from datetime import datetime, timedelta from collections import Counter from sqlalchemy import select, func, and_, or_ from sqlalchemy.ext.asyncio import AsyncSession from app.models.query_history import QueryHistory, QuickQueryModel class QueryHistoryService: """查询历史服务（SQLite 持久化）。""" _max_history_per_user: int = 500 # 每个用户最多保留500条 # ── 查询历史 ── @staticmethod async def record_query( db: AsyncSession, 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, ) -> QueryHistory: """记录一次查询。 每次查询自动调用，无需用户干预。 """ entry = QueryHistory( 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, ) db.add(entry) await db.flush() # 限制每个用户的历史记录数量 count_stmt = select(func.count()).select_from(QueryHistory).where( QueryHistory.user_id == user_id ) total = (await db.execute(count_stmt)).scalar() if total > QueryHistoryService._max_history_per_user: # 删除最旧的条目 to_remove = total - QueryHistoryService._max_history_per_user old_stmt = ( select(QueryHistory.id) .where(QueryHistory.user_id == user_id) .order_by(QueryHistory.created_at.asc()) .limit(to_remove) ) old_result = await db.execute(old_stmt) old_ids = [row[0] for row in old_result.all()] if old_ids: delete_stmt = QueryHistory.__table__.delete().where( QueryHistory.id.in_(old_ids) ) await db.execute(delete_stmt) await db.flush() return entry @staticmethod async def get_history( db: AsyncSession, 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]: """获取用户查询历史。""" stmt = select(QueryHistory).where(QueryHistory.user_id == user_id) if db_type: stmt = stmt.where(QueryHistory.db_type == db_type) if instance_id: stmt = stmt.where(QueryHistory.instance_id == instance_id) if keyword: stmt = stmt.where(QueryHistory.query_content.ilike(f"%{keyword}%")) # 总数 total_stmt = select(func.count()).select_from(stmt.subquery()) total = (await db.execute(total_stmt)).scalar() # 排序 + 分页 stmt = stmt.order_by(QueryHistory.created_at.desc()) stmt = stmt.offset((page - 1) * page_size).limit(page_size) result = await db.execute(stmt) entries = result.scalars().all() 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 or "")[: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": e.is_favorite, "note": e.note, "created_at": e.created_at.isoformat() if e.created_at else "", } for e in entries ], } @staticmethod async def toggle_favorite(db: AsyncSession, user_id: int, history_id: int) -> bool: """切换收藏状态。""" stmt = select(QueryHistory).where( and_( QueryHistory.id == history_id, QueryHistory.user_id == user_id, ) ) result = await db.execute(stmt) entry = result.scalar_one_or_none() if entry is None: # 没有找到该条记录（可能不属于当前用户），直接返回 False return False entry.is_favorite = not entry.is_favorite await db.flush() return entry.is_favorite @staticmethod async def get_favorites( db: AsyncSession, user_id: int, page: int = 1, page_size: int = 20 ) -> Dict[str, Any]: """获取收藏的查询。""" stmt = select(QueryHistory).where( and_( QueryHistory.user_id == user_id, QueryHistory.is_favorite == True, # noqa: E712 ) ) # 总数 total_stmt = select(func.count()).select_from(stmt.subquery()) total = (await db.execute(total_stmt)).scalar() # 排序 + 分页 stmt = stmt.order_by(QueryHistory.created_at.desc()) stmt = stmt.offset((page - 1) * page_size).limit(page_size) result = await db.execute(stmt) entries = result.scalars().all() 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 or "")[:200], "query_type": e.query_type, "execution_time_ms": e.execution_time_ms, "result_rows": e.result_rows, "created_at": e.created_at.isoformat() if e.created_at else "", } for e in entries ], } # ── 快捷查询 ── @staticmethod async def create_quick_query( db: AsyncSession, name: str, description: str, query_content: str, db_type: str, database_pattern: str, is_shared: bool, created_by: int, created_by_name: str, ) -> QuickQueryModel: """创建快捷查询。""" qq = QuickQueryModel( 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, ) db.add(qq) await db.flush() return qq @staticmethod async def get_quick_queries( db: AsyncSession, user_id: int, db_type: Optional[str] = None, keyword: Optional[str] = None, ) -> List[Dict[str, Any]]: """获取快捷查询列表（个人+团队共享）。""" stmt = select(QuickQueryModel).where( or_( QuickQueryModel.is_shared == True, # noqa: E712 QuickQueryModel.created_by == user_id, ) ) if db_type: stmt = stmt.where(QuickQueryModel.db_type == db_type) if keyword: stmt = stmt.where(QuickQueryModel.name.ilike(f"%{keyword}%")) stmt = stmt.order_by(QuickQueryModel.created_at.desc()) result = await db.execute(stmt) queries = result.scalars().all() return [ { "id": qq.id, "name": qq.name, "description": qq.description, "query_content": (qq.query_content or "")[: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.isoformat() if qq.created_at else "", } for qq in queries ] @staticmethod async def use_quick_query(db: AsyncSession, quick_query_id: int) -> Optional[str]: """使用快捷查询，返回查询内容并增加使用次数。""" stmt = select(QuickQueryModel).where(QuickQueryModel.id == quick_query_id) result = await db.execute(stmt) qq = result.scalar_one_or_none() if qq is None: return None qq.usage_count += 1 await db.flush() return qq.query_content @staticmethod async def get_query_statistics(db: AsyncSession, user_id: int) -> Dict[str, Any]: """获取用户查询统计。 人性化：帮助用户了解自己的数据使用习惯。 """ # 总数 count_stmt = select(func.count()).select_from(QueryHistory).where( QueryHistory.user_id == user_id ) total_queries = (await db.execute(count_stmt)).scalar() if not total_queries: return {"message": "暂无查询记录"} # 按数据库类型分布 db_type_stmt = ( select(QueryHistory.db_type, func.count().label("cnt")) .where(QueryHistory.user_id == user_id) .group_by(QueryHistory.db_type) ) db_type_result = await db.execute(db_type_stmt) db_type_dist = {row[0]: row[1] for row in db_type_result.all()} # 高频查询表（从SQL中提取表名） sql_stmt = ( select(QueryHistory.query_content) .where( and_( QueryHistory.user_id == user_id, QueryHistory.db_type.in_(["mysql", "postgres"]), ) ) ) sql_result = await db.execute(sql_stmt) tables = [] for row in sql_result.all(): content = row[0] or "" found = re.findall(r"FROM\s+(\w+)", 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") day_start = datetime.strptime(day, "%Y-%m-%d") day_end = day_start + timedelta(days=1) day_count_stmt = select(func.count()).select_from(QueryHistory).where( and_( QueryHistory.user_id == user_id, QueryHistory.created_at >= day_start, QueryHistory.created_at < day_end, ) ) day_count = (await db.execute(day_count_stmt)).scalar() daily_counts[day] = day_count # 平均执行时间 avg_stmt = select(func.avg(QueryHistory.execution_time_ms)).where( QueryHistory.user_id == user_id ) avg_time = (await db.execute(avg_stmt)).scalar() or 0 # 成功率 success_stmt = select(func.count()).select_from(QueryHistory).where( and_( QueryHistory.user_id == user_id, QueryHistory.is_success == True, # noqa: E712 ) ) success_count = (await db.execute(success_stmt)).scalar() return { "total_queries": total_queries, "avg_execution_time_ms": round(float(avg_time), 2), "db_type_distribution": db_type_dist, "top_tables": [{"table": t, "count": c} for t, c in table_dist], "daily_trend": daily_counts, "success_rate": round(success_count / total_queries * 100, 2), } # ── 预置快捷查询模板 ── @staticmethod async def seed_default_quick_queries(db: AsyncSession): """预置默认快捷查询模板（仅在表为空时插入）。""" # 检查是否已有数据 count_stmt = select(func.count()).select_from(QuickQueryModel) count = (await db.execute(count_stmt)).scalar() if count and count > 0: return defaults = [ QuickQueryModel( name="查看所有表", description="列出数据库中所有表", query_content="SHOW TABLES", db_type="mysql", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), QuickQueryModel( name="查看表结构", description="查看指定表的列信息", query_content="DESCRIBE {table_name}", db_type="mysql", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), QuickQueryModel( 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="系统", ), QuickQueryModel( name="查看活跃连接", description="当前数据库连接情况", query_content="SHOW PROCESSLIST", db_type="mysql", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), QuickQueryModel( 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="系统", ), QuickQueryModel( 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="系统", ), QuickQueryModel( name="Redis查看服务器信息", description="Redis INFO命令", query_content="INFO server", db_type="redis", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), QuickQueryModel( 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="系统", ), QuickQueryModel( 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="系统", ), QuickQueryModel( name="MongoDB统计文档数", description="统计集合文档数量", query_content='{"collection": "{collection_name}", "operation": "countDocuments"}', db_type="mongo", database_pattern="*", is_shared=True, created_by=0, created_by_name="系统", ), ] db.add_all(defaults) await db.flush()