butubb
2c32c397c8
「日志」页原来只能 tail 固定行数、且文件下拉写死了 4 项（新增的 notify.log
根本选不到）。本次：
- core/logger.py 新增读取接口：list_log_files()（模块文件 + .1/.2 滚动历史，
白名单）、query_log()（关键字/最低级别/时间过滤，返回结构化行）、
read_log_text()（导出）。
· 时间/级别过滤对**续行**用继承值（traceback 缩进行本身没有前缀），
按 ERROR 筛不会把堆栈拆散；
· 过滤按"最近 N 条命中"返回，保持文件原顺序；
- GET /api/logs 换新协议（rows/files/matched/scanned/truncated），
新增 GET /api/logs/download（附件下载，支持同样过滤；无一条件时整文件）；
- 顺带修 **目录穿越**：原实现 os.path.join(_LOG_DIR, file) 可用 ../../ 逃逸，
现在 file 必须命中白名单；
- 顺带修 web/admin_api.py 里 _log 未定义（用户增删改的日志行会 NameError）；
- 前端：文件下拉改为按接口渲染、加关键字(命中高亮)/级别/时间范围/下载/重置，
自动刷新时不再把正在上翻的用户拽回底部。
文档：doc/API.md §10.1（参数与返回）、README 日志节。
279 lines
11 KiB
Python
279 lines
11 KiB
Python
"""统一日志系统：按模块分文件，带时间戳和级别，自动滚动。
|
||
|
||
用法：
|
||
from core.logger import get_logger
|
||
log = get_logger("task") # 写 logs/task.log + 控制台
|
||
log.info("开始任务")
|
||
log.error("失败", exc_info=True)
|
||
|
||
模块划分：
|
||
core — 核心程序（STF/adb/worker/task_manager）
|
||
task — 任务执行（worker 业务逻辑）
|
||
web — web_server 请求/管理
|
||
action — 操作执行（点赞/评论等）
|
||
|
||
所有日志同时输出到控制台和对应文件，logs/ 目录自动创建。
|
||
文件按 10MB 滚动，保留 5 个历史文件。
|
||
"""
|
||
import os
|
||
import re
|
||
import logging
|
||
import datetime
|
||
from collections import deque
|
||
from logging.handlers import RotatingFileHandler
|
||
|
||
_LOG_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "logs")
|
||
os.makedirs(_LOG_DIR, exist_ok=True)
|
||
|
||
# 已创建的 logger，避免重复添加 handler
|
||
_LOGGERS = {}
|
||
|
||
# 模块 → 文件名映射
|
||
_MODULE_FILES = {
|
||
"core": "core.log",
|
||
"task": "task.log",
|
||
"web": "web.log",
|
||
"action": "action.log",
|
||
# 通知单独一个文件：排障时"哪条通知发了/失败了/为什么"要能一眼翻到
|
||
"notify": "notify.log",
|
||
}
|
||
|
||
_FORMAT = "%(asctime)s [%(levelname)s] [%(name)s] %(message)s"
|
||
_DATE_FMT = "%Y-%m-%d %H:%M:%S"
|
||
|
||
|
||
def get_logger(name="core"):
|
||
"""获取指定模块的 logger。
|
||
|
||
name: 模块名（core/task/web/action），决定写哪个文件。
|
||
也可传子模块名如 "task.generic"，会归到 task.log。
|
||
返回配置好的 logging.Logger。
|
||
"""
|
||
if name in _LOGGERS:
|
||
return _LOGGERS[name]
|
||
|
||
# 归类到对应文件：取顶层模块名
|
||
top = name.split(".")[0]
|
||
filename = _MODULE_FILES.get(top, "core.log")
|
||
|
||
log = logging.getLogger(name)
|
||
log.setLevel(logging.DEBUG)
|
||
# 避免向 root logger 传播导致重复输出
|
||
log.propagate = False
|
||
|
||
fmt = logging.Formatter(_FORMAT, _DATE_FMT)
|
||
|
||
# 文件 handler：10MB 滚动，保留 5 个
|
||
file_path = os.path.join(_LOG_DIR, filename)
|
||
fh = RotatingFileHandler(file_path, maxBytes=10 * 1024 * 1024,
|
||
backupCount=5, encoding="utf-8")
|
||
fh.setLevel(logging.DEBUG)
|
||
fh.setFormatter(fmt)
|
||
|
||
# 控制台 handler
|
||
ch = logging.StreamHandler()
|
||
ch.setLevel(logging.INFO)
|
||
ch.setFormatter(fmt)
|
||
|
||
log.addHandler(fh)
|
||
log.addHandler(ch)
|
||
|
||
_LOGGERS[name] = log
|
||
return log
|
||
|
||
|
||
# 兼容旧代码的 print 风格：紧急排查时可临时用
|
||
def log_print(msg, level="info", module="core"):
|
||
"""print 的替代品，转发到 logging。"""
|
||
log = get_logger(module)
|
||
getattr(log, level, log.info)(msg)
|
||
|
||
|
||
# ================== 日志读取（「日志」页的筛选 / 下载） ==================
|
||
#
|
||
# 读日志的代码放这里而不是 web 层：文件命名与滚动规则归本模块管
|
||
# （_MODULE_FILES + RotatingFileHandler 的 backupCount），**白名单校验必须和
|
||
# 写入侧用同一份事实**，否则改了文件名就会漏掉一处，变成目录穿越。
|
||
|
||
# 行格式与 _FORMAT 对应：2026-09-16 00:27:11 [INFO] [web.notify] 消息
|
||
_LINE_RE = re.compile(r"^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) "
|
||
r"\[([A-Z]+)\] \[([^\]]*)\] ?(.*)$")
|
||
_LEVEL_ORDER = {"DEBUG": 10, "INFO": 20, "WARNING": 30, "ERROR": 40, "CRITICAL": 50}
|
||
|
||
# 单次读取的行数上限：文件本身被限在 10MB（约 10 万行），这里再兜一层，
|
||
# 万一将来有人调大 maxBytes，也不会把内存吃爆。
|
||
_MAX_SCAN_LINES = 500000
|
||
|
||
_MODULE_LABELS = {
|
||
"core.log": "核心(adb/调度)",
|
||
"task.log": "任务执行",
|
||
"web.log": "Web/AI",
|
||
"action.log": "动作执行",
|
||
"notify.log": "通知发送",
|
||
}
|
||
|
||
|
||
def _rotation_files(base):
|
||
"""基础名 + 磁盘上实际存在的滚动副本（core.log → core.log.1/.2…）。
|
||
|
||
RotatingFileHandler 的备份编号**越大越旧**，所以 base 最新，接着 .1/.2…
|
||
"""
|
||
names = []
|
||
if os.path.exists(os.path.join(_LOG_DIR, base)):
|
||
names.append(base)
|
||
i = 1
|
||
while i <= 99 and os.path.exists(os.path.join(_LOG_DIR, f"{base}.{i}")):
|
||
names.append(f"{base}.{i}")
|
||
i += 1
|
||
return names
|
||
|
||
|
||
def list_log_files():
|
||
"""可查看的日志文件（模块文件 + 滚动历史），最近写入的排前面。
|
||
|
||
`file` 参数只能取这里的 name（白名单），杜绝 `../../` 之类的路径穿越。
|
||
"""
|
||
out = []
|
||
for base in sorted(set(_MODULE_FILES.values())):
|
||
label = _MODULE_LABELS.get(base, base)
|
||
for name in _rotation_files(base):
|
||
try:
|
||
st = os.stat(os.path.join(_LOG_DIR, name))
|
||
except OSError:
|
||
continue
|
||
rotated = name != base
|
||
out.append({
|
||
"name": name,
|
||
"label": f"{label} · {name}" + ("（历史）" if rotated else ""),
|
||
"module": base,
|
||
"size": st.st_size,
|
||
"mtime": int(st.st_mtime),
|
||
"rotated": rotated,
|
||
})
|
||
out.sort(key=lambda f: (-f["mtime"], f["name"]))
|
||
return out
|
||
|
||
|
||
def _norm_ts(value, end=False):
|
||
"""把前端的时间输入归一成 'YYYY-MM-DD HH:MM:SS'（可比字符串）。
|
||
|
||
接受 datetime-local 的 'YYYY-MM-DDTHH:MM'、'YYYY-MM-DD HH:MM' 和纯日期；
|
||
纯日期按整天算（end=True 取 23:59:59，否则 00:00:00）。
|
||
"""
|
||
s = (value or "").strip().replace("T", " ")
|
||
if not s:
|
||
return ""
|
||
if len(s) == 10: # 只给了日期 → 按整天
|
||
return s + (" 23:59:59" if end else " 00:00:00")
|
||
if len(s) == 16: # YYYY-MM-DD HH:MM
|
||
return s + (":59" if end else ":00")
|
||
return s[:19]
|
||
|
||
|
||
def _iter_filtered(fh, keyword, min_level, since, until, stats=None):
|
||
"""逐行读文件，产出命中的 {ts, level, module, msg, cont}。
|
||
|
||
过滤口径：
|
||
- 时间/级别用**继承值**——续行（traceback 的缩进行等）本身没有前缀，
|
||
继承上一条带前缀的行，这样按 ERROR 筛选时整段堆栈不会被拆散；
|
||
- 关键字匹配本行原文（续行也参与），大小写不敏感。
|
||
|
||
stats（可选 dict）里回填 scanned：本次实际读了多少行（用于告诉用户
|
||
"是读到文件开头了，还是被行数上限截断的"）。
|
||
"""
|
||
kw = (keyword or "").strip().lower()
|
||
floor = _LEVEL_ORDER.get((min_level or "").upper(), 0)
|
||
cur_ts = cur_level = cur_module = ""
|
||
for raw in fh:
|
||
if stats is not None:
|
||
stats["scanned"] += 1
|
||
line = raw.rstrip("\r\n")
|
||
m = _LINE_RE.match(line)
|
||
if m:
|
||
cur_ts, cur_level, cur_module, msg = m.groups()
|
||
cont = False
|
||
else:
|
||
# 续行（traceback、被切成多行的长消息）：沿用上一条的元信息
|
||
msg, cont = line, True
|
||
if since and (not cur_ts or cur_ts < since):
|
||
continue
|
||
if until and (not cur_ts or cur_ts > until):
|
||
continue
|
||
if floor and _LEVEL_ORDER.get(cur_level, 0) < floor:
|
||
continue
|
||
if kw and kw not in line.lower():
|
||
continue
|
||
yield {"ts": cur_ts, "level": cur_level, "module": cur_module,
|
||
"msg": msg, "cont": cont}
|
||
|
||
|
||
def query_log(name, keyword="", min_level="", since="", until="",
|
||
limit=300, max_scan=_MAX_SCAN_LINES):
|
||
"""取日志文件里**最近** limit 条满足条件的行。
|
||
|
||
返回 dict：
|
||
rows — [{ts, level, module, msg, cont}]，保持文件原顺序（旧→新）
|
||
matched — 命中总数（可能大于 len(rows)：更早的命中没返回）
|
||
scanned — 实际读了多少行（达到 max_scan 说明是被上限截断的）
|
||
truncated — matched > len(rows)，即"还有更早的命中没显示"
|
||
error — 出错时的原因（如文件不存在）
|
||
|
||
正序读整文件 + deque 保留最后 N 条：文件被 RollingFileHandler 限在 10MB
|
||
（约 10 万行），整读一次百毫秒级；换来的是续行能正确继承时间戳——倒着
|
||
读的写法在遇到 traceback 时必须把续行攒着等前面那行，容易出错。
|
||
"""
|
||
# 白名单：`name` 必须来自 list_log_files()，否则一律拒绝（防目录穿越）
|
||
if name not in {f["name"] for f in list_log_files()}:
|
||
return {"rows": [], "matched": 0, "scanned": 0, "truncated": False,
|
||
"error": "日志文件不存在或不可读取"}
|
||
since, until = _norm_ts(since), _norm_ts(until, end=True)
|
||
try:
|
||
keep = max(1, min(int(limit or 300), 5000))
|
||
except (TypeError, ValueError):
|
||
keep = 300
|
||
rows, matched, stats = deque(maxlen=keep), 0, {"scanned": 0}
|
||
try:
|
||
with open(os.path.join(_LOG_DIR, name), encoding="utf-8",
|
||
errors="replace") as fh:
|
||
for row in _iter_filtered(fh, keyword, min_level, since, until, stats):
|
||
matched += 1
|
||
rows.append(row)
|
||
if stats["scanned"] >= max_scan:
|
||
break
|
||
except OSError as e:
|
||
return {"rows": [], "matched": 0, "scanned": 0, "truncated": False,
|
||
"error": f"读取失败: {e}"}
|
||
return {"rows": list(rows), "matched": matched, "scanned": stats["scanned"],
|
||
"truncated": matched > len(rows), "error": ""}
|
||
|
||
|
||
def read_log_text(name, keyword="", min_level="", since="", until="",
|
||
limit=200000):
|
||
"""导出用的整段文本：命中行按原格式拼回去（下载按钮）。
|
||
|
||
没给任何条件时直接读整个文件（含滚动历史由调用方选定的那一个）。
|
||
"""
|
||
if name not in {f["name"] for f in list_log_files()}:
|
||
return None, "日志文件不存在或不可读取"
|
||
since, until = _norm_ts(since), _norm_ts(until, end=True)
|
||
any_filter = bool((keyword or "").strip() or min_level or since or until)
|
||
path = os.path.join(_LOG_DIR, name)
|
||
try:
|
||
with open(path, encoding="utf-8", errors="replace") as fh:
|
||
if not any_filter:
|
||
return fh.read(), ""
|
||
buf, n = [], 0
|
||
for row in _iter_filtered(fh, keyword, min_level, since, until):
|
||
if row["cont"]:
|
||
buf.append(row["msg"]) # 续行原样输出
|
||
else:
|
||
buf.append(f"{row['ts']} [{row['level']}] "
|
||
f"[{row['module']}] {row['msg']}")
|
||
n += 1
|
||
if n >= limit:
|
||
buf.append(f"...（已达导出上限 {limit} 行，请缩小时间范围）")
|
||
break
|
||
return "\n".join(buf) + ("\n" if buf else ""), ""
|
||
except OSError as e:
|
||
return None, f"读取失败: {e}"
|