Add operation log related interfaces (#92)

* Add operation log related interfaces

* Update to native ASGI middleware

* add the opera model class to the __init__.py

* Executable code collation

* Reply to the access middleware

* Using the request extension params in the login log

* Fix the whitelist list

* Fix username resolution

Author: wu-clan

backend/app/api/routers.py

@@ -11,6 +11,7 @@
 from backend.app.api.v1.api import router as api_router
 from backend.app.api.v1.config import router as config_router
 from backend.app.api.v1.login_log import router as login_log_router
+from backend.app.api.v1.opera_log import router as opera_log_router
 from backend.app.api.v1.task_demo import router as task_demo_router
 
 v1 = APIRouter(prefix='/v1')
@@ -24,4 +25,5 @@
 v1.include_router(api_router, prefix='/apis', tags=['API管理'])
 v1.include_router(config_router, prefix='/configs', tags=['系统配置'])
 v1.include_router(login_log_router, prefix='/login_logs', tags=['登录日志管理'])
+v1.include_router(opera_log_router, prefix='/opera_logs', tags=['操作日志管理'])
 v1.include_router(task_demo_router, prefix='/tasks', tags=['任务管理'])

backend/app/api/v1/opera_log.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from typing import Annotated
+
+from fastapi import APIRouter, Query
+
+from backend.app.common.casbin_rbac import DependsRBAC
+from backend.app.common.jwt import DependsJwtAuth
+from backend.app.common.pagination import PageDepends, paging_data
+from backend.app.common.response.response_schema import response_base
+from backend.app.database.db_mysql import CurrentSession
+from backend.app.schemas.opera_log import GetAllOperaLog
+from backend.app.services.opera_log_service import OperaLogService
+
+router = APIRouter()
+
+
+@router.get('', summary='（模糊条件）分页获取操作日志', dependencies=[DependsJwtAuth, PageDepends])
+async def get_all_opera_logs(
+        db: CurrentSession,
+        username: Annotated[str | None, Query()] = None,
+        status: Annotated[bool | None, Query()] = None,
+        ipaddr: Annotated[str | None, Query()] = None,
+):
+    log_select = await OperaLogService.get_select(username=username, status=status, ipaddr=ipaddr)
+    page_data = await paging_data(db, log_select, GetAllOperaLog)
+    return response_base.success(data=page_data)
+
+
+@router.delete('', summary='（批量）删除操作日志', dependencies=[DependsRBAC])
+async def delete_opera_log(pk: Annotated[list[int], Query(...)]):
+    count = await OperaLogService.delete(pk)
+    if count > 0:
+        return response_base.success()
+    return response_base.fail()
+
+
+@router.delete('/all', summary='清空操作日志', dependencies=[DependsRBAC])
+async def delete_all_opera_logs():
+    count = await OperaLogService.delete_all()
+    if count > 0:
+        return response_base.success()
+    return response_base.fail()

backend/app/core/conf.py

@@ -96,10 +96,18 @@ def validator_api_url(cls, values):
     # Casbin
     CASBIN_RBAC_MODEL_NAME: str = 'rbac_model.conf'
     CASBIN_EXCLUDE: list[dict[str, str], dict[str, str]] = [
-        {'method': 'POST', 'path': '/api/v1/auth/swagger_login'},
-        {'method': 'POST', 'path': '/api/v1/auth/login'},
-        {'method': 'POST', 'path': '/api/v1/auth/register'},
-        {'method': 'POST', 'path': '/api/v1/auth/password/reset'},
+        {'method': 'POST', 'path': '/v1/auth/swagger_login'},
+        {'method': 'POST', 'path': '/v1/auth/login'},
+        {'method': 'POST', 'path': '/v1/auth/register'},
+        {'method': 'POST', 'path': '/v1/auth/password/reset'},
+    ]
+
+    # Opera log
+    OPERA_LOG_EXCLUDE: list[str] = [
+        DOCS_URL,
+        REDOCS_URL,
+        OPENAPI_URL,
+        '/v1/auth/swagger_login',
     ]
 
     class Config:

backend/app/core/registrar.py

@@ -13,6 +13,7 @@
 from backend.app.common.task import scheduler
 from backend.app.core.conf import settings
 from backend.app.database.db_mysql import create_table
+from backend.app.middleware.opera_log_middleware import OperaLogMiddleware
 from backend.app.middleware.jwt_auth_middleware import JwtAuthMiddleware
 from backend.app.utils.health_check import ensure_unique_route_names
 from backend.app.utils.openapi import simplify_operation_ids
@@ -91,16 +92,25 @@ def register_static_file(app: FastAPI):
 
 
 def register_middleware(app: FastAPI):
+    """
+    中间件，执行顺序从下往上
+
+    :param app:
+    :return:
+    """
     # Gzip
     if settings.MIDDLEWARE_GZIP:
         from fastapi.middleware.gzip import GZipMiddleware
 
         app.add_middleware(GZipMiddleware)
-    # Api access logs
+    # Access log
+    # TODO: opera log 中间件完全可行时将被删除
     if settings.MIDDLEWARE_ACCESS:
         from backend.app.middleware.access_middleware import AccessMiddleware
 
         app.add_middleware(AccessMiddleware)
+    # Opera log
+    app.add_middleware(OperaLogMiddleware)
     # JWT auth: Always open
     app.add_middleware(
         AuthenticationMiddleware, backend=JwtAuthMiddleware(), on_error=JwtAuthMiddleware.auth_exception_handler

backend/app/crud/crud_login_log.py

@@ -11,7 +11,9 @@
 
 
 class CRUDLoginLog(CRUDBase[LoginLog, CreateLoginLog, UpdateLoginLog]):
-    async def get_all(self, username: str = None, status: bool = None, ipaddr: str = None) -> Select:
+    async def get_all(
+        self, username: str | None = None, status: bool | None = None, ipaddr: str | None = None
+    ) -> Select:
         se = select(self.model).order_by(desc(self.model.create_time))
         where_list = []
         if username:

backend/app/crud/crud_opera_log.py

@@ -0,0 +1,39 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from typing import NoReturn
+
+from sqlalchemy import select, desc, and_, delete, Select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.crud.base import CRUDBase
+from backend.app.models import OperaLog
+from backend.app.schemas.opera_log import CreateOperaLog, UpdateOperaLog
+
+
+class CRUDOperaLogDao(CRUDBase[OperaLog, CreateOperaLog, UpdateOperaLog]):
+    async def get_all(self, username: str | None = None, status: bool | None = None, ipaddr: str | None = None) -> Select:
+        se = select(self.model).order_by(desc(self.model.create_time))
+        where_list = []
+        if username:
+            where_list.append(self.model.username.like(f'%{username}%'))
+        if status is not None:
+            where_list.append(self.model.status == status)
+        if ipaddr:
+            where_list.append(self.model.ipaddr.like(f'%{ipaddr}%'))
+        if where_list:
+            se = se.where(and_(*where_list))
+        return se
+
+    async def create(self, db: AsyncSession, obj_in: CreateOperaLog) -> NoReturn:
+        await self.create_(db, obj_in)
+
+    async def delete(self, db: AsyncSession, pk: list[int]) -> int:
+        logs = await db.execute(delete(self.model).where(self.model.id.in_(pk)))
+        return logs.rowcount
+
+    async def delete_all(self, db: AsyncSession) -> int:
+        logs = await db.execute(delete(self.model))
+        return logs.rowcount
+
+
+OperaLogDao: CRUDOperaLogDao = CRUDOperaLogDao(OperaLog)

backend/app/middleware/access_middleware.py

@@ -3,15 +3,15 @@
 from datetime import datetime
 
 from fastapi import Request, Response
-from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
 
 from backend.app.common.log import log
 
 
 class AccessMiddleware(BaseHTTPMiddleware):
     """记录请求日志中间件"""
 
-    async def dispatch(self, request: Request, call_next) -> Response:
+    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
         start_time = datetime.now()
         response = await call_next(request)
         end_time = datetime.now()

backend/app/middleware/jwt_auth_middleware.py

@@ -52,4 +52,6 @@ async def authenticate(self, request: Request):
 
             raise _AuthenticationError(msg=traceback.format_exc() if settings.ENVIRONMENT == 'dev' else None)
 
+        # 请注意，此返回使用非标准模式，所以在认证通过时，将丢失某些标准特性
+        # 标准返回模式请查看：https://www.starlette.io/authentication/
         return auth, user

backend/app/middleware/opera_log_middleware.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from datetime import datetime
+from typing import Any
+
+from starlette.background import BackgroundTask
+from starlette.requests import Request
+from starlette.types import ASGIApp, Scope, Receive, Send
+from user_agents import parse
+
+from backend.app.common.log import log
+from backend.app.core.conf import settings
+from backend.app.schemas.opera_log import CreateOperaLog
+from backend.app.services.opera_log_service import OperaLogService
+from backend.app.utils import request_parse
+
+
+class OperaLogMiddleware:
+    """操作日志中间件"""
+
+    def __init__(self, app: ASGIApp):
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope['type'] != 'http':
+            await self.app(scope, receive, send)
+            return
+
+        request = Request(scope=scope, receive=receive)
+
+        # 排除记录白名单
+        path = request.url.path
+        if path in settings.OPERA_LOG_EXCLUDE:
+            await self.app(scope, receive, send)
+            return
+
+        # 请求信息解析
+        ip = await request_parse.get_request_ip(request)
+        user_agent = request.headers.get('User-Agent')
+        user_agent_parsed = parse(user_agent)
+        os = user_agent_parsed.get_os()
+        browser = user_agent_parsed.get_browser()
+        if settings.LOCATION_PARSE == 'online':
+            location = await request_parse.get_location_online(ip, user_agent)
+        elif settings.LOCATION_PARSE == 'offline':
+            location = request_parse.get_location_offline(ip)
+        else:
+            location = '未知'
+        try:
+            # 此信息依赖于 jwt 中间件
+            username = request.user.username
+        except AttributeError:
+            username = None
+        method = request.method
+        args = dict(request.query_params)
+        # TODO: 注释说明，详见 https://github.com/fastapi-practices/fastapi_best_architecture/pull/92
+        # form_data = await request.form()
+        # if len(form_data) > 0:
+        #     args = json.dumps(
+        #         args.update({k: v.filename if isinstance(v, UploadFile) else v for k, v in form_data.items()}),
+        #         ensure_ascii=False,
+        #     )
+        # else:
+        #     body = await request.body()
+        #     if body:
+        #         json_data = await request.json()
+        #         args = json.dumps(args.update(json_data), ensure_ascii=False)
+        args = str(args) if len(args) > 0 else None
+
+        # 设置附加请求信息
+        request.state.ip = ip
+        request.state.location = location
+        request.state.os = os
+        request.state.browser = browser
+
+        # 预置响应信息
+        code: int = 200
+        msg: str = 'Success'
+        status: bool = True
+        err: Any = None
+
+        # 执行请求
+        start_time = datetime.now()
+        try:
+            await self.app(request.scope, request.receive, send)
+        except Exception as e:
+            log.exception(e)
+            code = getattr(e, 'code', 500)
+            msg = getattr(e, 'msg', 'Internal Server Error')
+            status = False
+            err = e
+        end_time = datetime.now()
+        summary = request.scope.get('route').summary
+        title = summary if summary != '' else request.scope.get('route').summary
+        cost_time = (end_time - start_time).total_seconds() / 1000.0
+
+        # 日志创建
+        opera_log_in = CreateOperaLog(
+            username=username,
+            method=method,
+            title=title,
+            path=path,
+            ipaddr=ip,
+            location=location,
+            args=args,
+            status=status,
+            code=code,
+            msg=msg,
+            cost_time=cost_time,
+            opera_time=start_time,
+        )
+        back = BackgroundTask(OperaLogService.create, opera_log_in)
+        await back()
+
+        # 错误抛出
+        if err:
+            raise err from None

backend/app/models/__init__.py

@@ -12,3 +12,4 @@
 from backend.app.models.sys_role import Role
 from backend.app.models.sys_user import User
 from backend.app.models.sys_login_log import LoginLog
+from backend.app.models.sys_opera_log import OperaLog

backend/app/models/sys_login_log.py

@@ -14,13 +14,13 @@ class LoginLog(DataClassBase):
     __tablename__ = 'sys_login_log'
 
     id: Mapped[id_key] = mapped_column(init=False)
-    user_uuid: Mapped[str] = mapped_column(String(50), nullable=False, comment='用户UUID')
-    username: Mapped[str] = mapped_column(String(20), nullable=False, comment='用户名')
+    user_uuid: Mapped[str] = mapped_column(String(50), comment='用户UUID')
+    username: Mapped[str] = mapped_column(String(20), comment='用户名')
     status: Mapped[bool] = mapped_column(insert_default=0, comment='登录状态(0失败 1成功)')
-    ipaddr: Mapped[str] = mapped_column(String(50), nullable=False, comment='登录IP地址')
-    location: Mapped[str] = mapped_column(String(255), nullable=False, comment='归属地')
-    browser: Mapped[str] = mapped_column(String(255), nullable=False, comment='浏览器')
-    os: Mapped[str] = mapped_column(String(255), nullable=False, comment='操作系统')
-    msg: Mapped[str] = mapped_column(String(255), nullable=False, comment='提示消息')
-    login_time: Mapped[datetime] = mapped_column(nullable=False, comment='登录时间')
+    ipaddr: Mapped[str] = mapped_column(String(50), comment='登录IP地址')
+    location: Mapped[str] = mapped_column(String(50), comment='归属地')
+    browser: Mapped[str] = mapped_column(String(50), comment='浏览器')
+    os: Mapped[str] = mapped_column(String(50), comment='操作系统')
+    msg: Mapped[str] = mapped_column(String(255), comment='提示消息')
+    login_time: Mapped[datetime] = mapped_column(comment='登录时间')
     create_time: Mapped[datetime] = mapped_column(init=False, default=func.now(), comment='创建时间')

backend/app/models/sys_opera_log.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from datetime import datetime
+
+from sqlalchemy import String, func
+from sqlalchemy.dialects.mysql import JSON
+from sqlalchemy.orm import Mapped, mapped_column
+
+from backend.app.database.base_class import DataClassBase, id_key
+
+
+class OperaLog(DataClassBase):
+    """操作日志表"""
+
+    __tablename__ = 'sys_opera_log'
+
+    id: Mapped[id_key] = mapped_column(init=False)
+    username: Mapped[str | None] = mapped_column(String(20), comment='用户名')
+    method: Mapped[str] = mapped_column(String(20), comment='请求类型')
+    title: Mapped[str] = mapped_column(String(255), comment='操作模块')
+    path: Mapped[str] = mapped_column(String(500), comment='请求路径')
+    ipaddr: Mapped[str] = mapped_column(String(50), comment='IP地址')
+    location: Mapped[str] = mapped_column(String(50), comment='归属地')
+    args: Mapped[str | None] = mapped_column(JSON(), comment='请求参数')
+    status: Mapped[bool] = mapped_column(comment='操作状态（0异常 1正常）')
+    code: Mapped[int] = mapped_column(insert_default=200, comment='操作状态码')
+    msg: Mapped[str | None] = mapped_column(String(255), comment='提示消息')
+    cost_time: Mapped[float] = mapped_column(insert_default=0.0, comment='请求耗时ms')
+    opera_time: Mapped[datetime] = mapped_column(comment='操作时间')
+    create_time: Mapped[datetime] = mapped_column(init=False, default=func.now(), comment='创建时间')