Add V2 admin console

ai_knowledge_base_v2/apps/backend/app/services/admin_service.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from fastapi import HTTPException, status
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.core.config import get_settings
+from app.core.security import create_access_token, hash_password, verify_password
+from app.models.admin import Admin
+from app.models.chat import ChatMessage, ChatSession
+from app.models.knowledge import Knowledge
+from app.models.logs import AiRequestLog, OperationLog
+from app.models.user import User
+
+
+class AdminAuthService:
+    @classmethod
+    def login(cls, db: Session, username: str, password: str) -> dict:
+        cls.ensure_bootstrap_admin(db)
+        admin = db.scalar(select(Admin).where(Admin.username == username))
+        if admin is None or not verify_password(password, admin.password):
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="管理员账号或密码错误")
+        if admin.status != 1:
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="管理员已禁用")
+
+        admin.last_login_at = datetime.now(UTC).replace(tzinfo=None)
+        db.add(admin)
+        db.commit()
+        db.refresh(admin)
+
+        token, expired_at = create_access_token(str(admin.id), "admin")
+        return {"token": token, "expiredAt": expired_at, "admin": admin}
+
+    @staticmethod
+    def ensure_bootstrap_admin(db: Session) -> None:
+        has_admin = db.scalar(select(func.count(Admin.id))) or 0
+        if has_admin:
+            return
+        settings = get_settings()
+        admin = Admin(
+            username=settings.bootstrap_admin_username,
+            password=hash_password(settings.bootstrap_admin_password),
+            name=settings.bootstrap_admin_name,
+            status=1,
+        )
+        db.add(admin)
+        db.commit()
+
+
+class AdminDashboardService:
+    @staticmethod
+    def stats(db: Session) -> dict:
+        return {
+            "userCount": db.scalar(select(func.count(User.id)).where(User.is_deleted == 0)) or 0,
+            "sessionCount": db.scalar(select(func.count(ChatSession.id)).where(ChatSession.is_deleted == 0)) or 0,
+            "messageCount": db.scalar(select(func.count(ChatMessage.id))) or 0,
+            "aiRequestCount": db.scalar(select(func.count(AiRequestLog.id))) or 0,
+            "knowledgeCount": db.scalar(select(func.count(Knowledge.id))) or 0,
+        }
+
+
+class OperationLogService:
+    @staticmethod
+    def write(
+        db: Session,
+        *,
+        admin_id: int | None,
+        module: str,
+        action: str,
+        target_id: int | None = None,
+        result: str = "SUCCESS",
+    ) -> None:
+        db.add(
+            OperationLog(
+                admin_id=admin_id,
+                module=module,
+                action=action,
+                target_id=target_id,
+                result=result,
+            )
+        )