iTi-Flask/iti/applications/extensions/sys_log.py

from iti.applications.models import SysLog
from .db import db
from flask import request, current_app, g
import time
import json
from flask_jwt_extended import current_user
from iti.applications.common.utils.sys_log_helper import (
    summarize_response_json,
    safe_json_dumps,
    truncate_text,
)
from iti.applications.common.enums import LogType


def _summarize_response_json(obj: dict, items_sample_size: int) -> dict:
    """
    响应结果摘要化
    """
    try:
        if isinstance(obj, dict) and isinstance(obj.get("data"), dict):
            data = obj["data"]
            if isinstance(data.get("items"), list):
                items = data["items"]
                obj = {
                    **obj,
                    "data": {
                        **data,
                        "items": {
                            "count": len(items),
                            "sample": items[:items_sample_size],
                        },
                    },
                }
        return obj
    except Exception:
        return obj


def init_sys_log(app):
    max_body_chars = int(app.config.get("SYSLOG_MAX_BODY_CHARS", 2048))
    items_sample_size = int(app.config.get("SYSLOG_ITEMS_SAMPLE", 5))

    @app.before_request
    def _syslog_before_request():
        try:
            view_func = app.view_functions.get(request.endpoint)
            meta = getattr(view_func, "_sys_log_meta", None) if view_func else None
            g._syslog_meta = meta
            if not meta:
                return
            g._syslog_start = time.time()
            headers = {k: request.headers.getlist(k) for k in request.headers.keys()}
            g._syslog_req = {
                "method": request.method,
                "path": request.path,
                "headers": headers,
                "query": request.args.to_dict(),
                "body": request.get_json(silent=True) or {},
                "user_id": current_user.id if current_user else None,
                "user_agent": request.headers.get("User-Agent", "None"),
                "ip": request.remote_addr or request.headers.get("X-Forwarded-For"),
            }
        except Exception as e:
            current_app.logger.debug(f"syslog before_request error: {e}")

    @app.after_request
    def _syslog_after_request(response):
        try:
            meta = getattr(g, "_syslog_meta", None)
            if not meta or getattr(g, "_syslog_logged", False):
                return response

            execution_time = None
            if getattr(g, "_syslog_start", None) is not None and meta.get(
                "execute_time", True
            ):
                execution_time = (time.time() - g._syslog_start) * 1000

            is_stream = getattr(response, "direct_passthrough", False)
            resp_json = None
            resp_text = None
            if not is_stream and hasattr(response, "is_json") and response.is_json:
                resp_json = response.get_json(silent=True)
            if resp_json is not None and isinstance(resp_json, dict):
                summarized = summarize_response_json(resp_json, items_sample_size)
                resp_text = safe_json_dumps(
                    summarized, ensure_ascii=False, max_chars=max_body_chars
                )
            else:
                if not is_stream:
                    data = response.get_data(as_text=True)
                    resp_text = truncate_text(data or "", max_body_chars)
                else:
                    resp_text = json.dumps(
                        {
                            "stream": True,
                            "content_type": response.headers.get("Content-Type"),
                            "content_length": response.calculate_content_length()
                            if hasattr(response, "calculate_content_length")
                            else None,
                        },
                        ensure_ascii=False,
                    )

            # success rule: prefer business code, then HTTP 2xx
            success = False
            if isinstance(resp_json, dict) and "code" in resp_json:
                success = bool(resp_json.get("code") == 200)
            else:
                success = 200 <= getattr(response, "status_code", 0) < 300

            req = getattr(g, "_syslog_req", {})
            log = SysLog(
                name=meta.get("name", ""),
                desc=meta.get("desc", ""),
                type=(
                    meta.get("type")
                    if isinstance(meta.get("type"), LogType)
                    else (meta.get("type") or LogType.OPERATION)
                ),
                method=req.get("method"),
                user_id=req.get("user_id"),
                path=req.get("path"),
                ip=req.get("ip"),
                user_agent=req.get("user_agent"),
                headers=json.dumps(req.get("headers"), ensure_ascii=False),
                query_params=json.dumps(req.get("query"), ensure_ascii=False),
                body_params=json.dumps(req.get("body"), ensure_ascii=False),
                execution_time=execution_time,
                response=resp_text,
                success=success,
            )

            if meta.get("save_db", True):
                db.session.add(log)
                db.session.commit()
            g._syslog_logged = True
        except Exception as e:
            current_app.logger.debug(f"syslog after_request error: {e}")
        return response

    @app.teardown_request
    def _syslog_teardown_request(exc):
        try:
            meta = getattr(g, "_syslog_meta", None)
            if not meta or getattr(g, "_syslog_logged", False):
                return
            # only log exceptions that didn't produce a normal response
            req = getattr(g, "_syslog_req", {})
            execution_time = None
            if getattr(g, "_syslog_start", None) is not None and meta.get(
                "execute_time", True
            ):
                execution_time = (time.time() - g._syslog_start) * 1000

            log = SysLog(
                name=meta.get("name", ""),
                desc=meta.get("desc", ""),
                type=(
                    meta.get("type")
                    if isinstance(meta.get("type"), LogType)
                    else (meta.get("type") or LogType.OPERATION)
                ),
                method=req.get("method"),
                user_id=req.get("user_id"),
                path=req.get("path"),
                ip=req.get("ip"),
                user_agent=req.get("user_agent"),
                headers=json.dumps(req.get("headers"), ensure_ascii=False),
                query_params=json.dumps(req.get("query"), ensure_ascii=False),
                body_params=json.dumps(req.get("body"), ensure_ascii=False),
                execution_time=execution_time,
                response=None,
                success=False,
                exception=str(exc) if exc else None,
            )
            if meta.get("save_db", True):
                db.session.add(log)
                db.session.commit()
            g._syslog_logged = True
        except Exception as e:
            current_app.logger.debug(f"syslog teardown_request error: {e}")


def sys_log(
    name: str = "",
    desc: str = "",
    save_db: bool = True,
    execute_time: bool = True,
    type: LogType = LogType.OPERATION,
):
    """
    轻量级标注：仅为视图函数打上 _sys_log_meta 属性，
    具体日志在 init_log 注册的钩子里处理。
    """

    def decorator(func):
        setattr(
            func,
            "_sys_log_meta",
            {
                "name": name,
                "desc": desc,
                "save_db": save_db,
                "execute_time": execute_time,
                "type": type,
            },
        )
        return func

    return decorator