RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
134 lines
4.4 KiB
Python
134 lines
4.4 KiB
Python
# for_admin_catch_log.py
|
|
# 管理员日志查看：读取 bot-utility 的 systemd 日志或指定日志文件。
|
|
import asyncio
|
|
import os
|
|
import subprocess
|
|
|
|
LOG_PATH = os.environ.get("BOT_LOG_PATH", "")
|
|
LOG_UNIT = os.environ.get("BOT_LOG_UNIT", "bot-utility.service")
|
|
DEFAULT_LINES = 5
|
|
MAX_LINES = 100
|
|
|
|
LOG_ENCODING = os.environ.get("BOT_LOG_ENCODING", "utf-8")
|
|
|
|
|
|
def _admin_list():
|
|
from channel import admin_list
|
|
return [x for n in ("qq", "tg") for x in admin_list(n)]
|
|
|
|
|
|
def _tail_log(lines):
|
|
"""读取最后 lines 行；未指定 BOT_LOG_PATH 时使用 journal，返回 (ok, text)。"""
|
|
if not LOG_PATH:
|
|
return _tail_journal(lines)
|
|
if not os.path.exists(LOG_PATH):
|
|
return False, f"日志文件不存在: {LOG_PATH}"
|
|
try:
|
|
# 从文件尾部倒着读块, 避免一次性载入大文件
|
|
with open(LOG_PATH, "rb") as f:
|
|
f.seek(0, os.SEEK_END)
|
|
size = f.tell()
|
|
block = 8192
|
|
pos = size
|
|
buf = b""
|
|
while pos > 0 and buf.count(b"\n") < lines + 1:
|
|
read_len = min(block, pos)
|
|
pos -= read_len
|
|
f.seek(pos)
|
|
buf = f.read(read_len) + buf
|
|
tail_bytes = buf
|
|
text = tail_bytes.decode(LOG_ENCODING, errors="replace")
|
|
rows = text.splitlines()
|
|
if len(rows) > lines:
|
|
rows = rows[-lines:]
|
|
body = "\n".join(rows).strip()
|
|
if not body:
|
|
return False, "日志为空"
|
|
return True, body
|
|
except Exception as e:
|
|
return False, f"读取日志异常: {str(e)[:200]}"
|
|
|
|
|
|
def _tail_journal(lines):
|
|
try:
|
|
result = subprocess.run(
|
|
["journalctl", "--unit", LOG_UNIT, "--lines", str(lines),
|
|
"--no-pager", "--output=cat", "--quiet"],
|
|
capture_output=True, text=True, encoding="utf-8", errors="replace",
|
|
timeout=10, check=False,
|
|
)
|
|
if result.returncode or "permission" in result.stderr.lower():
|
|
return False, f"读取 systemd 日志失败: {result.stderr.strip()[:200]}"
|
|
body = result.stdout.strip()
|
|
return (True, body) if body else (False, "日志为空")
|
|
except (OSError, subprocess.TimeoutExpired) as e:
|
|
return False, f"读取 systemd 日志异常: {str(e)[:200]}"
|
|
|
|
|
|
def _dedupe_merge(text):
|
|
"""合并连续重复行: 相同行连续出现 N 次时只保留一行并标注 (xN)"""
|
|
rows = text.splitlines()
|
|
if not rows:
|
|
return ""
|
|
merged = []
|
|
prev = rows[0]
|
|
count = 1
|
|
for row in rows[1:]:
|
|
if row == prev:
|
|
count += 1
|
|
else:
|
|
merged.append(f"{prev} (x{count})" if count > 1 else prev)
|
|
prev = row
|
|
count = 1
|
|
merged.append(f"{prev} (x{count})" if count > 1 else prev)
|
|
return "\n".join(merged)
|
|
|
|
|
|
def _clip(text):
|
|
"""限制输出长度, 避免刷屏 (QQ 单条消息上限约 4000 字符)"""
|
|
if len(text) > 3800:
|
|
text = text[-3800:]
|
|
return text
|
|
|
|
|
|
HELP_TEXT = (
|
|
"日志查看指令\n"
|
|
"━━━━━━━━━━━━\n"
|
|
"• .log help 显示本帮助\n"
|
|
"• .log <int> 获取最近的 N 条日志 (最大 100, 默认 5)\n"
|
|
)
|
|
|
|
|
|
async def handle_command(ws, msg_type, target_id, raw_text):
|
|
from channel import send_message
|
|
parts = raw_text.split(maxsplit=1)
|
|
arg = parts[1].strip() if len(parts) >= 2 else ""
|
|
|
|
if arg.lower() == "help":
|
|
await send_message(ws, msg_type, target_id, HELP_TEXT)
|
|
return
|
|
|
|
if arg == "":
|
|
lines = DEFAULT_LINES
|
|
else:
|
|
try:
|
|
lines = int(arg)
|
|
except ValueError:
|
|
await send_message(ws, msg_type, target_id,
|
|
"参数必须为整数, 例如 .log 10 或 .log help")
|
|
return
|
|
if lines < 1:
|
|
await send_message(ws, msg_type, target_id, "日志条数至少为 1")
|
|
return
|
|
if lines > MAX_LINES:
|
|
lines = MAX_LINES
|
|
|
|
ok, text = await asyncio.to_thread(_tail_log, lines)
|
|
if not ok:
|
|
await send_message(ws, msg_type, target_id, text)
|
|
return
|
|
|
|
body = _clip(_dedupe_merge(text))
|
|
await send_message(ws, msg_type, target_id,
|
|
f"最近 {lines} 条日志:\n━━━━━━━━━━━━\n{body}")
|