logan
4c9427e546
- P0: 实例访问控制全线接入权限组 (829测试) - P1: permissions字段激活 + 操作级权限检查 (843测试) - 前端: 删除数据权限页面, 权限组合并为唯一入口 - 新增: CodeMirror编辑器 + 查询模板 + 慢查询分析 - 新增: MongoDB管理页面 + Dashboard增强 - 修复: Pydantic V2迁移 + 前端TS错误修复
343 lines
12 KiB
Python
343 lines
12 KiB
Python
"""SQL助手 — 人性化辅助功能。
|
||
|
||
提供便捷的数据操作辅助:
|
||
- SQL格式化: 统一SQL代码风格
|
||
- SQL语法高亮标记: 标记关键字、表名、字段名
|
||
- 表结构速查: 快速获取表列信息
|
||
- 智能补全建议: 基于表结构生成字段列表
|
||
- SQL Explain解析: 将执行计划转为人类可读格式
|
||
- 查询风险提示: 基于规则标记潜在风险
|
||
|
||
人性化设计原则:
|
||
- 减少重复操作
|
||
- 降低SQL编写门槛
|
||
- 提供即时反馈
|
||
- 统一操作体验
|
||
"""
|
||
import re
|
||
from typing import List, Dict, Optional, Any
|
||
from dataclasses import dataclass
|
||
|
||
|
||
# SQL关键字（用于语法高亮）
|
||
SQL_KEYWORDS = {
|
||
"SELECT", "FROM", "WHERE", "INSERT", "UPDATE", "DELETE", "CREATE",
|
||
"DROP", "ALTER", "TABLE", "INDEX", "JOIN", "LEFT", "RIGHT", "INNER",
|
||
"OUTER", "ON", "AND", "OR", "NOT", "NULL", "IS", "IN", "EXISTS",
|
||
"BETWEEN", "LIKE", "GROUP", "BY", "ORDER", "HAVING", "LIMIT", "OFFSET",
|
||
"UNION", "ALL", "DISTINCT", "AS", "ASC", "DESC", "COUNT", "SUM",
|
||
"AVG", "MAX", "MIN", "IF", "CASE", "WHEN", "THEN", "ELSE", "END",
|
||
"SHOW", "DESCRIBE", "EXPLAIN", "USE", "SET", "COMMIT", "ROLLBACK",
|
||
}
|
||
|
||
# 高风险操作标记
|
||
RISK_MARKERS = {
|
||
"DELETE": {"level": "high", "message": "删除操作不可逆，请确认WHERE条件正确"},
|
||
"DROP": {"level": "critical", "message": "DROP操作会永久删除对象，请谨慎操作"},
|
||
"TRUNCATE": {"level": "critical", "message": "TRUNCATE会清空全表且不可回滚"},
|
||
"UPDATE": {"level": "medium", "message": "UPDATE可能影响多行数据，建议先SELECT验证"},
|
||
"ALTER": {"level": "medium", "message": "ALTER TABLE可能锁表，建议在低峰期执行"},
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class SQLFormatResult:
|
||
"""SQL格式化结果。"""
|
||
formatted: str
|
||
original: str
|
||
is_valid: bool
|
||
error: Optional[str] = None
|
||
|
||
|
||
@dataclass
|
||
class SQLHighlightResult:
|
||
"""SQL语法高亮结果。"""
|
||
highlighted: str # HTML格式的语法高亮
|
||
keywords: List[str] # 提取的关键字
|
||
tables: List[str] # 提取的表名
|
||
columns: List[str] # 提取的列名
|
||
risks: List[Dict[str, str]] # 风险提示
|
||
|
||
|
||
@dataclass
|
||
class ExplainResult:
|
||
"""Explain解析结果。"""
|
||
is_simple: bool
|
||
has_index: bool
|
||
has_full_scan: bool
|
||
estimated_rows: int
|
||
warnings: List[str]
|
||
human_readable: str # 人类可读的执行计划描述
|
||
|
||
|
||
class SQLAssistant:
|
||
"""SQL助手 — 人性化辅助功能。"""
|
||
|
||
# ── SQL格式化 ──
|
||
|
||
@staticmethod
|
||
def format_sql(sql: str) -> SQLFormatResult:
|
||
"""格式化SQL语句。
|
||
|
||
- 关键字大写
|
||
- 合理换行和缩进
|
||
- 统一空格
|
||
"""
|
||
if not sql or not sql.strip():
|
||
return SQLFormatResult(formatted="", original=sql, is_valid=False, error="SQL为空")
|
||
|
||
try:
|
||
import sqlparse
|
||
formatted = sqlparse.format(
|
||
sql,
|
||
keyword_case="upper",
|
||
identifier_case="lower",
|
||
strip_comments=False,
|
||
reindent=True,
|
||
indent_width=4,
|
||
wrap_after=80,
|
||
)
|
||
return SQLFormatResult(formatted=formatted, original=sql, is_valid=True)
|
||
except ImportError:
|
||
# 如果没有sqlparse，使用简单格式化
|
||
return SQLAssistant._simple_format(sql)
|
||
except Exception as e:
|
||
return SQLFormatResult(formatted=sql, original=sql, is_valid=False, error=str(e))
|
||
|
||
@staticmethod
|
||
def _simple_format(sql: str) -> SQLFormatResult:
|
||
"""简单SQL格式化（不依赖sqlparse）。"""
|
||
# 去除多余空格
|
||
formatted = " ".join(sql.split())
|
||
|
||
# 关键字大写
|
||
for kw in SQL_KEYWORDS:
|
||
pattern = re.compile(r"\b" + kw + r"\b", re.IGNORECASE)
|
||
formatted = pattern.sub(kw, formatted)
|
||
|
||
# 在主要关键字前换行
|
||
line_breaks = [" SELECT ", " FROM ", " WHERE ", " GROUP BY ",
|
||
" ORDER BY ", " HAVING ", " LIMIT ", " JOIN ",
|
||
" LEFT JOIN ", " RIGHT JOIN ", " INNER JOIN ", " UNION "]
|
||
for br in line_breaks:
|
||
formatted = formatted.replace(br, "\n" + br.strip().ljust(8))
|
||
|
||
return SQLFormatResult(formatted=formatted, original=sql, is_valid=True)
|
||
|
||
# ── 语法高亮与风险分析 ──
|
||
|
||
@staticmethod
|
||
def highlight_sql(sql: str) -> SQLHighlightResult:
|
||
"""SQL语法高亮和风险分析。
|
||
|
||
返回HTML格式的语法高亮文本和风险提示。
|
||
"""
|
||
upper = sql.upper().strip()
|
||
|
||
# 提取关键字
|
||
words = re.findall(r"\b\w+\b", upper)
|
||
keywords = [w for w in words if w in SQL_KEYWORDS]
|
||
|
||
# 提取表名（简单规则）
|
||
tables = []
|
||
table_patterns = [
|
||
r"FROM\s+(\w+)",
|
||
r"JOIN\s+(\w+)",
|
||
r"INTO\s+(\w+)",
|
||
r"UPDATE\s+(\w+)",
|
||
r"TABLE\s+(\w+)",
|
||
]
|
||
for pattern in table_patterns:
|
||
matches = re.findall(pattern, upper, re.IGNORECASE)
|
||
tables.extend(matches)
|
||
tables = list(set(t.lower() for t in tables))
|
||
|
||
# 风险提示
|
||
risks = []
|
||
for op, info in RISK_MARKERS.items():
|
||
if re.search(r"\b" + op + r"\b", upper):
|
||
# 检查是否有WHERE
|
||
if op in ("UPDATE", "DELETE") and "WHERE" not in upper:
|
||
risks.append({
|
||
"level": "critical",
|
||
"operation": op,
|
||
"message": f"{op} 缺少 WHERE 条件，将操作全表数据！",
|
||
})
|
||
else:
|
||
risks.append({"level": info["level"], "operation": op, "message": info["message"]})
|
||
|
||
# 构建HTML高亮（简化版）
|
||
highlighted = sql
|
||
for kw in SQL_KEYWORDS:
|
||
highlighted = re.sub(
|
||
r"\b" + kw + r"\b",
|
||
f'<span class="sql-keyword">{kw}</span>',
|
||
highlighted,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
|
||
return SQLHighlightResult(
|
||
highlighted=highlighted,
|
||
keywords=list(set(keywords)),
|
||
tables=tables,
|
||
columns=[], # 简单实现不提取列
|
||
risks=risks,
|
||
)
|
||
|
||
# ── Explain解析 ──
|
||
|
||
@staticmethod
|
||
def parse_explain(explain_rows: List[Dict[str, Any]]) -> ExplainResult:
|
||
"""解析MySQL EXPLAIN结果为人类可读格式。
|
||
|
||
Args:
|
||
explain_rows: EXPLAIN输出结果行列表
|
||
"""
|
||
if not explain_rows:
|
||
return ExplainResult(
|
||
is_simple=True, has_index=False, has_full_scan=False,
|
||
estimated_rows=0, warnings=["EXPLAIN结果为空"],
|
||
human_readable="无法解析执行计划",
|
||
)
|
||
|
||
warnings = []
|
||
total_rows = 0
|
||
has_full_scan = False
|
||
has_index = False
|
||
parts = []
|
||
|
||
for i, row in enumerate(explain_rows):
|
||
table = row.get("table") or row.get("TABLE", "unknown")
|
||
select_type = row.get("select_type") or row.get("SELECT_TYPE", "SIMPLE")
|
||
scan_type = row.get("type") or row.get("TYPE", "ALL")
|
||
possible_keys = row.get("possible_keys") or row.get("POSSIBLE_KEYS")
|
||
key_used = row.get("key") or row.get("KEY")
|
||
rows_est = row.get("rows") or row.get("ROWS", 0) or row.get("EST_ROWS", 0)
|
||
extra = row.get("Extra") or row.get("EXTRA", "")
|
||
|
||
total_rows += int(rows_est) if rows_est else 0
|
||
|
||
if scan_type in ("ALL", "index") and select_type == "SIMPLE":
|
||
has_full_scan = True
|
||
warnings.append(f"表 '{table}' 使用了全表扫描（{scan_type}），建议添加索引")
|
||
|
||
if key_used:
|
||
has_index = True
|
||
parts.append(f"表 '{table}': 使用了索引 {key_used}，预估扫描 {rows_est} 行")
|
||
else:
|
||
parts.append(f"表 '{table}': 未使用索引（类型: {scan_type}），预估扫描 {rows_est} 行")
|
||
|
||
if "Using filesort" in str(extra):
|
||
warnings.append(f"表 '{table}' 使用了文件排序，建议优化ORDER BY")
|
||
if "Using temporary" in str(extra):
|
||
warnings.append(f"表 '{table}' 使用了临时表，建议优化GROUP BY")
|
||
|
||
human_readable = "\n".join(parts)
|
||
|
||
if total_rows > 100000:
|
||
warnings.append(f"预估扫描总行数 {total_rows} 较大，查询可能较慢")
|
||
|
||
return ExplainResult(
|
||
is_simple=len(explain_rows) == 1,
|
||
has_index=has_index,
|
||
has_full_scan=has_full_scan,
|
||
estimated_rows=total_rows,
|
||
warnings=warnings,
|
||
human_readable=human_readable,
|
||
)
|
||
|
||
# ── 查询建议 ──
|
||
|
||
@staticmethod
|
||
def get_query_suggestions(sql: str, db_type: str = "mysql") -> List[Dict[str, str]]:
|
||
"""基于SQL内容提供人性化建议。
|
||
|
||
帮助用户写出更好的SQL。
|
||
"""
|
||
suggestions = []
|
||
upper = sql.upper().strip()
|
||
|
||
# SELECT * 建议
|
||
if re.search(r"SELECT\s+\*", upper):
|
||
suggestions.append({
|
||
"type": "optimization",
|
||
"message": "使用 SELECT * 会返回所有列，建议只选择需要的字段以减少网络传输",
|
||
})
|
||
|
||
# 缺少LIMIT
|
||
if "SELECT" in upper and "LIMIT" not in upper and db_type in ("mysql", "postgres"):
|
||
suggestions.append({
|
||
"type": "safety",
|
||
"message": "建议添加 LIMIT 子句限制返回行数，避免大数据量查询",
|
||
})
|
||
|
||
# ORDER BY RAND()
|
||
if "ORDER BY RAND()" in upper or "ORDER BY RANDOM()" in upper:
|
||
suggestions.append({
|
||
"type": "optimization",
|
||
"message": "ORDER BY RAND() 在大表上性能极差，建议使用其他随机化方案",
|
||
})
|
||
|
||
# LIKE %xxx%
|
||
if re.search(r"LIKE\s+'%.", upper):
|
||
suggestions.append({
|
||
"type": "optimization",
|
||
"message": "LIKE '%xxx%' 无法使用索引，如数据量大请考虑全文索引或其他方案",
|
||
})
|
||
|
||
# 多表JOIN
|
||
join_count = len(re.findall(r"\bJOIN\b", upper))
|
||
if join_count > 3:
|
||
suggestions.append({
|
||
"type": "optimization",
|
||
"message": f"当前有 {join_count} 个 JOIN，建议考虑子查询或视图简化",
|
||
})
|
||
|
||
# 没有WHERE的UPDATE/DELETE（前面已标记为critical）
|
||
# 这里给出更详细的建议
|
||
if re.search(r"\bUPDATE\b", upper) and "WHERE" not in upper:
|
||
suggestions.append({
|
||
"type": "danger",
|
||
"message": "UPDATE 缺少 WHERE 将修改全表！建议先用 SELECT 验证影响范围",
|
||
})
|
||
|
||
if re.search(r"\bDELETE\b", upper) and "WHERE" not in upper:
|
||
suggestions.append({
|
||
"type": "danger",
|
||
"message": "DELETE 缺少 WHERE 将删除全表！建议先用 SELECT 验证影响范围",
|
||
})
|
||
|
||
# 缺少注释的复杂查询
|
||
if len(sql) > 500 and "--" not in sql and "/*" not in sql:
|
||
suggestions.append({
|
||
"type": "style",
|
||
"message": "复杂查询建议添加注释，方便后续维护和理解",
|
||
})
|
||
|
||
return suggestions
|
||
|
||
# ── 统一辅助入口 ──
|
||
|
||
@staticmethod
|
||
def analyze(sql: str, db_type: str = "mysql") -> Dict[str, Any]:
|
||
"""全面分析SQL，返回格式化+高亮+风险+建议的完整结果。
|
||
|
||
人性化入口，一次调用获取所有辅助信息。
|
||
"""
|
||
format_result = SQLAssistant.format_sql(sql)
|
||
highlight_result = SQLAssistant.highlight_sql(sql)
|
||
suggestions = SQLAssistant.get_query_suggestions(sql, db_type)
|
||
|
||
return {
|
||
"formatted": format_result.formatted if format_result.is_valid else sql,
|
||
"is_valid": format_result.is_valid,
|
||
"format_error": format_result.error,
|
||
"highlighted": highlight_result.highlighted,
|
||
"keywords": highlight_result.keywords,
|
||
"tables": highlight_result.tables,
|
||
"risks": highlight_result.risks,
|
||
"suggestions": suggestions,
|
||
"risk_count": len(highlight_result.risks),
|
||
"suggestion_count": len(suggestions),
|
||
}
|