新增后台数据库备份与回滚

backend/app/api/router.py

@@ -1,6 +1,7 @@
 from fastapi import APIRouter
 
 from app.api.routes import (
+    admin_backups,
     admin_certificates,
     admin_dashboard,
     admin_exports,
@@ -25,4 +26,5 @@ api_router.include_router(admin_imports.router, prefix="/admin/import-batches",
 api_router.include_router(admin_exports.router, prefix="/admin/exports", tags=["admin-exports"])
 api_router.include_router(admin_logs.router, prefix="/admin/logs", tags=["admin-logs"])
 api_router.include_router(admin_pdf_cleanup.router, prefix="/admin/pdf-cleanup", tags=["admin-pdf-cleanup"])
+api_router.include_router(admin_backups.router, prefix="/admin/backups", tags=["admin-backups"])
 api_router.include_router(public.router, prefix="/public", tags=["public"])

backend/app/api/routes/admin_backups.py

@@ -0,0 +1,82 @@
+from fastapi import APIRouter, Body, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from app.api.deps import require_roles
+from app.db.session import SessionLocal, get_db
+from app.models import AdminUser
+from app.schemas.backup import (
+    CreateDatabaseBackupRequest,
+    DatabaseBackupOut,
+    RestoreDatabaseBackupOut,
+    RestoreDatabaseBackupRequest,
+)
+from app.services.db_backup import BackupError, create_database_backup, list_database_backups, restore_database_backup
+from app.services.logs import log_action
+
+router = APIRouter()
+
+
+@router.get("", response_model=list[DatabaseBackupOut])
+def list_backups(
+    _: AdminUser = Depends(require_roles("system_admin")),
+) -> list[dict[str, object]]:
+    return list_database_backups()
+
+
+@router.post("", response_model=DatabaseBackupOut)
+def create_backup(
+    payload: CreateDatabaseBackupRequest | None = Body(default=None),
+    db: Session = Depends(get_db),
+    current_admin: AdminUser = Depends(require_roles("system_admin")),
+) -> dict[str, object]:
+    try:
+        backup = create_database_backup(reason=payload.reason if payload else "manual")
+    except BackupError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
+
+    log_action(
+        db,
+        current_admin,
+        "create_database_backup",
+        "database_backup",
+        backup["filename"],
+        detail={
+            "filename": backup["filename"],
+            "database_type": backup["database_type"],
+            "size_label": backup["size_label"],
+            "reason": backup.get("reason"),
+        },
+    )
+    db.commit()
+    return backup
+
+
+@router.post("/{filename}/restore", response_model=RestoreDatabaseBackupOut)
+def restore_backup(
+    filename: str,
+    payload: RestoreDatabaseBackupRequest,
+    db: Session = Depends(get_db),
+    current_admin: AdminUser = Depends(require_roles("system_admin")),
+) -> dict[str, object]:
+    db.close()
+    try:
+        result = restore_database_backup(filename, payload.confirmation)
+    except BackupError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
+
+    with SessionLocal() as log_db:
+        log_action(
+            log_db,
+            current_admin,
+            "restore_database_backup",
+            "database_backup",
+            filename,
+            detail={
+                "filename": filename,
+                "restored_backup": result["restored_backup"],
+                "pre_restore_backup": result["pre_restore_backup"],
+            },
+        )
+        log_db.commit()
+
+    return result

backend/app/api/routes/admin_pdf_cleanup.py

@@ -1,7 +1,7 @@
 from fastapi import APIRouter, Depends, HTTPException, status
 from sqlalchemy.orm import Session
 
-from app.api.deps import get_current_admin
+from app.api.deps import get_current_admin, require_roles
 from app.db.session import get_db
 from app.models import AdminUser
 from app.services.pdf_cleanup import (
@@ -23,29 +23,18 @@ def get_pdf_stats(
 
 @router.post("/cleanup")
 def trigger_cleanup(
-    current_admin: AdminUser = Depends(get_current_admin),
+    _: AdminUser = Depends(require_roles("system_admin")),
 ) -> dict:
     """手动触发PDF清理"""
-    if current_admin.role_code != "admin":
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="只有管理员可以执行此操作",
-        )
     return cleanup_expired_pdfs()
 
 
 @router.post("/cleanup/{certificate_no}")
 def cleanup_single_certificate(
     certificate_no: str,
-    current_admin: AdminUser = Depends(get_current_admin),
+    _: AdminUser = Depends(require_roles("system_admin")),
 ) -> dict:
     """手动清理指定证书的PDF"""
-    if current_admin.role_code != "admin":
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="只有管理员可以执行此操作",
-        )
-
     success = manual_cleanup_certificate(certificate_no)
     if success:
         return {"message": f"证书 {certificate_no} 的PDF已清理"}