from typing import Optional, List, Dict, Any from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy import select, func, or_ 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.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.services.audit_engine import SQLAuditEngine from app.services.audit_log_service import AuditLogService import json router = APIRouter(prefix="/tickets", tags=["工单管理"]) async def _enrich_ticket(ticket: Ticket, db: AsyncSession) -> Dict[str, Any]: data = { "id": ticket.id, "title": ticket.title, "db_type": ticket.db_type, "instance_id": ticket.instance_id, "target_database": ticket.target_database, "sql_content": ticket.sql_content, "rollback_sql": ticket.rollback_sql, "status": ticket.status, "priority": ticket.priority, "applicant_id": ticket.applicant_id, "approver_id": ticket.approver_id, "reason": ticket.reason, "audit_result": ticket.audit_result, "execute_result": ticket.execute_result, "create_time": ticket.create_time, "review_time": ticket.review_time, "execute_time": ticket.execute_time, } # Resolve applicant name if ticket.applicant_id: r = await db.execute(select(User).where(User.id == ticket.applicant_id)) applicant = r.scalar_one_or_none() data["applicant_name"] = applicant.display_name if applicant else "未知" # Resolve approver name if ticket.approver_id: r = await db.execute(select(User).where(User.id == ticket.approver_id)) approver = r.scalar_one_or_none() data["approver_name"] = approver.display_name if approver else "未知" return data @router.get("") async def list_tickets( page: int = Query(1, ge=1), page_size: int = Query(10, ge=1, le=100), status: Optional[str] = None, db_type: Optional[str] = None, keyword: Optional[str] = None, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): stmt = select(Ticket) # developer 只能看到自己的工单 if current_user.role == "developer": stmt = stmt.where(Ticket.applicant_id == current_user.id) if status: stmt = stmt.where(Ticket.status == status) if db_type: stmt = stmt.where(Ticket.db_type == db_type) if keyword: stmt = stmt.where( or_( Ticket.title.contains(keyword), Ticket.sql_content.contains(keyword), ) ) total_stmt = select(func.count()).select_from(stmt.subquery()) total = (await db.execute(total_stmt)).scalar() stmt = stmt.order_by(Ticket.create_time.desc()).offset((page - 1) * page_size).limit(page_size) result = await db.execute(stmt) tickets = result.scalars().all() items = [] for t in tickets: items.append(await _enrich_ticket(t, db)) return {"total": total, "page": page, "page_size": page_size, "items": items} @router.post("", status_code=201) async def create_ticket( data: TicketCreate, db: AsyncSession = Depends(get_db), 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(): 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) ticket = Ticket( title=data.title, db_type=data.db_type, instance_id=data.instance_id, target_database=data.target_database, sql_content=data.sql_content, rollback_sql=data.rollback_sql, 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, ) db.add(ticket) await db.flush() # flush to get ticket.id for audit log # 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}", ) await db.commit() await db.refresh(ticket) result = await _enrich_ticket(ticket, db) result["audit_report"] = report.to_dict() return result @router.get("/statistics") async def ticket_statistics( db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): stmt = select(Ticket.status, func.count(Ticket.id)).group_by(Ticket.status) result = await db.execute(stmt) stats = {row[0]: row[1] for row in result.all()} total = sum(stats.values()) return { "total": total, "pending": stats.get("pending", 0), "reviewing": stats.get("reviewing", 0), "approved": stats.get("approved", 0), "rejected": stats.get("rejected", 0), "executing": stats.get("executing", 0), "success": stats.get("success", 0), "failed": stats.get("failed", 0), } @router.get("/{ticket_id}") async def get_ticket( 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="工单不存在") if current_user.role == "developer" and ticket.applicant_id != current_user.id: raise HTTPException(status_code=403, detail="无权查看该工单") return await _enrich_ticket(ticket, db) @router.post("/{ticket_id}/review") async def review_ticket( ticket_id: int, data: TicketReview, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_role("admin", "dba")), ): 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="工单不存在") if ticket.status not in ("pending", "reviewing"): raise HTTPException(status_code=400, detail=f"工单状态为 {ticket.status}，无法审核") if data.action == "approve": ticket.status = "approved" else: ticket.status = "rejected" ticket.approver_id = current_user.id ticket.review_time = datetime.now(timezone.utc) ticket.audit_result = data.comment await AuditLogService.log( db, user_id=current_user.id, action="REVIEW_TICKET", target_type="ticket", target_id=ticket.id, detail=f"审核动作: {data.action}, 备注: {data.comment}", ) await db.commit() await db.refresh(ticket) return await _enrich_ticket(ticket, db) @router.post("/{ticket_id}/execute") async def execute_ticket( ticket_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_role("admin", "dba")), ): 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="工单不存在") if ticket.status != "approved": raise HTTPException(status_code=400, detail="只能执行已审批的工单") ticket.status = "executing" await db.commit() 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.query_service import QueryService instance.password = decrypt_password(instance.password) exec_result = await QueryService.execute_dml(instance, ticket.target_database, ticket.sql_content) ticket.status = "success" ticket.execute_result = exec_result ticket.execute_time = datetime.now(timezone.utc) except Exception as e: ticket.status = "failed" ticket.execute_result = str(e) ticket.execute_time = datetime.now(timezone.utc) await db.commit() await db.refresh(ticket) return await _enrich_ticket(ticket, db)