'wip(logging):调整日志记录方式'

applications/common/admin_log.py

@@ -1,56 +0,0 @@
-from flask_login import current_user
-
-from applications.extensions import db
-from applications.models import AdminLog
-
-
-def login_log(request, uid, is_access):
-    info = {
-        'method': request.method,
-        'url': request.path,
-        'ip': request.remote_addr,
-        'user_agent': request.headers.get('User-Agent'),
-        'desc': request.form.get('username'),
-        'uid': uid,
-        'success': int(is_access)
-
-    }
-    log = AdminLog(
-        url=info.get('url'),
-        ip=info.get('ip'),
-        user_agent=info.get('user_agent'),
-        desc=info.get('desc'),
-        uid=info.get('uid'),
-        method=info.get('method'),
-        success=info.get('success')
-    )
-    db.session.add(log)
-    db.session.flush()
-    db.session.commit()
-    return log.id
-
-
-def admin_log(request, is_access):
-    info = {
-        'method': request.method,
-        'url': request.path,
-        'ip': request.remote_addr,
-        'user_agent': request.headers.get('User-Agent'),
-        'desc': str(dict(request.values)),
-        'uid': current_user.id,
-        'success': int(is_access)
-
-    }
-    log = AdminLog(
-        url=info.get('url'),
-        ip=info.get('ip'),
-        user_agent=info.get('user_agent'),
-        desc=info.get('desc'),
-        uid=info.get('uid'),
-        method=info.get('method'),
-        success=info.get('success')
-    )
-    db.session.add(log)
-    db.session.commit()
-
-    return log.id

applications/common/utils/rights.py

@@ -1,7 +1,64 @@
+import typing as t
 from functools import wraps
+
 from flask import abort, request, jsonify, session
 from flask_login import login_required
-from applications.common.admin_log import admin_log
+from flask_login import current_user
+
+from applications.extensions import db
+from applications.models import LoggingModel
+
+
+def record_logging(success: bool = True) -> None:
+    """
+    记录用户日志数据
+    """
+    info = {
+        'method': request.method,
+        'url': request.path,
+        'ip': request.remote_addr,
+        'user_agent': request.headers.get('User-Agent'),
+        'desc': str(dict(request.values)),
+        'uid': current_user.id,
+        'success': success
+    }
+    log = LoggingModel()
+    for key, value in info.items():
+        setattr(log, key, value)
+
+    db.session.add(log)
+    db.session.commit()
+
+
+def view_logging_required(func: t.Callable) -> t.Callable:
+    """
+    日志装饰器，用于记录请求
+    """
+
+    @wraps(func)
+    def wrapper(*args, **kwargs) -> t.Callable:
+        record_logging()
+        return func(*args, **kwargs)
+
+    return wrapper
+
+
+def permission_required(permission: str) -> t.Callable:
+    """
+    权限装饰器，用于过滤需要的权限
+    """
+
+    def decorator(func: t.Callable):
+        @wraps(func)
+        def wrapper(*args, **kwargs) -> t.Callable:
+            if permission not in session.get('permissions'):
+                record_logging(success=False)
+                abort(403)
+            return func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator
 
 
 def authorize(power: str, log: bool = False):
@@ -11,17 +68,15 @@ def authorize(power: str, log: bool = False):
         def wrapper(*args, **kwargs):
             if power not in session.get('permissions'):
                 if log:
-                    admin_log(request=request, is_access=False)
+                    record_logging()
                 if request.method == 'GET':
                     abort(403)
                 else:
                     return jsonify(success=False, msg="权限不足!")
             if log:
-                admin_log(request=request, is_access=True)
+                record_logging()
             return func(*args, **kwargs)
 
         return wrapper
 
     return decorator
-
-