sync: 同步 sql-platform-pro 最新代码 — 权限体系安全修复 v0.8.1

- P0: 实例访问控制全线接入权限组 (829测试)
- P1: permissions字段激活 + 操作级权限检查 (843测试)
- 前端: 删除数据权限页面, 权限组合并为唯一入口
- 新增: CodeMirror编辑器 + 查询模板 + 慢查询分析
- 新增: MongoDB管理页面 + Dashboard增强
- 修复: Pydantic V2迁移 + 前端TS错误修复

backend/app/api/v1/endpoints/audit_rule.py

@@ -11,6 +11,7 @@ from app.models.user import User
 from app.models.audit_rule import AuditRule
 from app.schemas.audit_rule import AuditRuleCreate, AuditRuleUpdate, AuditRuleResponse
 from app.services.audit_rule_seeds import SEED_RULES_MAP
+from app.services.audit_rule_seeds_v2 import seed_all_rules
 
 router = APIRouter(prefix="/audit-rules", tags=["审核规则"])
 
@@ -171,3 +172,17 @@ async def toggle_audit_rule(
     await db.commit()
     await db.refresh(rule)
     return rule
+
+
+@router.post("/seed-v2", response_model=List[AuditRuleResponse])
+async def seed_v2_audit_rules(
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin")),
+):
+    """批量初始化 SQL 审核引擎 V2 的所有规则种子数据（40+ 条规则）。
+
+    涵盖 DDL/DML/DQL/命名规范/性能/安全等多个维度，
+    跳过数据库中已存在的规则（按名称去重）。
+    """
+    created = await seed_all_rules(db)
+    return created

backend/app/api/v1/endpoints/impact.py

@@ -0,0 +1,33 @@
+"""影响分析 API。"""
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import get_db
+from app.core.security import get_current_user
+from app.models.user import User
+from app.models.instance import Instance
+from app.schemas.impact import ImpactAnalysisRequest, ImpactAnalysisResponse
+from app.services.impact_analyzer import ImpactAnalyzer
+
+router = APIRouter(prefix="/impact", tags=["影响分析"])
+
+
+@router.post("/analyze", response_model=ImpactAnalysisResponse)
+async def analyze_impact(
+    data: ImpactAnalysisRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """分析 SQL 执行影响。
+
+    预估影响行数、执行时间、风险等级。
+    """
+    result = await db.execute(select(Instance).where(Instance.id == data.instance_id))
+    instance = result.scalar_one_or_none()
+    if not instance:
+        raise HTTPException(status_code=404, detail="数据源不存在")
+
+    analyzer = ImpactAnalyzer()
+    analysis = await analyzer.analyze(instance, data.database, data.sql_content)
+    return analysis.to_dict()

backend/app/api/v1/endpoints/instance.py

@@ -7,8 +7,10 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.database import get_db
 from app.core.security import get_current_user, require_role
+from app.core.permissions import check_instance_access, get_user_accessible_group_ids
 from app.models.user import User
 from app.models.instance import Instance
+from app.models.instance_group import user_groups
 from app.schemas.instance import InstanceCreate, InstanceUpdate, InstanceResponse, ConnectionTestResult
 from app.services.instance_service import InstanceService
 from app.core.crypto import encrypt_password, decrypt_password
@@ -52,6 +54,21 @@ async def list_instances(
         stmt = stmt.where(Instance.db_type == db_type)
     if env:
         stmt = stmt.where(Instance.env == env)
+    
+    # P0: 权限组过滤 — 非 admin 只能看到自己权限组内的 + 公开实例
+    if current_user.role != "admin":
+        result = await db.execute(
+            select(user_groups.c.group_id).where(user_groups.c.user_id == current_user.id)
+        )
+        user_group_ids = [row[0] for row in result.all()]
+        if user_group_ids:
+            stmt = stmt.where(
+                (Instance.group_id.in_(user_group_ids)) | (Instance.group_id.is_(None))
+            )
+        else:
+            # 没有任何权限组，只能看公开实例
+            stmt = stmt.where(Instance.group_id.is_(None))
+    
     result = await db.execute(stmt)
     instances = result.scalars().all()
     return [_serialize_json_fields(i) for i in instances]
@@ -114,6 +131,8 @@ async def get_instance(
     instance = result.scalar_one_or_none()
     if not instance:
         raise HTTPException(status_code=404, detail="数据源不存在")
+    # P0: 检查访问权限
+    await check_instance_access(current_user, instance, db)
     return _serialize_json_fields(instance)
 
 
@@ -194,6 +213,9 @@ async def list_databases(
     if not instance:
         raise HTTPException(status_code=404, detail="数据源不存在")
 
+    # P0: 检查访问权限
+    await check_instance_access(current_user, instance, db)
+
     return await InstanceService.list_databases(instance)
 
 
@@ -209,4 +231,7 @@ async def list_tables(
     if not instance:
         raise HTTPException(status_code=404, detail="数据源不存在")
 
+    # P0: 检查访问权限
+    await check_instance_access(current_user, instance, db)
+
     return await InstanceService.list_tables(instance, database)

backend/app/api/v1/endpoints/instance_detail.py

@@ -8,6 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.database import get_db
 from app.core.security import require_role
+from app.core.permissions import check_instance_access
 from app.models.user import User
 from app.models.instance import Instance
 from app.core.crypto import decrypt_password
@@ -70,6 +71,8 @@ async def list_databases(
     current_user: User = Depends(require_role("admin", "dba")),
 ):
     instance = await _get_decrypted_instance(db, instance_id)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
     if instance.db_type != "mysql":
         raise HTTPException(status_code=400, detail="仅支持MySQL实例")
     try:
@@ -93,6 +96,8 @@ async def list_tables(
     current_user: User = Depends(require_role("admin", "dba")),
 ):
     instance = await _get_decrypted_instance(db, instance_id)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
     if instance.db_type != "mysql":
         raise HTTPException(status_code=400, detail="仅支持MySQL实例")
     try:
@@ -126,6 +131,8 @@ async def table_detail(
     current_user: User = Depends(require_role("admin", "dba")),
 ):
     instance = await _get_decrypted_instance(db, instance_id)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
     if instance.db_type != "mysql":
         raise HTTPException(status_code=400, detail="仅支持MySQL实例")
     try:
@@ -169,6 +176,8 @@ async def list_users(
     current_user: User = Depends(require_role("admin", "dba")),
 ):
     instance = await _get_decrypted_instance(db, instance_id)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
     if instance.db_type != "mysql":
         raise HTTPException(status_code=400, detail="仅支持MySQL实例")
     try:
@@ -190,6 +199,8 @@ async def show_grants(
     current_user: User = Depends(require_role("admin", "dba")),
 ):
     instance = await _get_decrypted_instance(db, instance_id)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
     if instance.db_type != "mysql":
         raise HTTPException(status_code=400, detail="仅支持MySQL实例")
     try:
@@ -216,6 +227,8 @@ async def slow_queries(
     current_user: User = Depends(require_role("admin", "dba")),
 ):
     instance = await _get_decrypted_instance(db, instance_id)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
     if instance.db_type != "mysql":
         raise HTTPException(status_code=400, detail="仅支持MySQL实例")
     try:
@@ -249,6 +262,8 @@ async def show_processlist(
     current_user: User = Depends(require_role("admin", "dba")),
 ):
     instance = await _get_decrypted_instance(db, instance_id)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
     if instance.db_type != "mysql":
         raise HTTPException(status_code=400, detail="仅支持MySQL实例")
     try:
@@ -268,6 +283,8 @@ async def kill_process(
     current_user: User = Depends(require_role("admin", "dba")),
 ):
     instance = await _get_decrypted_instance(db, instance_id)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
     if instance.db_type != "mysql":
         raise HTTPException(status_code=400, detail="仅支持MySQL实例")
     try:
@@ -286,6 +303,8 @@ async def show_status(
     current_user: User = Depends(require_role("admin", "dba")),
 ):
     instance = await _get_decrypted_instance(db, instance_id)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
     if instance.db_type != "mysql":
         raise HTTPException(status_code=400, detail="仅支持MySQL实例")
     try:
@@ -322,6 +341,8 @@ async def show_variables(
     current_user: User = Depends(require_role("admin", "dba")),
 ):
     instance = await _get_decrypted_instance(db, instance_id)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
     if instance.db_type != "mysql":
         raise HTTPException(status_code=400, detail="仅支持MySQL实例")
     try:

backend/app/api/v1/endpoints/masking_rule.py

@@ -0,0 +1,337 @@
+"""脱敏规则 API 路由 — 提供脱敏规则的 CRUD 和批量操作。"""
+import json
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import select, func, and_
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import get_db
+from app.core.security import get_current_user, require_role
+from app.models.user import User
+from app.models.instance import Instance
+from app.models.masking_rule import MaskingRule
+from app.schemas.masking import (
+    MaskingRuleCreate,
+    MaskingRuleUpdate,
+    MaskingRuleResponse,
+    MaskingRuleBatchCreate,
+    MaskingRuleListResponse,
+)
+
+router = APIRouter(prefix="/instances", tags=["数据脱敏规则"])
+
+
+def _serialize_rule(rule: MaskingRule) -> dict:
+    """Serialize a MaskingRule model, parsing mask_config JSON string."""
+    config = json.loads(rule.mask_config) if rule.mask_config else None
+    return {
+        "id": rule.id,
+        "instance_id": rule.instance_id,
+        "database": rule.database,
+        "table_name": rule.table_name,
+        "column_name": rule.column_name,
+        "mask_type": rule.mask_type,
+        "mask_config": config,
+        "is_enabled": rule.is_enabled,
+        "created_at": rule.created_at,
+        "updated_at": rule.updated_at,
+    }
+
+
+async def _verify_instance(
+    db: AsyncSession, instance_id: int, current_user: User
+) -> Instance:
+    """验证实例存在且用户有权限访问。
+
+    Args:
+        db: 数据库会话
+        instance_id: 实例ID
+        current_user: 当前用户
+
+    Returns:
+        实例对象
+
+    Raises:
+        HTTPException: 实例不存在或无权访问
+    """
+    result = await db.execute(select(Instance).where(Instance.id == instance_id))
+    instance = result.scalar_one_or_none()
+    if not instance:
+        raise HTTPException(status_code=404, detail="数据源不存在")
+    return instance
+
+
+async def _check_duplicate_rule(
+    db: AsyncSession,
+    instance_id: int,
+    database: str,
+    table_name: str,
+    column_name: str,
+    exclude_id: Optional[int] = None,
+) -> bool:
+    """检查是否存在重复规则。
+
+    Args:
+        db: 数据库会话
+        instance_id: 实例ID
+        database: 数据库名
+        table_name: 表名
+        column_name: 列名
+        exclude_id: 排除的规则ID（用于更新时）
+
+    Returns:
+        如果存在重复规则返回 True
+    """
+    stmt = select(MaskingRule).where(
+        MaskingRule.instance_id == instance_id,
+        MaskingRule.database == database,
+        MaskingRule.table_name == table_name,
+        MaskingRule.column_name == column_name,
+    )
+    if exclude_id:
+        stmt = stmt.where(MaskingRule.id != exclude_id)
+    result = await db.execute(stmt)
+    return result.scalar_one_or_none() is not None
+
+
+@router.get("/{instance_id}/masking-rules", response_model=MaskingRuleListResponse)
+async def list_masking_rules(
+    instance_id: int,
+    database: Optional[str] = Query(None, description="数据库筛选"),
+    table_name: Optional[str] = Query(None, description="表名筛选"),
+    keyword: Optional[str] = Query(None, description="关键词搜索"),
+    page: int = Query(1, ge=1, description="页码"),
+    page_size: int = Query(20, ge=1, le=100, description="每页数量"),
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """列出指定实例的脱敏规则。
+
+    Args:
+        instance_id: 数据源实例ID
+        database: 按数据库名筛选
+        table_name: 按表名筛选
+        keyword: 关键词搜索（匹配列名）
+        page: 页码
+        page_size: 每页数量
+        db: 数据库会话
+        current_user: 当前用户
+
+    Returns:
+        脱敏规则列表
+    """
+    await _verify_instance(db, instance_id, current_user)
+
+    stmt = select(MaskingRule).where(MaskingRule.instance_id == instance_id)
+
+    if database:
+        stmt = stmt.where(MaskingRule.database == database)
+    if table_name:
+        stmt = stmt.where(MaskingRule.table_name == table_name)
+    if keyword:
+        stmt = stmt.where(
+            MaskingRule.column_name.contains(keyword)
+        )
+
+    # 总数量
+    total_stmt = select(func.count()).select_from(stmt.subquery())
+    total = (await db.execute(total_stmt)).scalar() or 0
+
+    # 分页
+    stmt = (
+        stmt.order_by(MaskingRule.created_at.desc())
+        .offset((page - 1) * page_size)
+        .limit(page_size)
+    )
+    result = await db.execute(stmt)
+    rules = result.scalars().all()
+
+    return {
+        "total": total,
+        "items": [_serialize_rule(r) for r in rules],
+    }
+
+
+@router.post(
+    "/{instance_id}/masking-rules",
+    response_model=MaskingRuleResponse,
+    status_code=201,
+)
+async def create_masking_rule(
+    instance_id: int,
+    data: MaskingRuleCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """创建脱敏规则。"""
+    await _verify_instance(db, instance_id, current_user)
+
+    is_dup = await _check_duplicate_rule(
+        db, instance_id, data.database, data.table_name, data.column_name
+    )
+    if is_dup:
+        raise HTTPException(
+            status_code=409,
+            detail="该实例+数据库+表+列的脱敏规则已存在",
+        )
+
+    rule = MaskingRule(
+        instance_id=instance_id,
+        database=data.database,
+        table_name=data.table_name,
+        column_name=data.column_name,
+        mask_type=data.mask_type,
+        mask_config=json.dumps(data.mask_config) if data.mask_config else None,
+        is_enabled=data.is_enabled,
+    )
+    db.add(rule)
+    await db.commit()
+    await db.refresh(rule)
+    return _serialize_rule(rule)
+
+
+@router.put("/masking-rules/{rule_id}", response_model=MaskingRuleResponse)
+async def update_masking_rule(
+    rule_id: int,
+    data: MaskingRuleUpdate,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """更新脱敏规则。
+
+    Args:
+        rule_id: 规则ID
+        data: 脱敏规则更新数据
+        db: 数据库会话
+        current_user: 当前用户
+
+    Returns:
+        更新后的脱敏规则
+
+    Raises:
+        HTTPException: 规则不存在或重复
+    """
+    result = await db.execute(select(MaskingRule).where(MaskingRule.id == rule_id))
+    rule = result.scalar_one_or_none()
+    if not rule:
+        raise HTTPException(status_code=404, detail="脱敏规则不存在")
+
+    # 如果更新了唯一键字段，检查重复
+    db_val = data.database or rule.database
+    tb_val = data.table_name or rule.table_name
+    col_val = data.column_name or rule.column_name
+    is_dup = await _check_duplicate_rule(
+        db, rule.instance_id, db_val, tb_val, col_val, exclude_id=rule_id
+    )
+    if is_dup:
+        raise HTTPException(
+            status_code=409,
+            detail="更新后与现有规则冲突",
+        )
+
+    if data.database is not None:
+        rule.database = data.database
+    if data.table_name is not None:
+        rule.table_name = data.table_name
+    if data.column_name is not None:
+        rule.column_name = data.column_name
+    if data.mask_type is not None:
+        rule.mask_type = data.mask_type
+    if data.mask_config is not None:
+        rule.mask_config = json.dumps(data.mask_config)
+    if data.is_enabled is not None:
+        rule.is_enabled = data.is_enabled
+
+    await db.commit()
+    await db.refresh(rule)
+    return _serialize_rule(rule)
+
+
+@router.delete("/masking-rules/{rule_id}", status_code=204)
+async def delete_masking_rule(
+    rule_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """删除脱敏规则。
+
+    Args:
+        rule_id: 规则ID
+        db: 数据库会话
+        current_user: 当前用户
+
+    Raises:
+        HTTPException: 规则不存在
+    """
+    result = await db.execute(select(MaskingRule).where(MaskingRule.id == rule_id))
+    rule = result.scalar_one_or_none()
+    if not rule:
+        raise HTTPException(status_code=404, detail="脱敏规则不存在")
+
+    await db.delete(rule)
+    await db.commit()
+
+
+@router.post("/masking-rules/batch", response_model=MaskingRuleListResponse, status_code=201)
+async def batch_create_masking_rules(
+    data: MaskingRuleBatchCreate,
+    instance_id: int = Query(..., description="数据源实例ID"),
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """批量创建脱敏规则。
+
+    Args:
+        data: 批量规则数据
+        instance_id: 数据源实例ID
+        db: 数据库会话
+        current_user: 当前用户
+
+    Returns:
+        成功创建的规则列表
+
+    Raises:
+        HTTPException: 实例不存在或无权限
+    """
+    await _verify_instance(db, instance_id, current_user)
+
+    created_rules: list[MaskingRule] = []
+    errors: list[str] = []
+
+    for i, rule_data in enumerate(data.rules):
+        is_dup = await _check_duplicate_rule(
+            db, instance_id,
+            rule_data.database, rule_data.table_name, rule_data.column_name
+        )
+        if is_dup:
+            errors.append(
+                f"第 {i + 1} 条规则重复: {rule_data.database}.{rule_data.table_name}."
+                f"{rule_data.column_name}"
+            )
+            continue
+
+        rule = MaskingRule(
+            instance_id=instance_id,
+            database=rule_data.database,
+            table_name=rule_data.table_name,
+            column_name=rule_data.column_name,
+            mask_type=rule_data.mask_type,
+            mask_config=json.dumps(rule_data.mask_config) if rule_data.mask_config else None,
+            is_enabled=rule_data.is_enabled,
+        )
+        db.add(rule)
+        created_rules.append(rule)
+
+    if created_rules:
+        await db.commit()
+        for rule in created_rules:
+            await db.refresh(rule)
+
+    if errors and not created_rules:
+        raise HTTPException(status_code=409, detail="; ".join(errors))
+
+    return {
+        "total": len(created_rules),
+        "items": [_serialize_rule(r) for r in created_rules],
+    }

backend/app/api/v1/endpoints/mongo.py

@@ -0,0 +1,256 @@
+"""MongoDB 管理 API endpoints.
+
+提供数据库/集合列表、集合统计、只读查询等功能。
+所有端点要求进行身份认证，部分端点限制为 admin/dba 角色。
+"""
+
+import json
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from pydantic import BaseModel, Field
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import get_db
+from app.core.security import get_current_user, require_role
+from app.core.permissions import check_instance_access
+from app.models.instance import Instance
+from app.models.user import User
+from app.services.mongo_connector import MongoConnector, validate_operation
+
+router = APIRouter(tags=["MongoDB管理"])
+
+
+# ── 辅助函数 ─────────────────────────────────────────────────────
+
+
+async def _get_mongo_instance(instance_id: int, db: AsyncSession) -> Instance:
+    """获取并验证 MongoDB 实例。
+
+    Args:
+        instance_id: 实例 ID。
+        db: 数据库会话。
+
+    Returns:
+        Instance 模型对象（db_type 为 "mongo" 或 "mongodb"）。
+
+    Raises:
+        HTTPException: 实例不存在或不是 MongoDB 类型时。
+    """
+    result = await db.execute(select(Instance).where(Instance.id == instance_id))
+    instance = result.scalar_one_or_none()
+    if not instance:
+        raise HTTPException(status_code=404, detail="数据源不存在")
+    if instance.db_type not in ("mongo", "mongodb"):
+        raise HTTPException(status_code=400, detail="该实例不是 MongoDB 类型")
+    return instance
+
+
+# ── 请求/响应模型 ────────────────────────────────────────────────
+
+
+class MongoQueryRequest(BaseModel):
+    """MongoDB 只读查询请求体。"""
+
+    database: Optional[str] = Field(None, description="目标数据库名称，为空时使用实例默认库")
+    collection: str = Field(..., description="目标集合名称")
+    operation: str = Field("find", description="操作类型: find/findOne/countDocuments/distinct/aggregate")
+    filter: Dict[str, Any] = Field(default_factory=dict, description="查询过滤器")
+    projection: Optional[Dict[str, Any]] = Field(None, description="字段投影")
+    sort: Optional[Dict[str, int]] = Field(None, description="排序规则，如 {\"age\": -1}")
+    limit: int = Field(1000, ge=1, le=10000, description="返回上限")
+    skip: int = Field(0, ge=0, description="跳过数量")
+    pipeline: Optional[List[Dict[str, Any]]] = Field(None, description="aggregate 管道")
+    field: Optional[str] = Field(None, description="distinct 字段名")
+
+
+class MongoChangeRequest(BaseModel):
+    """MongoDB 变更操作请求体（工单执行用）。"""
+
+    database: Optional[str] = Field(None, description="目标数据库名称")
+    collection: str = Field(..., description="目标集合名称")
+    operation: str = Field(..., description="操作类型: insertOne/insertMany/updateOne/updateMany/deleteOne/deleteMany")
+    document: Optional[Dict[str, Any]] = Field(None, description="insertOne 文档")
+    documents: Optional[List[Dict[str, Any]]] = Field(None, description="insertMany 文档列表")
+    update: Optional[Dict[str, Any]] = Field(None, description="update 操作文档")
+    replacement: Optional[Dict[str, Any]] = Field(None, description="replaceOne 替换文档")
+    filter: Dict[str, Any] = Field(default_factory=dict, description="查询过滤器")
+
+
+# ── Endpoints ────────────────────────────────────────────────────
+
+
+@router.get("/mongo/{instance_id}/databases")
+async def list_mongo_databases(
+    instance_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """列出 MongoDB 实例中的所有数据库。"""
+    instance = await _get_mongo_instance(instance_id, db)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
+    try:
+        databases = await MongoConnector.list_databases(instance)
+        return {"databases": databases}
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"获取数据库列表失败: {exc}")
+
+
+@router.get("/mongo/{instance_id}/databases/{database}/collections")
+async def list_mongo_collections(
+    instance_id: int,
+    database: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """列出指定数据库中的所有集合。"""
+    instance = await _get_mongo_instance(instance_id, db)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
+    try:
+        collections = await MongoConnector.list_collections(instance, database)
+        return {"database": database, "collections": collections}
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"获取集合列表失败: {exc}")
+
+
+@router.get("/mongo/{instance_id}/collections/{collection}/stats")
+async def get_mongo_collection_stats(
+    instance_id: int,
+    collection: str,
+    database: Optional[str] = Query(None, description="数据库名称"),
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """获取指定集合的统计信息。"""
+    instance = await _get_mongo_instance(instance_id, db)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
+    db_name = database or getattr(instance, "database", None)
+    if not db_name:
+        raise HTTPException(status_code=400, detail="未指定数据库名称")
+    try:
+        stats = await MongoConnector.get_collection_stats(instance, db_name, collection)
+        return stats
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"获取集合统计失败: {exc}")
+
+
+@router.post("/mongo/{instance_id}/query")
+async def execute_mongo_query(
+    instance_id: int,
+    request: MongoQueryRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """执行 MongoDB 只读查询。
+
+    支持的只读操作:
+    - find: 需 filter, projection, sort, limit, skip
+    - findOne: 需 filter, projection
+    - countDocuments: 需 filter
+    - estimatedDocumentCount: 无需额外参数
+    - distinct: 需 field, filter
+    - aggregate: 需 pipeline
+
+    请求示例:
+    ```json
+    {
+        "database": "test",
+        "collection": "users",
+        "operation": "find",
+        "filter": {"age": {"$gt": 18}},
+        "projection": {"name": 1, "age": 1},
+        "sort": {"age": -1},
+        "limit": 100
+    }
+    ```
+    """
+    instance = await _get_mongo_instance(instance_id, db)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
+    db_name = request.database or getattr(instance, "database", None)
+    if not db_name:
+        raise HTTPException(status_code=400, detail="未指定数据库名称")
+
+    # 验证只读操作
+    is_readonly, error_msg = validate_operation(request.operation)
+    if not is_readonly:
+        raise HTTPException(status_code=403, detail=error_msg)
+
+    try:
+        query_json = request.model_dump(exclude_none=True, exclude={"database"})
+        # 移除未设置的 pipeline/field 避免错误分发
+        if request.operation != "aggregate":
+            query_json.pop("pipeline", None)
+        if request.operation != "distinct":
+            query_json.pop("field", None)
+
+        result = await MongoConnector.execute_query_json(
+            instance, db_name, query_json
+        )
+
+        return {
+            "success": True,
+            "operation": request.operation,
+            "collection": request.collection,
+            "result": result,
+        }
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc))
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"查询执行失败: {exc}")
+
+
+@router.post("/mongo/{instance_id}/execute")
+async def execute_mongo_change(
+    instance_id: int,
+    request: MongoChangeRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """执行 MongoDB 变更操作（仅 admin/dba）。
+
+    用于工单变更执行，支持 insert/update/delete 等写操作。
+    """
+    instance = await _get_mongo_instance(instance_id, db)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
+    db_name = request.database or getattr(instance, "database", None)
+    if not db_name:
+        raise HTTPException(status_code=400, detail="未指定数据库名称")
+
+    try:
+        change_json = request.model_dump(exclude_none=True, exclude={"database"})
+        result = await MongoConnector.execute_change_json(
+            instance, db_name, change_json
+        )
+        return {
+            "success": True,
+            "operation": request.operation,
+            "collection": request.collection,
+            "result": result,
+        }
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc))
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"变更执行失败: {exc}")
+
+
+@router.post("/mongo/{instance_id}/test-connection")
+async def test_mongo_connection(
+    instance_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """测试 MongoDB 实例连接。"""
+    instance = await _get_mongo_instance(instance_id, db)
+    # P0: 权限检查
+    await check_instance_access(current_user, instance, db)
+    try:
+        result = await MongoConnector.test_connection(instance)
+        return result.to_dict()
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"连接测试失败: {exc}")

backend/app/api/v1/endpoints/pool.py

@@ -0,0 +1,64 @@
+"""连接池管理 API。"""
+from typing import List
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import get_db
+from app.core.security import require_role
+from app.models.user import User
+from app.models.instance import Instance
+from app.services.connection_pool import get_pool_manager
+
+router = APIRouter(prefix="/pools", tags=["连接池管理"])
+
+
+@router.get("/instances/{instance_id}/stats")
+async def pool_stats(
+    instance_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """获取连接池状态。"""
+    result = await db.execute(select(Instance).where(Instance.id == instance_id))
+    instance = result.scalar_one_or_none()
+    if not instance:
+        raise HTTPException(status_code=404, detail="数据源不存在")
+
+    manager = get_pool_manager()
+    stats = manager.get_pool_stats(instance_id)
+    if not stats:
+        return {
+            "instance_id": instance_id,
+            "status": "not_initialized",
+            "message": "连接池尚未初始化，将在首次查询时自动创建",
+        }
+
+    return stats.to_dict()
+
+
+@router.post("/instances/{instance_id}/reset")
+async def reset_pool(
+    instance_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """重置连接池。"""
+    manager = get_pool_manager()
+    await manager.reset_pool(instance_id)
+    return {"message": "连接池已重置", "instance_id": instance_id}
+
+
+@router.get("/overview")
+async def pools_overview(
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """所有连接池概览。"""
+    manager = get_pool_manager()
+    all_stats = manager.get_all_stats()
+
+    return {
+        "total_pools": len(all_stats),
+        "pools": [stats.to_dict() for stats in all_stats.values()],
+    }

backend/app/api/v1/endpoints/query.py

@@ -1,4 +1,14 @@
-from fastapi import APIRouter, Depends, HTTPException
+"""统一查询 API — 支持 MySQL、PostgreSQL、Redis、MongoDB 的只读查询。
+
+核心设计:
+- 一个入口查询所有数据库，减少学习成本
+- 自动审计记录所有查询（谁在什么时间查了什么）
+- 自动记录查询历史（方便复用）
+- 数据脱敏保护敏感信息
+- 完整错误处理和人性化提示
+"""
+import time
+from fastapi import APIRouter, Depends, HTTPException, Query as QueryParam
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -7,43 +17,210 @@ from app.core.security import get_current_user
 from app.models.user import User
 from app.models.instance import Instance
 from app.models.audit_log import AuditLog
-from app.schemas.query import QueryRequest, QueryResult
-from app.services.db_connector import DBConnector
+from app.schemas.query import QueryRequest, UnifiedQueryResult
+from app.services.query_service import QueryService
+from app.services.data_masker import DataMasker, load_masking_rules
+from app.services.audit_log_service import AuditLogService
+from app.services.query_history_service import QueryHistoryService
 from app.core.crypto import decrypt_password
+from app.core.permissions import check_instance_access, check_instance_permission
 
-router = APIRouter(prefix="/query", tags=["SQL 查询"])
+router = APIRouter(prefix="/query", tags=["统一查询"])
 
 
-@router.post("", response_model=QueryResult)
+@router.post("", response_model=UnifiedQueryResult)
 async def execute_query(
     req: QueryRequest,
+    mask: bool = QueryParam(True, description="是否启用数据脱敏"),
+    save_history: bool = QueryParam(True, description="是否保存到查询历史"),
     db: AsyncSession = Depends(get_db),
     current_user: User = Depends(get_current_user),
 ):
-    if current_user.role == "guest":
-        raise HTTPException(status_code=403, detail="访客无权执行查询")
+    """执行统一查询（只读）— 支持 MySQL、PostgreSQL、Redis、MongoDB。
 
+    **一个入口，查询所有数据库。** 系统会根据数据源类型自动路由到对应的查询引擎，
+    并进行只读安全验证。
+
+    **自动审计**: 每次查询都会被记录在案，包含查询内容、耗时、返回行数等。
+
+    **查询历史**: 自动保存查询记录，方便后续复用。
+
+    **不同数据库的查询格式:**
+    - **MySQL/PostgreSQL**: 标准 SQL SELECT 语句，如 `SELECT * FROM users LIMIT 10`
+    - **Redis**: 命令字符串，如 `GET mykey`、`HGETALL myhash`、`LRANGE mylist 0 -1`
+    - **MongoDB**: JSON 格式查询，如:
+      ```json
+      {"collection": "users", "operation": "find",
+       "filter": {"age": {"$gt": 18}}, "limit": 10}
+      ```
+
+    Args:
+        req: 查询请求（instance_id + database + sql）
+        mask: 是否启用数据脱敏（默认 True）
+        save_history: 是否保存到查询历史（默认 True）
+
+    Returns:
+        UnifiedQueryResult: 统一格式的查询结果
+    """
+    if current_user.role == "guest":
+        raise HTTPException(status_code=403, detail="访客无权执行查询，请联系管理员申请权限")
+
+    # 查询实例信息
     result = await db.execute(select(Instance).where(Instance.id == req.instance_id))
     instance = result.scalar_one_or_none()
     if not instance:
-        raise HTTPException(status_code=404, detail="数据源不存在")
+        raise HTTPException(status_code=404, detail="数据源不存在，请确认实例ID正确")
 
-    # Decrypt password for query execution
+    # 权限组检查：用户是否有权访问该数据源
+    await check_instance_access(current_user, instance, db)
+
+    # P1: 操作级权限检查
+    await check_instance_permission(current_user, instance, "query:select", db)
+
+    # 解密密码
     instance.password = decrypt_password(instance.password)
 
-    try:
-        query_result = await DBConnector.execute_query(instance, req.database, req.sql, current_user.id)
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
+    # 记录开始时间
+    start_time = time.time()
+    is_success = True
+    error_msg = None
 
-    # Record audit log
-    log = AuditLog(
-        user_id=current_user.id,
-        action="QUERY",
-        target_type="query",
-        target_id=instance.id,
-        detail=req.sql[:500],
-    )
-    db.add(log)
+    try:
+        query_result = await QueryService.execute_query(
+            instance, req.database, req.sql, current_user.id, req.options
+        )
+    except ValueError as e:
+        is_success = False
+        error_msg = str(e)
+        raise HTTPException(status_code=400, detail=str(e))
+    except Exception as e:
+        is_success = False
+        error_msg = f"{type(e).__name__}: {e}"
+        raise HTTPException(status_code=500, detail=f"查询执行失败: {e}")
+    finally:
+        # 无论成功失败都记录审计日志
+        elapsed_ms = int((time.time() - start_time) * 1000)
+
+        # 1. 记录审计日志（完整上下文）
+        try:
+            username = current_user.display_name or current_user.username
+            await AuditLogService.log(
+                db=db,
+                user_id=current_user.id,
+                action="QUERY",
+                target_type="query",
+                target_id=instance.id,
+                detail=f"查询{'成功' if is_success else '失败'}: {req.sql[:200]}",
+            )
+            # 直接写入增强字段（AuditLogService.log只写基础字段）
+            log_entry = AuditLog(
+                user_id=current_user.id,
+                username=username,
+                action="QUERY",
+                target_type="query",
+                target_id=instance.id,
+                instance_name=instance.name,
+                db_type=instance.db_type,
+                database_name=req.database or instance.database,
+                query_content=req.sql[:2000],
+                query_type=getattr(query_result, 'query_type', 'UNKNOWN') if is_success else None,
+                execution_time_ms=elapsed_ms,
+                rows_affected=getattr(query_result, 'total', 0) if is_success else 0,
+                result_status="success" if is_success else "failed",
+                detail=f"查询{'成功' if is_success else '失败'}: {req.sql[:200]}{f', 错误: {error_msg}' if error_msg else ''}",
+            )
+            db.add(log_entry)
+            await db.commit()
+        except Exception:
+            pass  # 审计记录失败不影响查询结果
+
+    # 数据脱敏处理
+    if is_success and mask and query_result.rows and instance.db_type in ("mysql", "postgres", "postgresql"):
+        try:
+            if req.database:
+                masking_rules = await load_masking_rules(db, req.instance_id, req.database)
+                if masking_rules:
+                    masker = DataMasker(masking_rules)
+                    from app.services.sql_analyzer import SQLAnalyzer
+                    analyzer = SQLAnalyzer()
+                    analysis = analyzer.analyze(req.sql)
+                    tables = analysis.tables if analysis.tables else ["*"]
+                    table_name = tables[0] if tables else "*"
+                    if masker.has_rules_for_table(table_name):
+                        query_result.rows = masker.mask_rows(table_name, query_result.rows)
+        except Exception:
+            pass  # 脱敏失败不影响查询结果
+
+    # 记录查询历史（人性化：方便用户复用）
+    if is_success and save_history:
+        try:
+            username = current_user.display_name or current_user.username
+            await QueryHistoryService.record_query(
+                user_id=current_user.id,
+                username=username,
+                instance_id=instance.id,
+                instance_name=instance.name,
+                db_type=instance.db_type,
+                database_name=req.database or instance.database,
+                query_content=req.sql,
+                query_type=getattr(query_result, 'query_type', 'QUERY'),
+                execution_time_ms=elapsed_ms,
+                result_rows=getattr(query_result, 'total', 0),
+                is_success=is_success,
+                error_message=error_msg,
+            )
+        except Exception:
+            pass  # 历史记录失败不影响查询结果
 
     return query_result
+
+
+@router.get("/supported-types")
+async def supported_db_types(
+    current_user: User = Depends(get_current_user),
+):
+    """获取支持的数据库类型和查询格式说明。
+
+    人性化设计：帮助用户了解如何查询不同类型的数据库。
+    """
+    return {
+        "types": [
+            {
+                "type": "mysql",
+                "name": "MySQL",
+                "query_format": "标准 SQL SELECT 语句",
+                "example": "SELECT id, name FROM users WHERE status = 'active' LIMIT 10",
+                "security": "AST语法分析验证只读，禁止DDL/DML",
+                "features": ["语法高亮", "EXPLAIN分析", "查询建议", "数据脱敏"],
+            },
+            {
+                "type": "postgres",
+                "name": "PostgreSQL",
+                "query_format": "标准 SQL SELECT 语句",
+                "example": "SELECT id, name FROM users WHERE status = 'active' LIMIT 10",
+                "security": "AST语法分析验证只读，禁止DDL/DML",
+                "features": ["语法高亮", "EXPLAIN分析", "查询建议", "数据脱敏"],
+            },
+            {
+                "type": "redis",
+                "name": "Redis",
+                "query_format": "Redis 命令字符串",
+                "example": "HGETALL user:1001",
+                "security": "130+只读命令白名单验证",
+                "features": ["命令自动补全", "结果格式化", "键值浏览"],
+                "readonly_commands": [
+                    "GET", "MGET", "HGET", "HGETALL", "LRANGE", "SMEMBERS",
+                    "ZRANGE", "SCAN", "KEYS", "INFO", "TTL", "TYPE", "EXISTS",
+                ],
+            },
+            {
+                "type": "mongo",
+                "name": "MongoDB",
+                "query_format": "JSON 格式查询",
+                "example": '{"collection": "users", "operation": "find", "filter": {}, "limit": 10}',
+                "security": "6个只读操作白名单：find/findOne/countDocuments/distinct/aggregate",
+                "features": ["JSON格式化", "结果表格化", "集合浏览"],
+                "readonly_operations": ["find", "findOne", "countDocuments", "estimatedDocumentCount", "distinct", "aggregate"],
+            },
+        ]
+    }

backend/app/api/v1/endpoints/query_history.py

@@ -0,0 +1,103 @@
+"""查询历史和快捷查询 API — 人性化查询管理。"""
+from typing import Optional
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import get_db
+from app.core.security import get_current_user
+from app.models.user import User
+from app.services.query_history_service import QueryHistoryService
+
+router = APIRouter(prefix="/query-history", tags=["查询历史"])
+
+
+@router.get("/my")
+async def my_query_history(
+    db_type: Optional[str] = None,
+    instance_id: Optional[int] = None,
+    keyword: Optional[str] = None,
+    page: int = 1,
+    page_size: int = 20,
+    current_user: User = Depends(get_current_user),
+):
+    """获取我的查询历史。
+
+    自动记录每次查询，支持按数据库类型、实例、关键词筛选。
+    """
+    return await QueryHistoryService.get_history(
+        user_id=current_user.id,
+        db_type=db_type,
+        instance_id=instance_id,
+        keyword=keyword,
+        page=page,
+        page_size=page_size,
+    )
+
+
+@router.post("/{history_id}/favorite")
+async def toggle_favorite(
+    history_id: int,
+    current_user: User = Depends(get_current_user),
+):
+    """切换查询历史收藏状态。"""
+    is_fav = await QueryHistoryService.toggle_favorite(current_user.id, history_id)
+    return {"is_favorite": is_fav, "message": "已收藏" if is_fav else "已取消收藏"}
+
+
+@router.get("/favorites")
+async def my_favorites(
+    page: int = 1,
+    page_size: int = 20,
+    current_user: User = Depends(get_current_user),
+):
+    """获取我收藏的查询。"""
+    return await QueryHistoryService.get_favorites(
+        user_id=current_user.id, page=page, page_size=page_size
+    )
+
+
+@router.get("/statistics")
+async def my_query_statistics(
+    current_user: User = Depends(get_current_user),
+):
+    """获取我的查询统计 — 了解自己的数据使用习惯。"""
+    return await QueryHistoryService.get_query_statistics(user_id=current_user.id)
+
+
+# ── 快捷查询 ──
+
+@router.get("/quick-queries")
+async def list_quick_queries(
+    db_type: Optional[str] = None,
+    keyword: Optional[str] = None,
+    current_user: User = Depends(get_current_user),
+):
+    """获取快捷查询列表（个人+团队共享）。"""
+    items = await QueryHistoryService.get_quick_queries(
+        user_id=current_user.id, db_type=db_type, keyword=keyword
+    )
+    return {"items": items, "total": len(items)}
+
+
+@router.post("/quick-queries")
+async def create_quick_query(
+    name: str,
+    query_content: str,
+    db_type: str,
+    description: str = "",
+    database_pattern: str = "*",
+    is_shared: bool = False,
+    current_user: User = Depends(get_current_user),
+):
+    """创建快捷查询。"""
+    qq = await QueryHistoryService.create_quick_query(
+        name=name,
+        description=description,
+        query_content=query_content,
+        db_type=db_type,
+        database_pattern=database_pattern,
+        is_shared=is_shared,
+        created_by=current_user.id,
+        created_by_name=current_user.display_name or current_user.username,
+    )
+    return {"id": qq.id, "name": qq.name, "message": "快捷查询已创建"}

backend/app/api/v1/endpoints/redis.py

@@ -1,4 +1,10 @@
-"""Redis management API endpoints."""
+"""Redis management API endpoints.
+
+集成 RedisCommandValidator 进行只读命令安全验证：
+- developer 角色仅允许执行只读命令
+- admin/dba 角色可执行所有命令
+- FLUSHDB 等高危操作仅限 admin
+"""
 from typing import Optional
 
 from fastapi import APIRouter, Depends, HTTPException, Query
@@ -11,6 +17,7 @@ from app.core.security import get_current_user, require_role
 from app.models.user import User
 from app.models.instance import Instance
 from app.services import redis_connector
+from app.services.redis_validator import RedisCommandValidator
 
 router = APIRouter(tags=["Redis管理"])
 
@@ -196,10 +203,38 @@ async def execute_redis_command(
     instance_id: int,
     data: RedisCommandRequest,
     db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(require_role("admin", "dba")),
+    current_user: User = Depends(get_current_user),
 ):
-    """执行命令."""
+    """执行 Redis 命令（带安全验证）。
+
+    权限控制:
+    - developer 角色只能执行只读命令（白名单内的 GET, HGETALL 等）
+    - admin/dba 角色可以执行所有命令（但 FLUSHDB/FLUSHALL 仍仅限 admin）
+    """
     instance = await _get_redis_instance(instance_id, db)
+
+    # 提取命令关键字用于权限判断
+    parts = data.command.strip().split()
+    if not parts:
+        raise HTTPException(status_code=400, detail="命令不能为空")
+    cmd = parts[0].upper()
+
+    # ── 权限校验 ────────────────────────────────────────────────
+
+    # FLUSHDB / FLUSHALL 仅 admin 可用
+    if cmd in ("FLUSHDB", "FLUSHALL"):
+        if current_user.role != "admin":
+            raise HTTPException(
+                status_code=403, detail=f"Redis 命令 '{cmd}' 仅限 admin 执行"
+            )
+    else:
+        # developer 只允许只读命令
+        if current_user.role == "developer":
+            is_safe, error_msg = RedisCommandValidator.validate(data.command)
+            if not is_safe:
+                raise HTTPException(status_code=403, detail=error_msg)
+        # admin/dba 允许所有命令（上面已拦截 FLUSHDB/FLUSHALL）
+
     try:
         output = await redis_connector.RedisConnector.execute_command(
             instance, data.db, data.command

backend/app/api/v1/endpoints/rollback.py

@@ -0,0 +1,176 @@
+"""回滚 SQL API 路由 — 提供回滚 SQL 生成和预览功能。"""
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import get_db
+from app.core.security import get_current_user, require_role
+from app.core.crypto import decrypt_password
+from app.models.user import User
+from app.models.instance import Instance
+from app.models.ticket import Ticket
+from app.schemas.rollback import (
+    RollbackPreviewRequest,
+    RollbackPreviewResponse,
+    RollbackGenerateResponse,
+    RollbackGenerateRequest,
+)
+from app.services.rollback_generator import RollbackGenerator, RollbackGeneratorError
+
+router = APIRouter(tags=["回滚 SQL"])
+
+
+async def _get_ticket(db: AsyncSession, ticket_id: int) -> Ticket:
+    """获取工单并检查是否存在。
+
+    Args:
+        db: 数据库会话
+        ticket_id: 工单ID
+
+    Returns:
+        工单对象
+
+    Raises:
+        HTTPException: 工单不存在
+    """
+    result = await db.execute(select(Ticket).where(Ticket.id == ticket_id))
+    ticket = result.scalar_one_or_none()
+    if not ticket:
+        raise HTTPException(status_code=404, detail="工单不存在")
+    return ticket
+
+
+async def _get_instance(db: AsyncSession, instance_id: int) -> Instance:
+    """获取实例并检查是否存在。
+
+    Args:
+        db: 数据库会话
+        instance_id: 实例ID
+
+    Returns:
+        实例对象
+
+    Raises:
+        HTTPException: 实例不存在
+    """
+    result = await db.execute(select(Instance).where(Instance.id == instance_id))
+    instance = result.scalar_one_or_none()
+    if not instance:
+        raise HTTPException(status_code=404, detail="数据源不存在")
+    return instance
+
+
+@router.post("/tickets/{ticket_id}/generate-rollback", response_model=RollbackGenerateResponse)
+async def generate_rollback_sql(
+    ticket_id: int,
+    data: RollbackGenerateRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """为工单生成回滚 SQL。
+
+    根据工单中的 SQL 内容，自动生成对应的回滚 SQL。
+    支持 INSERT/UPDATE/DELETE 三种 DML 操作。
+
+    Args:
+        ticket_id: 工单ID
+        data: 生成选项（是否自动保存）
+        db: 数据库会话
+        current_user: 当前用户
+
+    Returns:
+        生成的回滚 SQL 信息
+
+    Raises:
+        HTTPException: 工单不存在、无权访问或生成失败
+    """
+    ticket = await _get_ticket(db, ticket_id)
+
+    # 获取关联的数据源实例
+    instance = await _get_instance(db, ticket.instance_id)
+    instance.password = decrypt_password(instance.password)
+
+    # 生成回滚 SQL
+    generator = RollbackGenerator()
+    try:
+        rollback_sql = await generator.generate(
+            instance, ticket.target_database, ticket.sql_content
+        )
+    except RollbackGeneratorError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"回滚 SQL 生成失败: {str(e)}")
+
+    # 自动保存到工单
+    saved = False
+    if data.auto_save:
+        ticket.rollback_sql = rollback_sql
+        await db.commit()
+        saved = True
+
+    # 检测 SQL 类型
+    from app.services.sql_analyzer import SQLAnalyzer
+    analyzer = SQLAnalyzer()
+    sql_type = analyzer._detect_sql_type(ticket.sql_content)
+
+    return RollbackGenerateResponse(
+        ticket_id=ticket_id,
+        original_sql=ticket.sql_content,
+        rollback_sql=rollback_sql,
+        sql_type=sql_type.value if sql_type else None,
+        saved=saved,
+    )
+
+
+@router.post("/rollback/preview", response_model=RollbackPreviewResponse)
+async def preview_rollback_sql(
+    req: RollbackPreviewRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """预览回滚 SQL（不保存）。
+
+    为给定的 SQL 语句生成回滚 SQL 预览，不执行实际数据库操作，
+    也不保存到任何工单。
+
+    Args:
+        req: 预览请求（实例ID、数据库、SQL）
+        db: 数据库会话
+        current_user: 当前用户
+
+    Returns:
+        回滚 SQL 预览
+
+    Raises:
+        HTTPException: 实例不存在或无权限
+    """
+    instance = await _get_instance(db, req.instance_id)
+    instance.password = decrypt_password(instance.password)
+
+    generator = RollbackGenerator()
+    try:
+        rollback_sql = await generator.preview(instance, req.database, req.sql)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"回滚 SQL 预览失败: {str(e)}")
+
+    # 检测 SQL 类型
+    from app.services.sql_analyzer import SQLAnalyzer
+    analyzer = SQLAnalyzer()
+    sql_type = analyzer._detect_sql_type(req.sql)
+
+    warning = None
+    if rollback_sql.startswith("--"):
+        # 提取警告信息
+        lines = rollback_sql.split("\n")
+        warning_lines = [l for l in lines if l.startswith("--") and "无法" in l]
+        if warning_lines:
+            warning = warning_lines[0].replace("-- ", "")
+
+    return RollbackPreviewResponse(
+        original_sql=req.sql,
+        rollback_sql=rollback_sql,
+        sql_type=sql_type.value if sql_type else None,
+        warning=warning,
+    )

backend/app/api/v1/endpoints/schema_drift.py

@@ -0,0 +1,85 @@
+"""Schema 漂移检测 API。"""
+from typing import List
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import get_db
+from app.core.security import get_current_user, require_role
+from app.core.crypto import decrypt_password
+from app.models.user import User
+from app.models.instance import Instance
+from app.services.schema_drift_detector import SchemaDriftDetector
+
+router = APIRouter(prefix="/schema-drift", tags=["Schema 漂移检测"])
+
+
+def _serialize_drift_report(report):
+    return report.to_dict() if hasattr(report, "to_dict") else report
+
+
+@router.post("/instances/{instance_id}/capture")
+async def capture_schema(
+    instance_id: int,
+    database: str = "",
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """捕获 Schema 快照并保存为基线。"""
+    result = await db.execute(select(Instance).where(Instance.id == instance_id))
+    instance = result.scalar_one_or_none()
+    if not instance:
+        raise HTTPException(status_code=404, detail="数据源不存在")
+
+    if not database:
+        database = instance.database or ""
+
+    detector = SchemaDriftDetector()
+    instance.password = decrypt_password(instance.password)
+    snapshot = await detector.capture_snapshot(instance, database)
+    await detector.save_baseline(instance_id, database, snapshot)
+
+    return {
+        "message": "Schema 快照已捕获",
+        "instance_id": instance_id,
+        "database": database,
+        "total_columns": len(snapshot),
+    }
+
+
+@router.post("/instances/{instance_id}/check")
+async def check_drift(
+    instance_id: int,
+    database: str = "",
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """检测 Schema 漂移。"""
+    result = await db.execute(select(Instance).where(Instance.id == instance_id))
+    instance = result.scalar_one_or_none()
+    if not instance:
+        raise HTTPException(status_code=404, detail="数据源不存在")
+
+    if not database:
+        database = instance.database or ""
+
+    detector = SchemaDriftDetector()
+    instance.password = decrypt_password(instance.password)
+    report = await detector.detect_drift(instance, database)
+
+    return _serialize_drift_report(report)
+
+
+@router.get("/instances/{instance_id}/history")
+async def schema_history(
+    instance_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """获取 Schema 变更历史。"""
+    # 简化实现，实际应从数据库查询
+    return {
+        "instance_id": instance_id,
+        "history": [],
+        "message": "Schema 变更历史功能开发中，当前使用内存存储",
+    }

backend/app/api/v1/endpoints/slow_log.py

@@ -0,0 +1,103 @@
+"""慢日志 API。"""
+from typing import Optional
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import get_db
+from app.core.security import get_current_user, require_role
+from app.core.crypto import decrypt_password
+from app.models.user import User
+from app.models.instance import Instance
+from app.services.slow_log_collector import SlowLogCollector
+
+router = APIRouter(prefix="/slow-logs", tags=["慢日志分析"])
+
+
+@router.get("/instances/{instance_id}")
+async def list_slow_logs(
+    instance_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """获取指定实例的慢日志。"""
+    result = await db.execute(select(Instance).where(Instance.id == instance_id))
+    instance = result.scalar_one_or_none()
+    if not instance:
+        raise HTTPException(status_code=404, detail="数据源不存在")
+
+    collector = SlowLogCollector()
+    logs = collector.get_slow_logs(instance_id)
+
+    return {
+        "instance_id": instance_id,
+        "total": len(logs),
+        "logs": [log.to_dict() for log in logs],
+    }
+
+
+@router.post("/instances/{instance_id}/collect")
+async def collect_slow_logs(
+    instance_id: int,
+    database: str = "",
+    limit: int = 50,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """手动收集慢日志。"""
+    result = await db.execute(select(Instance).where(Instance.id == instance_id))
+    instance = result.scalar_one_or_none()
+    if not instance:
+        raise HTTPException(status_code=404, detail="数据源不存在")
+
+    if not database:
+        database = instance.database or ""
+
+    collector = SlowLogCollector()
+    instance.password = decrypt_password(instance.password)
+
+    try:
+        if instance.db_type == "mysql":
+            logs = await collector.collect_mysql_slow_logs(instance, database, limit)
+        elif instance.db_type in ("postgres", "postgresql"):
+            logs = await collector.collect_postgres_slow_logs(instance, database, limit)
+        else:
+            raise HTTPException(status_code=400, detail=f"不支持的类型: {instance.db_type}")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"收集失败: {str(e)}")
+
+    await collector.save_slow_logs(instance_id, logs)
+    report = collector.get_report(instance_id)
+
+    return report.to_dict()
+
+
+@router.get("/instances/{instance_id}/report")
+async def slow_log_report(
+    instance_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    """获取慢日志分析报告。"""
+    collector = SlowLogCollector()
+    report = collector.get_report(instance_id)
+    return report.to_dict()
+
+
+@router.get("/dashboard/top")
+async def dashboard_top_queries(
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """仪表盘慢查询 TOP 10。"""
+    collector = SlowLogCollector()
+    all_logs = []
+    for instance_id, logs in collector._logs.items():
+        all_logs.extend(logs)
+
+    # 全局 TOP 10
+    top = sorted(all_logs, key=lambda x: x.avg_time_ms, reverse=True)[:10]
+    return {
+        "total_instances": len(collector._logs),
+        "top_queries": [q.to_dict() for q in top],
+    }

backend/app/api/v1/endpoints/sql_assistant.py

@@ -0,0 +1,113 @@
+"""SQL助手 API — 人性化辅助功能。"""
+from typing import Optional
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+
+from app.core.security import get_current_user
+from app.models.user import User
+from app.services.sql_assistant import SQLAssistant
+
+router = APIRouter(prefix="/sql-assistant", tags=["SQL助手"])
+
+
+class SQLAnalyzeRequest(BaseModel):
+    sql: str
+    db_type: str = "mysql"
+
+
+class SQLFormatRequest(BaseModel):
+    sql: str
+
+
+@router.post("/analyze")
+async def analyze_sql(
+    req: SQLAnalyzeRequest,
+    current_user: User = Depends(get_current_user),
+):
+    """全面分析SQL — 格式化+高亮+风险+建议。
+
+    一次调用获取所有辅助信息，帮助用户写出更好的SQL。
+    """
+    if not req.sql or not req.sql.strip():
+        raise HTTPException(status_code=400, detail="SQL内容不能为空")
+
+    result = SQLAssistant.analyze(req.sql, req.db_type)
+    return result
+
+
+@router.post("/format")
+async def format_sql(
+    req: SQLFormatRequest,
+    current_user: User = Depends(get_current_user),
+):
+    """格式化SQL — 统一代码风格。
+
+    - 关键字大写
+    - 合理换行缩进
+    """
+    if not req.sql or not req.sql.strip():
+        raise HTTPException(status_code=400, detail="SQL内容不能为空")
+
+    result = SQLAssistant.format_sql(req.sql)
+    return {
+        "formatted": result.formatted,
+        "is_valid": result.is_valid,
+        "error": result.error,
+    }
+
+
+@router.post("/highlight")
+async def highlight_sql(
+    req: SQLAnalyzeRequest,
+    current_user: User = Depends(get_current_user),
+):
+    """SQL语法高亮+风险检测。
+
+    返回高亮文本和风险提示列表。
+    """
+    if not req.sql or not req.sql.strip():
+        raise HTTPException(status_code=400, detail="SQL内容不能为空")
+
+    result = SQLAssistant.highlight_sql(req.sql)
+    return {
+        "highlighted": result.highlighted,
+        "keywords": result.keywords,
+        "tables": result.tables,
+        "risks": result.risks,
+    }
+
+
+@router.post("/suggestions")
+async def query_suggestions(
+    req: SQLAnalyzeRequest,
+    current_user: User = Depends(get_current_user),
+):
+    """获取查询建议 — 优化提示。
+
+    基于SQL内容提供人性化的改进建议。
+    """
+    if not req.sql or not req.sql.strip():
+        raise HTTPException(status_code=400, detail="SQL内容不能为空")
+
+    suggestions = SQLAssistant.get_query_suggestions(req.sql, req.db_type)
+    return {"suggestions": suggestions, "count": len(suggestions)}
+
+
+@router.post("/parse-explain")
+async def parse_explain(
+    explain_rows: list,
+    current_user: User = Depends(get_current_user),
+):
+    """解析MySQL EXPLAIN结果为人类可读格式。"""
+    if not explain_rows:
+        raise HTTPException(status_code=400, detail="EXPLAIN结果不能为空")
+
+    result = SQLAssistant.parse_explain(explain_rows)
+    return {
+        "is_simple": result.is_simple,
+        "has_index": result.has_index,
+        "has_full_scan": result.has_full_scan,
+        "estimated_rows": result.estimated_rows,
+        "warnings": result.warnings,
+        "human_readable": result.human_readable,
+    }

backend/app/api/v1/endpoints/ticket.py

@@ -1,6 +1,9 @@
-from typing import Optional, List, Dict, Any
+from typing import Optional, List, Dict, Any, TYPE_CHECKING
 from datetime import datetime, timezone
 
+if TYPE_CHECKING:
+    from app.services.ticket_executor import ExecutionReport
+
 from fastapi import APIRouter, Depends, HTTPException, Query
 from pydantic import BaseModel
 from sqlalchemy import select, func, or_
@@ -13,8 +16,11 @@ from app.models.ticket import Ticket
 from app.models.instance import Instance
 from app.schemas.ticket import TicketCreate, TicketReview, TicketResponse
 from app.core.crypto import decrypt_password
+from app.core.permissions import check_instance_access, check_instance_permission
 from app.services.audit_engine import SQLAuditEngine
+from app.services.audit_engine_v2 import SQLAuditEngineV2
 from app.services.audit_log_service import AuditLogService
+from app.models.workflow import WorkflowDefinition, WorkflowStep, TicketApproval
 import json
 
 
@@ -111,13 +117,34 @@ async def create_ticket(
     current_user: User = Depends(get_current_user),
 ):
     # Verify instance exists
-    inst = await db.execute(select(Instance).where(Instance.id == data.instance_id))
-    if not inst.scalar_one_or_none():
+    inst_result = await db.execute(select(Instance).where(Instance.id == data.instance_id))
+    instance = inst_result.scalar_one_or_none()
+    if not instance:
         raise HTTPException(status_code=400, detail="数据源不存在")
 
-    # Auto-audit SQL before creating ticket
-    engine = SQLAuditEngine(env="production" if data.db_type == "mysql" else "development")
-    report = engine.audit(data.sql_content)
+    # 权限组检查：用户是否有权对该数据源提交工单
+    try:
+        await check_instance_access(current_user, instance, db)
+    except HTTPException:
+        raise HTTPException(status_code=403, detail="您没有权限对该数据源提交变更工单")
+
+    # P1: 操作级权限检查
+    await check_instance_permission(current_user, instance, "ticket:create", db)
+
+    # Auto-audit SQL before creating ticket (使用 V2 引擎)
+    v2_engine = SQLAuditEngineV2(db_session=db, env="production" if data.db_type == "mysql" else "development")
+    v2_report = await v2_engine.audit(data.sql_content, db_type=data.db_type)
+
+    # Check for default workflow to determine initial status
+    default_wf_result = await db.execute(
+        select(WorkflowDefinition).where(
+            WorkflowDefinition.is_default == True,
+            WorkflowDefinition.is_active == True,
+        )
+    )
+    default_workflow = default_wf_result.scalar_one_or_none()
+
+    initial_status = "reviewing" if default_workflow else "pending"
 
     ticket = Ticket(
         title=data.title,
@@ -129,23 +156,42 @@ async def create_ticket(
         reason=data.reason,
         priority=data.priority,
         applicant_id=current_user.id,
-        status="pending",
-        audit_result=json.dumps(report.to_dict(), ensure_ascii=False) if report.findings else None,
+        status=initial_status,
+        audit_result=json.dumps(v2_report.to_dict(), ensure_ascii=False) if v2_report.findings else None,
     )
     db.add(ticket)
-    await db.flush()  # flush to get ticket.id for audit log
+    await db.flush()  # flush to get ticket.id
+
+    # If default workflow exists, create TicketApproval entries for each step
+    if default_workflow:
+        wf_steps_result = await db.execute(
+            select(WorkflowStep)
+            .where(WorkflowStep.workflow_id == default_workflow.id)
+            .order_by(WorkflowStep.step_number)
+        )
+        wf_steps = wf_steps_result.scalars().all()
+        for step in wf_steps:
+            approval = TicketApproval(
+                ticket_id=ticket.id,
+                step_number=step.step_number,
+                step_name=step.name,
+                approver_role=step.approver_role,
+                action="pending",
+            )
+            db.add(approval)
+        await db.flush()
 
     # Log ticket creation
     await AuditLogService.log(
         db, user_id=current_user.id, action="CREATE_TICKET",
         target_type="ticket", target_id=ticket.id,
-        detail=f"SQL审核评分: {report.score}, 通过: {report.passed}",
+        detail=f"SQL审核评分: {v2_report.score}, 通过: {v2_report.passed}, 规则数: {len(v2_report.findings)}",
     )
 
     await db.commit()
     await db.refresh(ticket)
     result = await _enrich_ticket(ticket, db)
-    result["audit_report"] = report.to_dict()
+    result["audit_report"] = v2_report.to_dict()
     return result
 
 
@@ -184,15 +230,19 @@ async def audit_sql(
         raise HTTPException(status_code=400, detail="数据源不存在")
 
     env = "production" if instance.env == "production" else "development"
-    engine = SQLAuditEngine(env=env)
-    report = engine.audit(data.sql_content)
+    db_type = instance.db_type or "mysql"
+
+    # 使用 V2 引擎进行审核（支持数据库加载规则配置）
+    v2_engine = SQLAuditEngineV2(db_session=db, env=env)
+    report = await v2_engine.audit(data.sql_content, db_type=db_type)
 
     return {
         "score": report.score,
         "passed": report.passed,
         "has_warnings": report.has_warnings,
+        "summary": report.summary,
         "findings": [
-            {"rule": f.rule, "level": f.level.value, "message": f.message}
+            {"rule": f.rule_code, "level": f.level.value, "message": f.message}
             for f in report.findings
         ],
     }
@@ -215,6 +265,41 @@ async def get_ticket(
     return await _enrich_ticket(ticket, db)
 
 
+@router.get("/{ticket_id}/approvals")
+async def get_ticket_approvals(
+    ticket_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """获取工单的审批进度（多级审批流）"""
+    result = await db.execute(select(Ticket).where(Ticket.id == ticket_id))
+    ticket = result.scalar_one_or_none()
+    if not ticket:
+        raise HTTPException(status_code=404, detail="工单不存在")
+
+    approvals_result = await db.execute(
+        select(TicketApproval)
+        .where(TicketApproval.ticket_id == ticket_id)
+        .order_by(TicketApproval.step_number)
+    )
+    approvals = approvals_result.scalars().all()
+
+    return [
+        {
+            "id": a.id,
+            "step_number": a.step_number,
+            "step_name": a.step_name,
+            "approver_role": a.approver_role,
+            "action": a.action,
+            "approver_id": a.approver_id,
+            "comment": a.comment,
+            "created_at": a.created_at.isoformat() if a.created_at else None,
+            "updated_at": a.updated_at.isoformat() if a.updated_at else None,
+        }
+        for a in approvals
+    ]
+
+
 @router.post("/{ticket_id}/review")
 async def review_ticket(
     ticket_id: int,
@@ -231,8 +316,56 @@ async def review_ticket(
         raise HTTPException(status_code=400, detail=f"工单状态为 {ticket.status}，无法审核")
 
     if data.action == "approve":
-        ticket.status = "approved"
+        # Check for pending workflow steps (multi-step approval)
+        pending_approvals_result = await db.execute(
+            select(TicketApproval)
+            .where(TicketApproval.ticket_id == ticket_id)
+            .order_by(TicketApproval.step_number)
+        )
+        pending_approvals = pending_approvals_result.scalars().all()
+
+        if pending_approvals:
+            # Multi-step workflow: find the current pending step
+            current_step = None
+            for pa in pending_approvals:
+                if pa.action == "pending":
+                    current_step = pa
+                    break
+
+            if current_step:
+                # Approve current step
+                current_step.action = "approved"
+                current_step.approver_id = current_user.id
+                current_step.comment = data.comment or "审批通过"
+
+                # Check if there's a next pending step
+                has_next = any(
+                    pa.action == "pending"
+                    for pa in pending_approvals
+                    if pa.id != current_step.id
+                )
+                if has_next:
+                    ticket.status = "reviewing"
+                else:
+                    ticket.status = "approved"
+            else:
+                # All steps already processed
+                ticket.status = "approved"
+        else:
+            # No workflow steps — single-step approval (backward compat)
+            ticket.status = "approved"
     else:
+        # Reject: reject the whole ticket immediately
+        # Mark all pending approval steps as rejected
+        pending_approvals_result = await db.execute(
+            select(TicketApproval)
+            .where(TicketApproval.ticket_id == ticket_id, TicketApproval.action == "pending")
+        )
+        for pa in pending_approvals_result.scalars().all():
+            pa.action = "rejected"
+            pa.approver_id = current_user.id
+            pa.comment = data.comment or "审批拒绝"
+
         ticket.status = "rejected"
 
     ticket.approver_id = current_user.id
@@ -419,6 +552,24 @@ async def execute_ticket(
     db: AsyncSession = Depends(get_db),
     current_user: User = Depends(require_role("admin", "dba")),
 ):
+    """执行工单。
+
+    使用多数据库工单执行引擎，按 db_type 自动分发到对应执行器。
+    支持 MySQL、PostgreSQL、Redis、MongoDB。
+    执行流程包括：语法预校验、执行前快照、执行变更、执行后验证、生成报告。
+
+    Args:
+        ticket_id: 工单 ID
+        db: 数据库会话
+        current_user: 当前用户（需要 admin 或 dba 角色）
+
+    Returns:
+        执行后的工单详情，包含执行报告
+
+    Raises:
+        HTTPException: 工单不存在、状态不正确、数据源不存在等
+    """
+    # 1. 查询工单
     result = await db.execute(select(Ticket).where(Ticket.id == ticket_id))
     ticket = result.scalar_one_or_none()
     if not ticket:
@@ -427,27 +578,170 @@ async def execute_ticket(
     if ticket.status != "approved":
         raise HTTPException(status_code=400, detail="只能执行已审批的工单")
 
+    # 2. 查询数据源实例
+    inst_result = await db.execute(
+        select(Instance).where(Instance.id == ticket.instance_id)
+    )
+    instance = inst_result.scalar_one_or_none()
+    if not instance:
+        raise HTTPException(status_code=400, detail="数据源不存在")
+
+    # 3. 更新工单状态为执行中
     ticket.status = "executing"
     await db.commit()
+    await db.refresh(ticket)
+
+    # 4. 使用新的多数据库执行引擎
+    from app.services.ticket_executor import TicketExecutor
+    executor = TicketExecutor()
 
     try:
-        inst_result = await db.execute(select(Instance).where(Instance.id == ticket.instance_id))
-        instance = inst_result.scalar_one_or_none()
-        if not instance:
-            raise HTTPException(status_code=400, detail="数据源不存在")
-
-        from app.services.db_connector import DBConnector
+        # 解密密码
         instance.password = decrypt_password(instance.password)
-        exec_result = await DBConnector.execute_dml(instance, ticket.target_database, ticket.sql_content)
 
-        ticket.status = "success"
-        ticket.execute_result = exec_result
+        # 执行工单
+        report = await executor.execute(ticket, instance)
+
+        # 5. 更新工单状态和结果
+        if report.status == "success":
+            ticket.status = "success"
+        else:
+            ticket.status = "failed"
+
+        ticket.execute_result = report.to_json()
         ticket.execute_time = datetime.now(timezone.utc)
+
+        # 6. 记录执行日志
+        log_detail = (
+            f"执行结果: {report.status}, 影响行数: {report.affected_rows}, "
+            f"耗时: {report.duration_ms}ms, 验证通过: {report.validation_passed}"
+        )
+        if report.errors:
+            log_detail += f", 错误: {'; '.join(report.errors[:3])}"
+
+        await AuditLogService.log(
+            db,
+            user_id=current_user.id,
+            action="EXECUTE_TICKET",
+            target_type="ticket",
+            target_id=ticket.id,
+            detail=log_detail,
+        )
+
+        # 7. 发送通知（执行完成后）
+        try:
+            await _send_execution_notification(db, ticket, report)
+        except Exception:
+            # 通知失败不应影响执行结果
+            pass
+
     except Exception as e:
         ticket.status = "failed"
         ticket.execute_result = str(e)
         ticket.execute_time = datetime.now(timezone.utc)
 
+        await AuditLogService.log(
+            db,
+            user_id=current_user.id,
+            action="EXECUTE_TICKET",
+            target_type="ticket",
+            target_id=ticket.id,
+            detail=f"执行异常: {type(e).__name__}: {e}",
+        )
+
+    # 8. 提交事务并返回
     await db.commit()
     await db.refresh(ticket)
-    return await _enrich_ticket(ticket, db)
+
+    # 构建响应
+    ticket_data = await _enrich_ticket(ticket, db)
+
+    # 尝试解析执行报告
+    try:
+        import json
+        exec_report = json.loads(ticket.execute_result) if ticket.execute_result else {}
+        if isinstance(exec_report, dict):
+            from app.schemas.execution import ExecutionReportResponse
+            ticket_data["execution_report"] = ExecutionReportResponse(
+                status=exec_report.get("status", "unknown"),
+                affected_rows=exec_report.get("affected_rows", 0),
+                duration_ms=exec_report.get("duration_ms", 0),
+                validation_passed=exec_report.get("validation_passed", False),
+                warnings=exec_report.get("warnings", []),
+                errors=exec_report.get("errors", []),
+                raw_output=exec_report.get("raw_output"),
+                before_snapshot_summary=exec_report.get("before_snapshot"),
+                after_snapshot_summary=exec_report.get("after_snapshot"),
+            )
+    except Exception:
+        # 如果解析失败，不返回结构化报告
+        ticket_data["execution_report"] = None
+
+    return ticket_data
+
+
+async def _send_execution_notification(
+    db: AsyncSession, ticket: Ticket, report: "ExecutionReport"
+) -> None:
+    """发送工单执行完成通知。
+
+    查询已启用的通知渠道并发送执行结果通知。
+
+    Args:
+        db: 数据库会话
+        ticket: 工单对象
+        report: 执行报告
+    """
+    from app.models.notification import NotificationChannel
+    from app.services.notification_sender import NotificationSender
+
+    result = await db.execute(
+        select(NotificationChannel).where(NotificationChannel.is_enabled == True)
+    )
+    channels = result.scalars().all()
+
+    if not channels:
+        return
+
+    sender = NotificationSender()
+
+    # 确定事件类型
+    if report.status == "success":
+        event_type = "ticket_executed"
+    else:
+        event_type = "ticket_failed"
+
+    # 构建 SQL 预览（前 100 字符）
+    sql_preview = ticket.sql_content[:100] + "..." if len(ticket.sql_content) > 100 else ticket.sql_content
+
+    # 获取申请人信息
+    applicant_name = "未知"
+    if ticket.applicant_id:
+        result = await db.execute(
+            select(User).where(User.id == ticket.applicant_id)
+        )
+        applicant = result.scalar_one_or_none()
+        if applicant:
+            applicant_name = applicant.display_name
+
+    notification_data = {
+        "title": ticket.title,
+        "ticket_id": ticket.id,
+        "db_type": ticket.db_type,
+        "target_database": ticket.target_database,
+        "sql_preview": sql_preview,
+        "priority": ticket.priority,
+        "applicant": applicant_name,
+        "result": report.status,
+        "duration": report.duration_ms,
+        "affected_rows": report.affected_rows,
+        "error": "; ".join(report.errors[:3]) if report.errors else "",
+        "operator": "dba",  # 执行人角色
+    }
+
+    await sender.notify_ticket_event(
+        ticket_id=ticket.id,
+        event_type=event_type,
+        channels=channels,
+        ticket_data=notification_data,
+    )

backend/app/api/v1/endpoints/workflow.py

@@ -0,0 +1,244 @@
+"""审批流程管理 API。"""
+from typing import List, Optional
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.core.database import get_db
+from app.core.security import require_role
+from app.models.user import User
+from app.models.workflow import WorkflowDefinition, WorkflowStep
+from app.schemas.workflow import (
+    WorkflowCreate, WorkflowUpdate, WorkflowResponse,
+    WorkflowStepCreate,
+)
+
+router = APIRouter(prefix="/workflows", tags=["审批流程"])
+
+
+def _serialize_step(step: WorkflowStep) -> dict:
+    return {
+        "id": step.id,
+        "workflow_id": step.workflow_id,
+        "step_number": step.step_number,
+        "name": step.name,
+        "step_type": step.step_type,
+        "approver_role": step.approver_role,
+        "approver_ids": step.approver_ids,
+        "is_required": step.is_required,
+        "timeout_hours": step.timeout_hours,
+        "auto_approve_on_timeout": step.auto_approve_on_timeout,
+    }
+
+
+def _serialize_workflow(wf: WorkflowDefinition) -> dict:
+    return {
+        "id": wf.id,
+        "name": wf.name,
+        "description": wf.description,
+        "is_default": wf.is_default,
+        "is_active": wf.is_active,
+        "created_at": wf.created_at,
+        "steps": [_serialize_step(s) for s in wf.steps],
+    }
+
+
+@router.get("", response_model=List[WorkflowResponse])
+async def list_workflows(
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    result = await db.execute(
+        select(WorkflowDefinition)
+        .where(WorkflowDefinition.is_active == True)
+        .options(selectinload(WorkflowDefinition.steps))
+    )
+    workflows = result.scalars().unique().all()
+    return [_serialize_workflow(wf) for wf in workflows]
+
+
+@router.post("", response_model=WorkflowResponse, status_code=201)
+async def create_workflow(
+    data: WorkflowCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin")),
+):
+    # 如果设为默认，取消其他默认
+    if data.is_default:
+        from sqlalchemy import update as sa_update
+        await db.execute(
+            sa_update(WorkflowDefinition)
+            .where(WorkflowDefinition.is_default == True)
+            .values(is_default=False)
+        )
+
+    workflow = WorkflowDefinition(
+        name=data.name,
+        description=data.description,
+        is_default=data.is_default,
+    )
+    db.add(workflow)
+    await db.flush()
+
+    for step_data in data.steps:
+        step = WorkflowStep(
+            workflow_id=workflow.id,
+            step_number=step_data.step_number,
+            name=step_data.name,
+            step_type=step_data.step_type,
+            approver_role=step_data.approver_role,
+            approver_ids=step_data.approver_ids,
+            is_required=step_data.is_required,
+            timeout_hours=step_data.timeout_hours,
+            auto_approve_on_timeout=step_data.auto_approve_on_timeout,
+        )
+        db.add(step)
+
+    await db.commit()
+    # Reload with steps
+    result = await db.execute(
+        select(WorkflowDefinition)
+        .where(WorkflowDefinition.id == workflow.id)
+        .options(selectinload(WorkflowDefinition.steps))
+    )
+    workflow = result.scalar_one()
+    return _serialize_workflow(workflow)
+
+
+@router.get("/{workflow_id}", response_model=WorkflowResponse)
+async def get_workflow(
+    workflow_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin", "dba")),
+):
+    result = await db.execute(
+        select(WorkflowDefinition)
+        .where(WorkflowDefinition.id == workflow_id)
+        .options(selectinload(WorkflowDefinition.steps))
+    )
+    workflow = result.scalar_one_or_none()
+    if not workflow:
+        raise HTTPException(status_code=404, detail="审批流程不存在")
+    return _serialize_workflow(workflow)
+
+
+@router.put("/{workflow_id}", response_model=WorkflowResponse)
+async def update_workflow(
+    workflow_id: int,
+    data: WorkflowUpdate,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin")),
+):
+    result = await db.execute(
+        select(WorkflowDefinition)
+        .where(WorkflowDefinition.id == workflow_id)
+        .options(selectinload(WorkflowDefinition.steps))
+    )
+    workflow = result.scalar_one_or_none()
+    if not workflow:
+        raise HTTPException(status_code=404, detail="审批流程不存在")
+
+    if data.name is not None:
+        workflow.name = data.name
+    if data.description is not None:
+        workflow.description = data.description
+    if data.is_active is not None:
+        workflow.is_active = data.is_active
+
+    await db.commit()
+    await db.refresh(workflow)
+    return _serialize_workflow(workflow)
+
+
+@router.delete("/{workflow_id}")
+async def delete_workflow(
+    workflow_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin")),
+):
+    result = await db.execute(
+        select(WorkflowDefinition).where(WorkflowDefinition.id == workflow_id)
+    )
+    workflow = result.scalar_one_or_none()
+    if not workflow:
+        raise HTTPException(status_code=404, detail="审批流程不存在")
+
+    workflow.is_active = False
+    await db.commit()
+    return {"message": "审批流程已删除"}
+
+
+@router.post("/{workflow_id}/set-default")
+async def set_default_workflow(
+    workflow_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin")),
+):
+    """设为默认审批流程。"""
+    # 取消所有默认
+    result = await db.execute(select(WorkflowDefinition))
+    all_workflows = result.scalars().all()
+    for wf in all_workflows:
+        wf.is_default = False
+
+    # 设置指定流程为默认
+    result = await db.execute(
+        select(WorkflowDefinition).where(WorkflowDefinition.id == workflow_id)
+    )
+    workflow = result.scalar_one_or_none()
+    if not workflow:
+        raise HTTPException(status_code=404, detail="审批流程不存在")
+
+    workflow.is_default = True
+    await db.commit()
+    return {"message": "审批流程 '{}' 已设为默认".format(workflow.name)}
+
+
+@router.post("/{workflow_id}/clone", response_model=WorkflowResponse)
+async def clone_workflow(
+    workflow_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_role("admin")),
+):
+    """克隆审批流程。"""
+    result = await db.execute(
+        select(WorkflowDefinition)
+        .where(WorkflowDefinition.id == workflow_id)
+        .options(selectinload(WorkflowDefinition.steps))
+    )
+    source = result.scalar_one_or_none()
+    if not source:
+        raise HTTPException(status_code=404, detail="审批流程不存在")
+
+    new_workflow = WorkflowDefinition(
+        name="{} (副本)".format(source.name),
+        description=source.description,
+        is_default=False,
+    )
+    db.add(new_workflow)
+    await db.flush()
+
+    for step in source.steps:
+        new_step = WorkflowStep(
+            workflow_id=new_workflow.id,
+            step_number=step.step_number,
+            name=step.name,
+            step_type=step.step_type,
+            approver_role=step.approver_role,
+            approver_ids=step.approver_ids,
+            is_required=step.is_required,
+            timeout_hours=step.timeout_hours,
+            auto_approve_on_timeout=step.auto_approve_on_timeout,
+        )
+        db.add(new_step)
+
+    await db.commit()
+    # Reload with steps
+    result = await db.execute(
+        select(WorkflowDefinition)
+        .where(WorkflowDefinition.id == new_workflow.id)
+        .options(selectinload(WorkflowDefinition.steps))
+    )
+    new_workflow = result.scalar_one()
+    return _serialize_workflow(new_workflow)