221 lines
7.3 KiB
Python
221 lines
7.3 KiB
Python
|
# =============================================================================
|
||
|
|
# 企微IT智能服务台 — 运行期日志服务
|
||
|
|
# =============================================================================
|
||
|
|
# 说明：读取后端运行期日志目录（config.RUNTIME_LOG_DIR）下的 *.log 文件，
|
||
|
|
# 逐行 json.loads（容错跳过非 JSON 行），按 级别 >= 阈值 / 时间区间 /
|
||
|
|
# 关键字 过滤，按时间倒序分页返回结构化条目；
|
||
|
|
# download 模式返回原始 JSON 行（供流式下载）。
|
||
|
|
# 目录不存在 / 不可读时返回空结果（不抛 500）。
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import os
|
||
|
|
import re
|
||
|
|
from datetime import datetime
|
||
|
|
from typing import Any, AsyncGenerator, Dict, List, Optional
|
||
|
|
|
||
|
|
from app.config import settings
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
# 日志级别数值映射（用于 >= 阈值 比较）
|
||
|
|
_LEVEL_ORDER = {
|
||
|
|
"DEBUG": 10,
|
||
|
|
"INFO": 20,
|
||
|
|
"WARNING": 30,
|
||
|
|
"ERROR": 40,
|
||
|
|
"CRITICAL": 50,
|
||
|
|
}
|
||
|
|
|
||
|
|
# 轮转备份命名形如 wecom-it-desk.log.1 / .2 ...
|
||
|
|
_ROTATED_RE = re.compile(r"\.log\.\d+$")
|
||
|
|
|
||
|
|
|
||
|
|
def _level_num(level: str) -> int:
|
||
|
|
"""将级别名转为数值（用于 >= 阈值比较）。"""
|
||
|
|
return _LEVEL_ORDER.get((level or "INFO").upper(), 20)
|
||
|
|
|
||
|
|
|
||
|
|
def _parse_ts(ts: Optional[str]) -> datetime:
|
||
|
|
"""解析 JSONFormatter 写入的时间戳（ISO8601，可能带 Z）。
|
||
|
|
|
||
|
|
解析失败时返回 datetime.min，使其排在最后（不影响过滤语义）。
|
||
|
|
"""
|
||
|
|
if not ts:
|
||
|
|
return datetime.min
|
||
|
|
try:
|
||
|
|
return datetime.fromisoformat(ts.replace("Z", "+00:00"))
|
||
|
|
except (ValueError, TypeError):
|
||
|
|
try:
|
||
|
|
return datetime.strptime(ts, "%Y-%m-%dT%H:%M:%S.%f")
|
||
|
|
except (ValueError, TypeError):
|
||
|
|
return datetime.min
|
||
|
|
|
||
|
|
|
||
|
|
def _list_log_files() -> List[str]:
|
||
|
|
"""列出运行期日志目录下所有 .log 文件（含轮转备份）。
|
||
|
|
|
||
|
|
目录不存在 / 不可读时返回空列表（不抛异常）。
|
||
|
|
"""
|
||
|
|
log_dir = getattr(settings, "RUNTIME_LOG_DIR", None) or "/app/logs"
|
||
|
|
if not log_dir or not os.path.isdir(log_dir):
|
||
|
|
return []
|
||
|
|
files: List[str] = []
|
||
|
|
try:
|
||
|
|
for name in os.listdir(log_dir):
|
||
|
|
if name.endswith(".log") or _ROTATED_RE.search(name):
|
||
|
|
full = os.path.join(log_dir, name)
|
||
|
|
if os.path.isfile(full):
|
||
|
|
files.append(full)
|
||
|
|
except (OSError, PermissionError) as e:
|
||
|
|
logger.warning(f"读取运行期日志目录失败: {log_dir}: {e}")
|
||
|
|
return []
|
||
|
|
# 按文件名排序，保证读取顺序稳定（主日志优先于轮转备份）
|
||
|
|
files.sort()
|
||
|
|
return files
|
||
|
|
|
||
|
|
|
||
|
|
def _iter_raw_lines() -> List[str]:
|
||
|
|
"""逐文件读取所有日志行（原始文本，已去除行尾换行）。"""
|
||
|
|
lines: List[str] = []
|
||
|
|
for path in _list_log_files():
|
||
|
|
try:
|
||
|
|
with open(path, "r", encoding="utf-8", errors="ignore") as f:
|
||
|
|
for line in f:
|
||
|
|
stripped = line.rstrip("\n").rstrip("\r")
|
||
|
|
if stripped:
|
||
|
|
lines.append(stripped)
|
||
|
|
except (OSError, PermissionError) as e:
|
||
|
|
logger.warning(f"读取运行期日志文件失败: {path}: {e}")
|
||
|
|
continue
|
||
|
|
return lines
|
||
|
|
|
||
|
|
|
||
|
|
def _match(
|
||
|
|
entry: Dict[str, Any],
|
||
|
|
level_th: int,
|
||
|
|
from_time: Optional[datetime],
|
||
|
|
to_time: Optional[datetime],
|
||
|
|
keyword: Optional[str],
|
||
|
|
) -> bool:
|
||
|
|
"""判断单条日志是否命中筛选条件。"""
|
||
|
|
# 级别阈值：仅保留 >= 阈值的条目
|
||
|
|
if _level_num(entry.get("level", "INFO")) < level_th:
|
||
|
|
return False
|
||
|
|
# 时间区间（闭区间）
|
||
|
|
ts = _parse_ts(entry.get("timestamp"))
|
||
|
|
if from_time and ts < from_time:
|
||
|
|
return False
|
||
|
|
if to_time and ts > to_time:
|
||
|
|
return False
|
||
|
|
# 关键字：消息子串（大小写不敏感）；为空则跳过该过滤
|
||
|
|
if keyword:
|
||
|
|
msg = str(entry.get("message", ""))
|
||
|
|
if keyword.lower() not in msg.lower():
|
||
|
|
return False
|
||
|
|
return True
|
||
|
|
|
||
|
|
|
||
|
|
async def query_runtime_logs(
|
||
|
|
level: str = "INFO",
|
||
|
|
from_time: Optional[datetime] = None,
|
||
|
|
to_time: Optional[datetime] = None,
|
||
|
|
keyword: Optional[str] = None,
|
||
|
|
page: int = 1,
|
||
|
|
page_size: int = 50,
|
||
|
|
) -> Dict[str, Any]:
|
||
|
|
"""查询运行期日志（分页 + 多条件筛选）。
|
||
|
|
|
||
|
|
读取 config.RUNTIME_LOG_DIR 下 *.log（含轮转备份），逐行 json.loads
|
||
|
|
（容错跳过非 JSON 行），按 级别>=阈值 / 时间区间 / 关键字 过滤，
|
||
|
|
按时间倒序分页返回结构化条目。
|
||
|
|
|
||
|
|
目录不存在 / 不可读时返回空列表（不抛 500）。
|
||
|
|
|
||
|
|
Args:
|
||
|
|
level: 级别阈值（DEBUG/INFO/WARNING/ERROR/CRITICAL），返回 >= 该级别
|
||
|
|
from_time: 起始时间（可选）
|
||
|
|
to_time: 结束时间（可选）
|
||
|
|
keyword: 消息关键字子串（可选，大小写不敏感）
|
||
|
|
page: 页码，从 1 开始
|
||
|
|
page_size: 每页条数
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Dict: {items, total, page, page_size}
|
||
|
|
"""
|
||
|
|
level_th = _level_num(level)
|
||
|
|
entries: List[Dict[str, Any]] = []
|
||
|
|
for raw in _iter_raw_lines():
|
||
|
|
try:
|
||
|
|
entry = json.loads(raw)
|
||
|
|
except (json.JSONDecodeError, ValueError):
|
||
|
|
# 容错：跳过非 JSON 行
|
||
|
|
continue
|
||
|
|
if not isinstance(entry, dict):
|
||
|
|
continue
|
||
|
|
if _match(entry, level_th, from_time, to_time, keyword):
|
||
|
|
entries.append(entry)
|
||
|
|
|
||
|
|
# 按时间倒序（新 -> 旧）
|
||
|
|
entries.sort(key=lambda e: _parse_ts(e.get("timestamp")), reverse=True)
|
||
|
|
|
||
|
|
total = len(entries)
|
||
|
|
page = max(1, page)
|
||
|
|
page_size = max(1, page_size)
|
||
|
|
start = (page - 1) * page_size
|
||
|
|
page_items = entries[start:start + page_size]
|
||
|
|
|
||
|
|
items = []
|
||
|
|
for e in page_items:
|
||
|
|
items.append({
|
||
|
|
"timestamp": e.get("timestamp", ""),
|
||
|
|
"level": e.get("level", ""),
|
||
|
|
"logger": e.get("logger", ""),
|
||
|
|
"message": e.get("message", ""),
|
||
|
|
"module": e.get("module", ""),
|
||
|
|
"function": e.get("function", ""),
|
||
|
|
"line": e.get("line", ""),
|
||
|
|
"request_id": e.get("request_id"),
|
||
|
|
"user_id": e.get("user_id"),
|
||
|
|
})
|
||
|
|
|
||
|
|
return {
|
||
|
|
"items": items,
|
||
|
|
"total": total,
|
||
|
|
"page": page,
|
||
|
|
"page_size": page_size,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
async def iter_runtime_log_lines(
|
||
|
|
level: str = "INFO",
|
||
|
|
from_time: Optional[datetime] = None,
|
||
|
|
to_time: Optional[datetime] = None,
|
||
|
|
keyword: Optional[str] = None,
|
||
|
|
) -> AsyncGenerator[str, None]:
|
||
|
|
"""流式返回命中的原始 JSON 行（供下载）。
|
||
|
|
|
||
|
|
逐行读取并过滤，命中即 yield 原始行文本（带换行），
|
||
|
|
便于 StreamingResponse 直接流式输出为 .log 文件。
|
||
|
|
|
||
|
|
Args:
|
||
|
|
level: 级别阈值
|
||
|
|
from_time: 起始时间（可选）
|
||
|
|
to_time: 结束时间（可选）
|
||
|
|
keyword: 消息关键字子串（可选）
|
||
|
|
|
||
|
|
Yields:
|
||
|
|
str: 命中的原始 JSON 日志行（含行尾换行）
|
||
|
|
"""
|
||
|
|
level_th = _level_num(level)
|
||
|
|
for raw in _iter_raw_lines():
|
||
|
|
try:
|
||
|
|
entry = json.loads(raw)
|
||
|
|
except (json.JSONDecodeError, ValueError):
|
||
|
|
continue
|
||
|
|
if not isinstance(entry, dict):
|
||
|
|
continue
|
||
|
|
if _match(entry, level_th, from_time, to_time, keyword):
|
||
|
|
yield raw + "\n"
|