feat: 添加系统日志功能,记录运行时错误和AI调用失败,提供日志查看页面

This commit is contained in:
2026-03-14 00:32:20 +08:00
parent d2d63d5e61
commit dc7efb8ff8
10 changed files with 208 additions and 2 deletions

139
app.py
View File

@@ -11,11 +11,14 @@ import re
import csv
import json
import hmac
import logging
import difflib
import base64
import random
import socket
import string
import hashlib
import traceback
import time
import urllib.error
import urllib.parse
@@ -23,9 +26,11 @@ import urllib.request
from copy import deepcopy
from io import StringIO
from datetime import datetime, timedelta
from logging.handlers import RotatingFileHandler
from flask import Flask, Response, redirect, render_template, request, session, url_for
from flask_sqlalchemy import SQLAlchemy
from werkzeug.exceptions import HTTPException
from werkzeug.security import check_password_hash, generate_password_hash
@@ -33,6 +38,7 @@ BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DB_DIR = os.path.join(BASE_DIR, "data")
os.makedirs(DB_DIR, exist_ok=True)
DB_PATH = os.path.join(DB_DIR, "inventory.db")
APP_LOG_PATH = os.path.join(DB_DIR, "app.log")
# Flask 和 SQLAlchemy 基础初始化。
app = Flask(__name__)
@@ -44,6 +50,58 @@ app.config["SECRET_KEY"] = os.environ.get(
)
db = SQLAlchemy(app)
def _setup_app_logger() -> None:
"""初始化应用日志文件。
中文说明把运行期错误、AI调用失败等信息写入 data/app.log
方便在系统内直接查看,不再依赖控制台输出。
"""
if getattr(app, "_inventory_log_ready", False):
return
handler = RotatingFileHandler(
APP_LOG_PATH,
maxBytes=512 * 1024,
backupCount=3,
encoding="utf-8",
)
handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
for existing_handler in app.logger.handlers:
if getattr(existing_handler, "baseFilename", "") == handler.baseFilename:
app._inventory_log_ready = True
return
app.logger.addHandler(handler)
app.logger.setLevel(logging.INFO)
app._inventory_log_ready = True
def _log_event(level: int, event: str, **context) -> None:
parts = []
for key, value in context.items():
if value is None:
continue
text = str(value).replace("\n", " ").strip()
if not text:
continue
if len(text) > 280:
text = text[:277] + "..."
parts.append(f"{key}={text}")
message = event if not parts else f"{event} | {' | '.join(parts)}"
app.logger.log(level, message)
def _read_log_lines(limit: int = 200) -> list[str]:
if not os.path.exists(APP_LOG_PATH):
return []
with open(APP_LOG_PATH, "r", encoding="utf-8", errors="ignore") as file_obj:
return [line.rstrip("\n") for line in file_obj.readlines()[-limit:]]
_setup_app_logger()
# 这里集中放全局常量,避免后面函数里散落硬编码。
LOW_STOCK_THRESHOLD = 5
BOX_TYPES_OVERRIDE_PATH = os.path.join(DB_DIR, "box_types.json")
@@ -2807,6 +2865,8 @@ def _call_siliconflow_chat(
raise RuntimeError(f"AI 服务返回 HTTP {exc.code}: {detail[:200]}") from exc
except urllib.error.URLError as exc:
raise RuntimeError(f"AI 服务连接失败: {exc.reason}") from exc
except (TimeoutError, socket.timeout) as exc:
raise RuntimeError(f"AI 服务读取超时(>{timeout}秒),请稍后重试或在 AI 参数中调大超时") from exc
try:
data = json.loads(raw)
@@ -2856,6 +2916,28 @@ def logout_page():
return redirect(url_for("login_page"))
@app.route("/system/logs")
def system_logs_page():
"""查看系统日志。
中文说明:用于定位 AI 调用失败、未处理异常、配置错误等问题,
默认展示最近 200 行,可通过 query 参数 lines 调整。
"""
raw_lines = (request.args.get("lines", "200") or "200").strip()
try:
line_limit = max(50, min(int(raw_lines), 1000))
except ValueError:
line_limit = 200
log_lines = _read_log_lines(line_limit)
return render_template(
"logs.html",
log_lines=log_lines,
line_limit=line_limit,
log_path=APP_LOG_PATH,
)
@app.route("/account/password", methods=["GET", "POST"])
def change_password_page():
"""登录后修改当前账号密码。
@@ -4244,6 +4326,14 @@ def ai_restock_plan():
timeout=ai_settings["timeout"],
)
except RuntimeError as exc:
_log_event(
logging.WARNING,
"ai_restock_plan_runtime_error",
error=str(exc),
model=ai_settings.get("model", ""),
api_url=ai_settings.get("api_url", ""),
low_stock_count=len(data.get("low_stock_items", [])),
)
fallback_plan = _build_rule_based_plan()
return {
"ok": False,
@@ -4251,6 +4341,22 @@ def ai_restock_plan():
"plan": fallback_plan,
"data": data,
}, 400
except Exception as exc:
_log_event(
logging.ERROR,
"ai_restock_plan_unexpected_error",
error=str(exc),
traceback=traceback.format_exc(),
model=ai_settings.get("model", ""),
api_url=ai_settings.get("api_url", ""),
)
fallback_plan = _build_rule_based_plan()
return {
"ok": False,
"message": "服务器内部错误,请到系统日志查看详情",
"plan": fallback_plan,
"data": data,
}, 500
parse_warning = ""
try:
@@ -4799,6 +4905,39 @@ def clear_stats_logs():
return redirect(url_for("stats_page", days=days, box_type=box_type_filter, notice=notice))
@app.errorhandler(Exception)
def handle_app_exception(exc: Exception):
if isinstance(exc, HTTPException):
status_code = exc.code or 500
title = exc.name or "请求失败"
message = exc.description or "请求处理失败"
else:
status_code = 500
title = "服务器异常"
message = "服务器内部错误,请到系统日志查看详情。"
_log_event(
logging.ERROR if status_code >= 500 else logging.WARNING,
"unhandled_exception",
status_code=status_code,
path=request.path,
method=request.method,
error=str(exc),
traceback=traceback.format_exc() if status_code >= 500 else "",
)
return (
render_template(
"error.html",
status_code=status_code,
title=title,
message=message,
back_url=request.referrer or url_for("types_page"),
),
status_code,
)
def bootstrap() -> None:
"""应用启动时初始化数据库。