opsxsql/数据审核平台菜单优化/data-audit-platform/backend/main.py

"""
数据审核平台 - FastAPI 后端服务
支持 MySQL、Redis 等数据库的 SQL 审核、审批流程管理
"""

from fastapi import FastAPI, HTTPException, Depends, Query, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any, Literal
from datetime import datetime, timedelta
from enum import Enum
import uvicorn
import json

# 初始化 FastAPI 应用
app = FastAPI(
    title="数据审核平台 API",
    description="支持 MySQL、Redis 的数据库变更管理和 SQL 审核平台",
    version="1.0.0"
)

# CORS 配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ==================== 枚举类型定义 ====================

class DatabaseType(str, Enum):
    MYSQL = "mysql"
    REDIS = "redis"
    POSTGRESQL = "postgresql"
    MONGODB = "mongodb"

class InstanceStatus(str, Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"
    ERROR = "error"

class WorkflowStatus(str, Enum):
    PENDING = "pending"           # 待审核
    REVIEWING = "reviewing"       # 审核中
    APPROVED = "approved"         # 已通过
    REJECTED = "rejected"         # 已拒绝
    EXECUTING = "executing"       # 执行中
    SUCCESS = "success"           # 执行成功
    FAILED = "failed"             # 执行失败
    ROLLBACK = "rollback"         # 已回滚

class WorkflowType(str, Enum):
    DDL = "ddl"                   # 结构变更
    DML = "dml"                   # 数据变更
    QUERY = "query"               # 查询权限

class SQLRiskLevel(str, Enum):
    HIGH = "high"                 # 高风险
    MEDIUM = "medium"             # 中风险
    LOW = "low"                   # 低风险
    SAFE = "safe"                 # 安全

class UserRole(str, Enum):
    ADMIN = "admin"               # 管理员
    DBA = "dba"                   # DBA
    DEVELOPER = "developer"       # 开发者
    AUDITOR = "auditor"           # 审计员

# ==================== 数据模型定义 ====================

class User(BaseModel):
    id: int
    username: str
    email: str
    role: UserRole
    is_active: bool = True
    created_at: datetime
    avatar: Optional[str] = None
    department: Optional[str] = None
    phone: Optional[str] = None

class UserLogin(BaseModel):
    username: str
    password: str

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"
    user: User

class DatabaseInstance(BaseModel):
    id: int
    name: str
    db_type: DatabaseType
    host: str
    port: int
    database: Optional[str] = None
    username: str
    password: str = Field(..., description="数据库密码")
    environment: Literal["dev", "test", "staging", "prod"] = "dev"
    status: InstanceStatus = InstanceStatus.ACTIVE
    description: Optional[str] = None
    created_at: datetime
    updated_at: datetime
    created_by: int
    max_connections: int = 10
    connect_timeout: int = 30
    redis_mode: Optional[Literal["standalone", "sentinel", "cluster"]] = None

class DatabaseInstanceCreate(BaseModel):
    name: str
    db_type: DatabaseType
    host: str
    port: int
    database: Optional[str] = None
    username: str
    password: str
    environment: Literal["dev", "test", "staging", "prod"] = "dev"
    description: Optional[str] = None
    max_connections: int = 10
    redis_mode: Optional[Literal["standalone", "sentinel", "cluster"]] = None

class SQLAuditRule(BaseModel):
    id: int
    name: str
    description: str
    rule_type: Literal["syntax", "security", "performance", "standard"]
    risk_level: SQLRiskLevel
    enabled: bool = True
    config: Dict[str, Any] = {}

class SQLAuditResult(BaseModel):
    rule_id: int
    rule_name: str
    risk_level: SQLRiskLevel
    message: str
    line_number: Optional[int] = None
    suggestion: Optional[str] = None

class SQLWorkflow(BaseModel):
    id: int
    title: str
    description: Optional[str] = None
    workflow_type: WorkflowType
    status: WorkflowStatus
    instance_id: int
    instance_name: Optional[str] = None
    database: Optional[str] = None
    sql_content: str
    sql_results: List[SQLAuditResult] = []
    risk_level: SQLRiskLevel = SQLRiskLevel.SAFE
    risk_score: int = 0
    submitter_id: int
    submitter_name: Optional[str] = None
    reviewer_id: Optional[int] = None
    reviewer_name: Optional[str] = None
    executor_id: Optional[int] = None
    executor_name: Optional[str] = None
    review_comment: Optional[str] = None
    execute_result: Optional[str] = None
    rollback_sql: Optional[str] = None
    created_at: datetime
    updated_at: datetime
    executed_at: Optional[datetime] = None
    current_step: int = 1
    total_steps: int = 2
    approvers: List[Dict[str, Any]] = []

class SQLWorkflowCreate(BaseModel):
    title: str
    description: Optional[str] = None
    workflow_type: WorkflowType
    instance_id: int
    database: Optional[str] = None
    sql_content: str

class SQLWorkflowReview(BaseModel):
    workflow_id: int
    action: Literal["approve", "reject"]
    comment: Optional[str] = None

class SQLQueryRequest(BaseModel):
    instance_id: int
    database: Optional[str] = None
    sql: str
    limit: int = 100

class SQLQueryResult(BaseModel):
    columns: List[str]
    rows: List[List[Any]]
    total_rows: int
    execution_time: float
    sql: str

class DataDictionaryTable(BaseModel):
    name: str
    engine: Optional[str] = None
    rows: Optional[int] = None
    data_length: Optional[int] = None
    comment: Optional[str] = None
    create_time: Optional[datetime] = None

class DataDictionaryColumn(BaseModel):
    name: str
    data_type: str
    nullable: bool
    default: Optional[str] = None
    comment: Optional[str] = None
    is_primary: bool = False
    is_index: bool = False

class DataDictionaryIndex(BaseModel):
    name: str
    columns: List[str]
    unique: bool
    index_type: str

class DashboardStats(BaseModel):
    total_instances: int
    total_workflows: int
    pending_workflows: int
    today_workflows: int
    risk_distribution: Dict[str, int]
    recent_workflows: List[SQLWorkflow]
    db_type_distribution: Dict[str, int]

class SystemConfig(BaseModel):
    key: str
    value: Any
    description: Optional[str] = None

class Notification(BaseModel):
    id: int
    title: str
    content: str
    type: Literal["info", "warning", "success", "error"]
    is_read: bool = False
    created_at: datetime
    user_id: int

# ==================== Mock 数据存储 ====================

mock_users = [
    User(id=1, username="admin", email="admin@example.com", role=UserRole.ADMIN, 
         created_at=datetime.now() - timedelta(days=30), avatar="https://api.dicebear.com/7.x/avataaars/svg?seed=admin",
         department="技术部", phone="13800138001"),
    User(id=2, username="dba", email="dba@example.com", role=UserRole.DBA,
         created_at=datetime.now() - timedelta(days=25), avatar="https://api.dicebear.com/7.x/avataaars/svg?seed=dba",
         department="运维部", phone="13800138002"),
    User(id=3, username="dev1", email="dev1@example.com", role=UserRole.DEVELOPER,
         created_at=datetime.now() - timedelta(days=20), avatar="https://api.dicebear.com/7.x/avataaars/svg?seed=dev1",
         department="研发部", phone="13800138003"),
    User(id=4, username="dev2", email="dev2@example.com", role=UserRole.DEVELOPER,
         created_at=datetime.now() - timedelta(days=15), avatar="https://api.dicebear.com/7.x/avataaars/svg?seed=dev2",
         department="研发部", phone="13800138004"),
    User(id=5, username="auditor", email="auditor@example.com", role=UserRole.AUDITOR,
         created_at=datetime.now() - timedelta(days=10), avatar="https://api.dicebear.com/7.x/avataaars/svg?seed=auditor",
         department="安全部", phone="13800138005"),
]

mock_instances = [
    DatabaseInstance(
        id=1, name="生产-MySQL主库", db_type=DatabaseType.MYSQL, host="mysql-prod.example.com", 
        port=3306, database="production", username="admin", password="******",
        environment="prod", status=InstanceStatus.ACTIVE, description="生产环境 MySQL 主库",
        created_at=datetime.now() - timedelta(days=30), updated_at=datetime.now() - timedelta(days=1),
        created_by=1, max_connections=100
    ),
    DatabaseInstance(
        id=2, name="测试-MySQL", db_type=DatabaseType.MYSQL, host="mysql-test.example.com",
        port=3306, database="test_db", username="test_user", password="******",
        environment="test", status=InstanceStatus.ACTIVE, description="测试环境 MySQL",
        created_at=datetime.now() - timedelta(days=25), updated_at=datetime.now() - timedelta(days=2),
        created_by=1, max_connections=50
    ),
    DatabaseInstance(
        id=3, name="生产-Redis", db_type=DatabaseType.REDIS, host="redis-prod.example.com",
        port=6379, username="", password="******",
        environment="prod", status=InstanceStatus.ACTIVE, description="生产环境 Redis 缓存",
        created_at=datetime.now() - timedelta(days=28), updated_at=datetime.now() - timedelta(days=1),
        created_by=1, redis_mode="standalone"
    ),
    DatabaseInstance(
        id=4, name="开发-MySQL", db_type=DatabaseType.MYSQL, host="mysql-dev.example.com",
        port=3306, database="dev_db", username="dev_user", password="******",
        environment="dev", status=InstanceStatus.ACTIVE, description="开发环境 MySQL",
        created_at=datetime.now() - timedelta(days=20), updated_at=datetime.now() - timedelta(days=3),
        created_by=1, max_connections=30
    ),
    DatabaseInstance(
        id=5, name="生产-PostgreSQL", db_type=DatabaseType.POSTGRESQL, host="pg-prod.example.com",
        port=5432, database="production", username="postgres", password="******",
        environment="prod", status=InstanceStatus.ACTIVE, description="生产环境 PostgreSQL",
        created_at=datetime.now() - timedelta(days=22), updated_at=datetime.now() - timedelta(days=1),
        created_by=1, max_connections=80
    ),
]

mock_audit_rules = [
    SQLAuditRule(id=1, name="禁止DELETE无WHERE条件", description="DELETE语句必须包含WHERE条件", 
                 rule_type="security", risk_level=SQLRiskLevel.HIGH, enabled=True),
    SQLAuditRule(id=2, name="禁止UPDATE无WHERE条件", description="UPDATE语句必须包含WHERE条件",
                 rule_type="security", risk_level=SQLRiskLevel.HIGH, enabled=True),
    SQLAuditRule(id=3, name="禁止DROP TABLE", description="禁止直接删除表",
                 rule_type="security", risk_level=SQLRiskLevel.HIGH, enabled=True),
    SQLAuditRule(id=4, name="禁止TRUNCATE TABLE", description="禁止直接清空表",
                 rule_type="security", risk_level=SQLRiskLevel.HIGH, enabled=True),
    SQLAuditRule(id=5, name="建议使用索引", description="大表查询建议使用索引",
                 rule_type="performance", risk_level=SQLRiskLevel.MEDIUM, enabled=True),
    SQLAuditRule(id=6, name="SELECT * 建议指定列", description="避免使用SELECT *",
                 rule_type="standard", risk_level=SQLRiskLevel.LOW, enabled=True),
    SQLAuditRule(id=7, name="表名建议使用小写", description="表名建议使用小写字母",
                 rule_type="standard", risk_level=SQLRiskLevel.LOW, enabled=True),
    SQLAuditRule(id=8, name="建议添加LIMIT", description="查询语句建议添加LIMIT限制",
                 rule_type="performance", risk_level=SQLRiskLevel.MEDIUM, enabled=True),
]

mock_workflows = [
    SQLWorkflow(
        id=1, title="添加用户表索引", description="为用户表的email字段添加唯一索引",
        workflow_type=WorkflowType.DDL, status=WorkflowStatus.SUCCESS,
        instance_id=1, instance_name="生产-MySQL主库", database="production",
        sql_content="ALTER TABLE users ADD UNIQUE INDEX idx_email (email);",
        sql_results=[],
        risk_level=SQLRiskLevel.SAFE, risk_score=10,
        submitter_id=3, submitter_name="dev1",
        reviewer_id=2, reviewer_name="dba",
        executor_id=2, executor_name="dba",
        review_comment="审核通过，可以执行",
        execute_result="执行成功，耗时 0.5s",
        rollback_sql="ALTER TABLE users DROP INDEX idx_email;",
        created_at=datetime.now() - timedelta(days=2),
        updated_at=datetime.now() - timedelta(days=1),
        executed_at=datetime.now() - timedelta(days=1),
        current_step=2, total_steps=2,
        approvers=[{"step": 1, "role": "dba", "status": "approved", "user": "dba"}]
    ),
    SQLWorkflow(
        id=2, title="修复订单数据", description="修复订单状态异常数据",
        workflow_type=WorkflowType.DML, status=WorkflowStatus.PENDING,
        instance_id=1, instance_name="生产-MySQL主库", database="production",
        sql_content="UPDATE orders SET status = 'completed' WHERE id = 10086;",
        sql_results=[SQLAuditResult(
            rule_id=8, rule_name="建议添加LIMIT", 
            risk_level=SQLRiskLevel.LOW,
            message="UPDATE语句建议添加LIMIT限制",
            suggestion="建议添加 LIMIT 1"
        )],
        risk_level=SQLRiskLevel.LOW, risk_score=25,
        submitter_id=3, submitter_name="dev1",
        created_at=datetime.now() - timedelta(hours=5),
        updated_at=datetime.now() - timedelta(hours=5),
        current_step=1, total_steps=2,
        approvers=[{"step": 1, "role": "dba", "status": "pending"}]
    ),
    SQLWorkflow(
        id=3, title="创建新表", description="创建商品分类表",
        workflow_type=WorkflowType.DDL, status=WorkflowStatus.APPROVED,
        instance_id=2, instance_name="测试-MySQL", database="test_db",
        sql_content="""CREATE TABLE categories (
    id INT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(100) NOT NULL,
    parent_id INT DEFAULT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);""",
        sql_results=[],
        risk_level=SQLRiskLevel.SAFE, risk_score=5,
        submitter_id=4, submitter_name="dev2",
        reviewer_id=2, reviewer_name="dba",
        review_comment="表结构设计合理",
        created_at=datetime.now() - timedelta(days=1),
        updated_at=datetime.now() - timedelta(hours=3),
        current_step=2, total_steps=2,
        approvers=[{"step": 1, "role": "dba", "status": "approved", "user": "dba"}]
    ),
    SQLWorkflow(
        id=4, title="批量更新用户状态", description="将过期用户状态更新为inactive",
        workflow_type=WorkflowType.DML, status=WorkflowStatus.REJECTED,
        instance_id=1, instance_name="生产-MySQL主库", database="production",
        sql_content="UPDATE users SET status = 'inactive' WHERE last_login < '2024-01-01';",
        sql_results=[
            SQLAuditResult(
                rule_id=8, rule_name="建议添加LIMIT",
                risk_level=SQLRiskLevel.MEDIUM,
                message="UPDATE语句建议添加LIMIT限制",
                line_number=1
            ),
            SQLAuditResult(
                rule_id=5, rule_name="建议使用索引",
                risk_level=SQLRiskLevel.MEDIUM,
                message="该表数据量较大，建议使用索引优化查询",
                suggestion="为 last_login 字段添加索引"
            )
        ],
        risk_level=SQLRiskLevel.MEDIUM, risk_score=55,
        submitter_id=3, submitter_name="dev1",
        reviewer_id=2, reviewer_name="dba",
        review_comment="影响范围过大，请分批执行",
        created_at=datetime.now() - timedelta(days=3),
        updated_at=datetime.now() - timedelta(days=2),
        current_step=1, total_steps=2,
        approvers=[{"step": 1, "role": "dba", "status": "rejected", "user": "dba"}]
    ),
    SQLWorkflow(
        id=5, title="添加订单表字段", description="添加物流单号字段",
        workflow_type=WorkflowType.DDL, status=WorkflowStatus.EXECUTING,
        instance_id=1, instance_name="生产-MySQL主库", database="production",
        sql_content="ALTER TABLE orders ADD COLUMN tracking_no VARCHAR(50) NULL COMMENT '物流单号';",
        sql_results=[],
        risk_level=SQLRiskLevel.LOW, risk_score=20,
        submitter_id=4, submitter_name="dev2",
        reviewer_id=2, reviewer_name="dba",
        executor_id=2, executor_name="dba",
        review_comment="审核通过",
        created_at=datetime.now() - timedelta(hours=2),
        updated_at=datetime.now() - timedelta(minutes=5),
        current_step=2, total_steps=2,
        approvers=[{"step": 1, "role": "dba", "status": "approved", "user": "dba"}]
    ),
]

mock_tables = [
    DataDictionaryTable(
        name="users", engine="InnoDB", rows=150000, data_length=52428800,
        comment="用户表", create_time=datetime.now() - timedelta(days=100)
    ),
    DataDictionaryTable(
        name="orders", engine="InnoDB", rows=500000, data_length=268435456,
        comment="订单表", create_time=datetime.now() - timedelta(days=90)
    ),
    DataDictionaryTable(
        name="products", engine="InnoDB", rows=5000, data_length=2097152,
        comment="商品表", create_time=datetime.now() - timedelta(days=95)
    ),
    DataDictionaryTable(
        name="categories", engine="InnoDB", rows=50, data_length=16384,
        comment="分类表", create_time=datetime.now() - timedelta(days=95)
    ),
]

mock_columns = {
    "users": [
        DataDictionaryColumn(name="id", data_type="bigint(20)", nullable=False, 
                           default=None, comment="主键ID", is_primary=True),
        DataDictionaryColumn(name="username", data_type="varchar(50)", nullable=False,
                           default=None, comment="用户名"),
        DataDictionaryColumn(name="email", data_type="varchar(100)", nullable=False,
                           default=None, comment="邮箱"),
        DataDictionaryColumn(name="password", data_type="varchar(255)", nullable=False,
                           default=None, comment="密码"),
        DataDictionaryColumn(name="status", data_type="tinyint(1)", nullable=False,
                           default="1", comment="状态：1-启用，0-禁用"),
        DataDictionaryColumn(name="created_at", data_type="datetime", nullable=False,
                           default="CURRENT_TIMESTAMP", comment="创建时间"),
        DataDictionaryColumn(name="updated_at", data_type="datetime", nullable=True,
                           default=None, comment="更新时间"),
    ],
    "orders": [
        DataDictionaryColumn(name="id", data_type="bigint(20)", nullable=False,
                           default=None, comment="主键ID", is_primary=True),
        DataDictionaryColumn(name="user_id", data_type="bigint(20)", nullable=False,
                           default=None, comment="用户ID"),
        DataDictionaryColumn(name="order_no", data_type="varchar(50)", nullable=False,
                           default=None, comment="订单号"),
        DataDictionaryColumn(name="total_amount", data_type="decimal(10,2)", nullable=False,
                           default="0.00", comment="订单总金额"),
        DataDictionaryColumn(name="status", data_type="varchar(20)", nullable=False,
                           default="pending", comment="订单状态"),
        DataDictionaryColumn(name="created_at", data_type="datetime", nullable=False,
                           default="CURRENT_TIMESTAMP", comment="创建时间"),
    ]
}

mock_indexes = {
    "users": [
        DataDictionaryIndex(name="PRIMARY", columns=["id"], unique=True, index_type="BTREE"),
        DataDictionaryIndex(name="idx_username", columns=["username"], unique=True, index_type="BTREE"),
        DataDictionaryIndex(name="idx_email", columns=["email"], unique=True, index_type="BTREE"),
    ],
    "orders": [
        DataDictionaryIndex(name="PRIMARY", columns=["id"], unique=True, index_type="BTREE"),
        DataDictionaryIndex(name="idx_user_id", columns=["user_id"], unique=False, index_type="BTREE"),
        DataDictionaryIndex(name="idx_order_no", columns=["order_no"], unique=True, index_type="BTREE"),
        DataDictionaryIndex(name="idx_created_at", columns=["created_at"], unique=False, index_type="BTREE"),
    ]
}

mock_notifications = [
    Notification(id=1, title="工单审核通过", content="您的工单《添加用户表索引》已审核通过",
                type="success", is_read=False, created_at=datetime.now() - timedelta(hours=2), user_id=3),
    Notification(id=2, title="工单被拒绝", content="您的工单《批量更新用户状态》被拒绝，原因：影响范围过大",
                type="warning", is_read=False, created_at=datetime.now() - timedelta(days=1), user_id=3),
    Notification(id=3, title="系统维护通知", content="系统将于今晚 02:00-04:00 进行维护升级",
                type="info", is_read=True, created_at=datetime.now() - timedelta(days=2), user_id=1),
]

mock_configs = [
    SystemConfig(key="audit_enabled", value=True, description="是否启用SQL审核"),
    SystemConfig(key="auto_execute_test", value=False, description="测试环境是否自动执行"),
    SystemConfig(key="require_dba_approval", value=True, description="是否需要DBA审批"),
    SystemConfig(key="max_sql_length", value=10000, description="最大SQL长度限制"),
    SystemConfig(key="query_timeout", value=30, description="查询超时时间(秒)"),
]

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")

def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    return mock_users[0]

@app.get("/")
async def root():
    return {"message": "数据审核平台 API", "version": "1.0.0", "docs": "/docs"}

@app.post("/api/auth/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = next((u for u in mock_users if u.username == form_data.username), None)
    if not user or form_data.password != "123456":
        raise HTTPException(status_code=401, detail="用户名或密码错误")
    return Token(access_token=f"mock_token_{user.id}", user=user)

@app.get("/api/auth/me", response_model=User)
async def get_me(current_user: User = Depends(get_current_user)):
    return current_user

@app.post("/api/auth/logout")
async def logout(current_user: User = Depends(get_current_user)):
    return {"message": "登出成功"}

@app.get("/api/users", response_model=List[User])
async def list_users(
    role: Optional[UserRole] = None,
    keyword: Optional[str] = None,
    current_user: User = Depends(get_current_user)
):
    result = mock_users
    if role:
        result = [u for u in result if u.role == role]
    if keyword:
        result = [u for u in result if keyword.lower() in u.username.lower() or 
                  keyword.lower() in u.email.lower()]
    return result

@app.get("/api/users/{user_id}", response_model=User)
async def get_user(user_id: int, current_user: User = Depends(get_current_user)):
    user = next((u for u in mock_users if u.id == user_id), None)
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user

@app.get("/api/instances", response_model=List[DatabaseInstance])
async def list_instances(
    db_type: Optional[DatabaseType] = None,
    environment: Optional[str] = None,
    status: Optional[InstanceStatus] = None,
    keyword: Optional[str] = None,
    current_user: User = Depends(get_current_user)
):
    result = mock_instances
    if db_type:
        result = [i for i in result if i.db_type == db_type]
    if environment:
        result = [i for i in result if i.environment == environment]
    if status:
        result = [i for i in result if i.status == status]
    if keyword:
        result = [i for i in result if keyword.lower() in i.name.lower()]
    return result

@app.post("/api/instances", response_model=DatabaseInstance)
async def create_instance(
    instance: DatabaseInstanceCreate,
    current_user: User = Depends(get_current_user)
):
    new_instance = DatabaseInstance(
        id=len(mock_instances) + 1,
        **instance.dict(),
        status=InstanceStatus.ACTIVE,
        created_at=datetime.now(),
        updated_at=datetime.now(),
        created_by=current_user.id
    )
    mock_instances.append(new_instance)
    return new_instance

@app.get("/api/instances/{instance_id}", response_model=DatabaseInstance)
async def get_instance(instance_id: int, current_user: User = Depends(get_current_user)):
    instance = next((i for i in mock_instances if i.id == instance_id), None)
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    return instance

@app.put("/api/instances/{instance_id}", response_model=DatabaseInstance)
async def update_instance(
    instance_id: int,
    instance_update: DatabaseInstanceCreate,
    current_user: User = Depends(get_current_user)
):
    instance = next((i for i in mock_instances if i.id == instance_id), None)
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    for key, value in instance_update.dict().items():
        setattr(instance, key, value)
    instance.updated_at = datetime.now()
    return instance

@app.delete("/api/instances/{instance_id}")
async def delete_instance(instance_id: int, current_user: User = Depends(get_current_user)):
    instance = next((i for i in mock_instances if i.id == instance_id), None)
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    mock_instances.remove(instance)
    return {"message": "删除成功"}

@app.post("/api/instances/{instance_id}/test")
async def test_connection(instance_id: int, current_user: User = Depends(get_current_user)):
    instance = next((i for i in mock_instances if i.id == instance_id), None)
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    return {"success": True, "message": "连接成功", "latency_ms": 12}

@app.get("/api/workflows", response_model=List[SQLWorkflow])
async def list_workflows(
    status: Optional[WorkflowStatus] = None,
    workflow_type: Optional[WorkflowType] = None,
    risk_level: Optional[SQLRiskLevel] = None,
    submitter_id: Optional[int] = None,
    keyword: Optional[str] = None,
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    current_user: User = Depends(get_current_user)
):
    result = mock_workflows
    if status:
        result = [w for w in result if w.status == status]
    if workflow_type:
        result = [w for w in result if w.workflow_type == workflow_type]
    if risk_level:
        result = [w for w in result if w.risk_level == risk_level]
    if submitter_id:
        result = [w for w in result if w.submitter_id == submitter_id]
    if keyword:
        result = [w for w in result if keyword.lower() in w.title.lower()]
    result.sort(key=lambda x: x.created_at, reverse=True)
    total = len(result)
    start = (page - 1) * page_size
    end = start + page_size
    return result[start:end]

@app.get("/api/workflows/statistics")
async def get_workflow_statistics(current_user: User = Depends(get_current_user)):
    total = len(mock_workflows)
    pending = len([w for w in mock_workflows if w.status == WorkflowStatus.PENDING])
    approved = len([w for w in mock_workflows if w.status == WorkflowStatus.APPROVED])
    rejected = len([w for w in mock_workflows if w.status == WorkflowStatus.REJECTED])
    success = len([w for w in mock_workflows if w.status == WorkflowStatus.SUCCESS])
    failed = len([w for w in mock_workflows if w.status == WorkflowStatus.FAILED])
    risk_distribution = {
        "high": len([w for w in mock_workflows if w.risk_level == SQLRiskLevel.HIGH]),
        "medium": len([w for w in mock_workflows if w.risk_level == SQLRiskLevel.MEDIUM]),
        "low": len([w for w in mock_workflows if w.risk_level == SQLRiskLevel.LOW]),
        "safe": len([w for w in mock_workflows if w.risk_level == SQLRiskLevel.SAFE]),
    }
    return {
        "total": total,
        "pending": pending,
        "approved": approved,
        "rejected": rejected,
        "success": success,
        "failed": failed,
        "risk_distribution": risk_distribution
    }

@app.post("/api/workflows", response_model=SQLWorkflow)
async def create_workflow(
    workflow: SQLWorkflowCreate,
    current_user: User = Depends(get_current_user)
):
    instance = next((i for i in mock_instances if i.id == workflow.instance_id), None)
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    audit_results = []
    risk_score = 0
    sql_upper = workflow.sql_content.upper()
    if "DELETE" in sql_upper and "WHERE" not in sql_upper:
        audit_results.append(SQLAuditResult(
            rule_id=1, rule_name="禁止DELETE无WHERE条件",
            risk_level=SQLRiskLevel.HIGH,
            message="DELETE语句缺少WHERE条件，将删除全表数据",
            suggestion="请添加WHERE条件限制删除范围"
        ))
        risk_score += 50
    if "UPDATE" in sql_upper and "WHERE" not in sql_upper:
        audit_results.append(SQLAuditResult(
            rule_id=2, rule_name="禁止UPDATE无WHERE条件",
            risk_level=SQLRiskLevel.HIGH,
            message="UPDATE语句缺少WHERE条件，将更新全表数据",
            suggestion="请添加WHERE条件限制更新范围"
        ))
        risk_score += 50
    if "SELECT *" in sql_upper:
        audit_results.append(SQLAuditResult(
            rule_id=6, rule_name="SELECT * 建议指定列",
            risk_level=SQLRiskLevel.LOW,
            message="建议使用具体的列名替代 *",
            suggestion="列出需要的字段名"
        ))
        risk_score += 10
    if risk_score >= 50:
        risk_level = SQLRiskLevel.HIGH
    elif risk_score >= 30:
        risk_level = SQLRiskLevel.MEDIUM
    elif risk_score >= 10:
        risk_level = SQLRiskLevel.LOW
    else:
        risk_level = SQLRiskLevel.SAFE
    new_workflow = SQLWorkflow(
        id=len(mock_workflows) + 1,
        title=workflow.title,
        description=workflow.description,
        workflow_type=workflow.workflow_type,
        status=WorkflowStatus.PENDING,
        instance_id=workflow.instance_id,
        instance_name=instance.name,
        database=workflow.database,
        sql_content=workflow.sql_content,
        sql_results=audit_results,
        risk_level=risk_level,
        risk_score=risk_score,
        submitter_id=current_user.id,
        submitter_name=current_user.username,
        created_at=datetime.now(),
        updated_at=datetime.now(),
        current_step=1,
        total_steps=2,
        approvers=[{"step": 1, "role": "dba", "status": "pending"}]
    )
    mock_workflows.append(new_workflow)
    return new_workflow

@app.get("/api/workflows/{workflow_id}", response_model=SQLWorkflow)
async def get_workflow(workflow_id: int, current_user: User = Depends(get_current_user)):
    workflow = next((w for w in mock_workflows if w.id == workflow_id), None)
    if not workflow:
        raise HTTPException(status_code=404, detail="工单不存在")
    return workflow

@app.post("/api/workflows/{workflow_id}/review")
async def review_workflow(
    workflow_id: int,
    review: SQLWorkflowReview,
    current_user: User = Depends(get_current_user)
):
    workflow = next((w for w in mock_workflows if w.id == workflow_id), None)
    if not workflow:
        raise HTTPException(status_code=404, detail="工单不存在")
    if workflow.status != WorkflowStatus.PENDING and workflow.status != WorkflowStatus.REVIEWING:
        raise HTTPException(status_code=400, detail="当前状态不允许审核")
    if review.action == "approve":
        workflow.status = WorkflowStatus.APPROVED
        workflow.reviewer_id = current_user.id
        workflow.reviewer_name = current_user.username
        workflow.review_comment = review.comment
        workflow.current_step = 2
    else:
        workflow.status = WorkflowStatus.REJECTED
        workflow.reviewer_id = current_user.id
        workflow.reviewer_name = current_user.username
        workflow.review_comment = review.comment
    workflow.updated_at = datetime.now()
    return workflow

@app.post("/api/workflows/{workflow_id}/execute")
async def execute_workflow(
    workflow_id: int,
    current_user: User = Depends(get_current_user)
):
    workflow = next((w for w in mock_workflows if w.id == workflow_id), None)
    if not workflow:
        raise HTTPException(status_code=404, detail="工单不存在")
    if workflow.status != WorkflowStatus.APPROVED:
        raise HTTPException(status_code=400, detail="工单未通过审核，无法执行")
    workflow.status = WorkflowStatus.EXECUTING
    workflow.executor_id = current_user.id
    workflow.executor_name = current_user.username
    workflow.updated_at = datetime.now()
    workflow.status = WorkflowStatus.SUCCESS
    workflow.execute_result = "执行成功，耗时 0.8s，影响行数: 1"
    workflow.executed_at = datetime.now()
    if workflow.workflow_type == WorkflowType.DDL:
        workflow.rollback_sql = f"-- 回滚语句\n-- {workflow.sql_content}"
    return workflow

@app.post("/api/workflows/{workflow_id}/rollback")
async def rollback_workflow(
    workflow_id: int,
    current_user: User = Depends(get_current_user)
):
    workflow = next((w for w in mock_workflows if w.id == workflow_id), None)
    if not workflow:
        raise HTTPException(status_code=404, detail="工单不存在")
    if workflow.status != WorkflowStatus.SUCCESS:
        raise HTTPException(status_code=400, detail="只有执行成功的工单才能回滚")
    workflow.status = WorkflowStatus.ROLLBACK
    workflow.updated_at = datetime.now()
    return workflow

@app.get("/api/audit-rules", response_model=List[SQLAuditRule])
async def list_audit_rules(
    rule_type: Optional[str] = None,
    enabled: Optional[bool] = None,
    current_user: User = Depends(get_current_user)
):
    result = mock_audit_rules
    if rule_type:
        result = [r for r in result if r.rule_type == rule_type]
    if enabled is not None:
        result = [r for r in result if r.enabled == enabled]
    return result

@app.put("/api/audit-rules/{rule_id}")
async def update_audit_rule(
    rule_id: int,
    rule_update: dict,
    current_user: User = Depends(get_current_user)
):
    rule = next((r for r in mock_audit_rules if r.id == rule_id), None)
    if not rule:
        raise HTTPException(status_code=404, detail="规则不存在")
    for key, value in rule_update.items():
        if hasattr(rule, key):
            setattr(rule, key, value)
    return rule

@app.post("/api/query", response_model=SQLQueryResult)
async def execute_query(
    query: SQLQueryRequest,
    current_user: User = Depends(get_current_user)
):
    instance = next((i for i in mock_instances if i.id == query.instance_id), None)
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    sql_upper = query.sql.upper().strip()
    if not sql_upper.startswith("SELECT") and not sql_upper.startswith("SHOW"):
        raise HTTPException(status_code=400, detail="只允许执行 SELECT/SHOW 查询")
    if "users" in query.sql.lower():
        columns = ["id", "username", "email", "status", "created_at"]
        rows = [
            [1, "user1", "user1@example.com", "active", "2024-01-15 10:30:00"],
            [2, "user2", "user2@example.com", "active", "2024-01-16 11:00:00"],
            [3, "user3", "user3@example.com", "inactive", "2024-01-17 09:15:00"],
        ]
    elif "orders" in query.sql.lower():
        columns = ["id", "order_no", "user_id", "total_amount", "status"]
        rows = [
            [1, "ORD2024001", 1, 199.99, "completed"],
            [2, "ORD2024002", 2, 299.99, "pending"],
            [3, "ORD2024003", 1, 99.99, "completed"],
        ]
    else:
        columns = ["result"]
        rows = [["Query executed successfully"]]
    return SQLQueryResult(
        columns=columns,
        rows=rows,
        total_rows=len(rows),
        execution_time=15.5,
        sql=query.sql
    )

@app.get("/api/dictionary/instances/{instance_id}/databases")
async def list_databases(
    instance_id: int,
    current_user: User = Depends(get_current_user)
):
    instance = next((i for i in mock_instances if i.id == instance_id), None)
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    return [
        {"name": "information_schema", "charset": "utf8", "collation": "utf8_general_ci"},
        {"name": "mysql", "charset": "utf8", "collation": "utf8_general_ci"},
        {"name": instance.database or "production", "charset": "utf8mb4", "collation": "utf8mb4_unicode_ci"},
        {"name": "test_db", "charset": "utf8mb4", "collation": "utf8mb4_unicode_ci"},
    ]

@app.get("/api/dictionary/instances/{instance_id}/tables")
async def list_tables(
    instance_id: int,
    database: str,
    current_user: User = Depends(get_current_user)
):
    instance = next((i for i in mock_instances if i.id == instance_id), None)
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    return mock_tables

@app.get("/api/dictionary/instances/{instance_id}/tables/{table_name}")
async def get_table_info(
    instance_id: int,
    database: str,
    table_name: str,
    current_user: User = Depends(get_current_user)
):
    instance = next((i for i in mock_instances if i.id == instance_id), None)
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    table = next((t for t in mock_tables if t.name == table_name), None)
    if not table:
        raise HTTPException(status_code=404, detail="表不存在")
    columns = mock_columns.get(table_name, [])
    indexes = mock_indexes.get(table_name, [])
    return {
        "table": table,
        "columns": columns,
        "indexes": indexes
    }

@app.get("/api/dashboard/stats", response_model=DashboardStats)
async def get_dashboard_stats(current_user: User = Depends(get_current_user)):
    total_instances = len(mock_instances)
    total_workflows = len(mock_workflows)
    pending_workflows = len([w for w in mock_workflows if w.status == WorkflowStatus.PENDING])
    today_workflows = len([w for w in mock_workflows if w.created_at.date() == datetime.now().date()])
    risk_distribution = {
        "high": len([w for w in mock_workflows if w.risk_level == SQLRiskLevel.HIGH]),
        "medium": len([w for w in mock_workflows if w.risk_level == SQLRiskLevel.MEDIUM]),
        "low": len([w for w in mock_workflows if w.risk_level == SQLRiskLevel.LOW]),
        "safe": len([w for w in mock_workflows if w.risk_level == SQLRiskLevel.SAFE]),
    }
    db_type_distribution = {}
    for instance in mock_instances:
        db_type = instance.db_type.value
        db_type_distribution[db_type] = db_type_distribution.get(db_type, 0) + 1
    recent_workflows = sorted(mock_workflows, key=lambda x: x.created_at, reverse=True)[:5]
    return DashboardStats(
        total_instances=total_instances,
        total_workflows=total_workflows,
        pending_workflows=pending_workflows,
        today_workflows=today_workflows,
        risk_distribution=risk_distribution,
        recent_workflows=recent_workflows,
        db_type_distribution=db_type_distribution
    )

@app.get("/api/notifications")
async def list_notifications(
    is_read: Optional[bool] = None,
    current_user: User = Depends(get_current_user)
):
    result = [n for n in mock_notifications if n.user_id == current_user.id]
    if is_read is not None:
        result = [n for n in result if n.is_read == is_read]
    return sorted(result, key=lambda x: x.created_at, reverse=True)

@app.put("/api/notifications/{notification_id}/read")
async def mark_notification_read(
    notification_id: int,
    current_user: User = Depends(get_current_user)
):
    notification = next((n for n in mock_notifications if n.id == notification_id), None)
    if not notification:
        raise HTTPException(status_code=404, detail="通知不存在")
    notification.is_read = True
    return notification

@app.get("/api/notifications/unread-count")
async def get_unread_notification_count(current_user: User = Depends(get_current_user)):
    count = len([n for n in mock_notifications if n.user_id == current_user.id and not n.is_read])
    return {"count": count}

@app.get("/api/configs")
async def list_configs(current_user: User = Depends(get_current_user)):
    return mock_configs

@app.put("/api/configs/{key}")
async def update_config(
    key: str,
    value: dict,
    current_user: User = Depends(get_current_user)
):
    config = next((c for c in mock_configs if c.key == key), None)
    if not config:
        raise HTTPException(status_code=404, detail="配置不存在")
    config.value = value.get("value")
    return config

@app.get("/api/redis/{instance_id}/info")
async def get_redis_info(
    instance_id: int,
    current_user: User = Depends(get_current_user)
):
    instance = next((i for i in mock_instances if i.id == instance_id), None)
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if instance.db_type != DatabaseType.REDIS:
        raise HTTPException(status_code=400, detail="该实例不是 Redis")
    return {
        "version": "7.0.12",
        "mode": instance.redis_mode or "standalone",
        "uptime_in_seconds": 86400 * 30,
        "connected_clients": 50,
        "used_memory_human": "512M",
        "used_memory_peak_human": "1G",
        "total_keys": 100000,
        "keyspace_hits": 950000,
        "keyspace_misses": 50000,
        "hit_rate": 95.0,
        "databases": [
            {"db": 0, "keys": 50000, "expires": 10000},
            {"db": 1, "keys": 30000, "expires": 5000},
            {"db": 2, "keys": 20000, "expires": 2000},
        ]
    }

@app.post("/api/redis/{instance_id}/command")
async def execute_redis_command(
    instance_id: int,
    command: dict,
    current_user: User = Depends(get_current_user)
):
    instance = next((i for i in mock_instances if i.id == instance_id), None)
    if not instance:
        raise HTTPException(status_code=404, detail="实例不存在")
    if instance.db_type != DatabaseType.REDIS:
        raise HTTPException(status_code=400, detail="该实例不是 Redis")
    cmd = command.get("command", "").upper().strip()
    readonly_commands = ["GET", "MGET", "HGET", "HGETALL", "LRANGE", "ZRANGE", "KEYS", "SCAN", 
                        "INFO", "DBSIZE", "TTL", "TYPE", "EXISTS", "STRLEN", "HLEN", "LLEN", 
                        "SCARD", "ZCARD", "SMEMBERS"]
    cmd_name = cmd.split()[0]
    if cmd_name not in readonly_commands:
        raise HTTPException(status_code=400, detail=f"不允许执行命令: {cmd_name}")
    if "INFO" in cmd:
        return {"result": "# Server\nredis_version:7.0.12\nredis_mode:standalone"}
    elif "DBSIZE" in cmd:
        return {"result": 100000}
    elif "GET" in cmd:
        return {"result": "value_example"}
    elif "HGETALL" in cmd:
        return {"result": {"field1": "value1", "field2": "value2"}}
    else:
        return {"result": "OK"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)