Jeason
7950cd8237
1. 新增 resolve_tenant_by_chat_id() 根据飞书群 chat_id 查找绑定的租户 2. 新增 get_tenant_feishu_config() 获取租户级飞书凭证 3. FeishuService 支持传入自定义 app_id/app_secret（租户级别） 4. feishu_bot.py 收到消息时自动解析租户，使用租户凭证回复 5. feishu_longconn_service.py 同样按 chat_id 解析租户并传递 tenant_id 6. 租户管理 UI 新增飞书配置字段：App ID、App Secret、绑定群 Chat ID 7. 租户列表展示飞书绑定状态和群数量 8. 保存租户时同步更新飞书配置到 config JSON
130 lines
4.0 KiB
Python
130 lines
4.0 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
HTTP 错误日志分析脚本
|
||
|
||
扫描 logs/*/dashboard.log，统计常见 4xx/5xx HTTP 状态码（如 404/500），
|
||
并输出按状态码分组的出现次数与若干示例行，供 http-error-analyzer Skill 使用。
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Dict, List, Tuple
|
||
|
||
|
||
LOG_ROOT = Path("logs")
|
||
LOG_FILENAME = "dashboard.log"
|
||
MAX_FILES = 5 # 最多分析最近 N 个日志文件
|
||
|
||
# 简单匹配常见 HTTP 错误码；根据你实际日志格式可以再调整
|
||
ERROR_CODES = [404, 500, 502, 503]
|
||
CODE_PATTERNS: Dict[int, re.Pattern] = {
|
||
code: re.compile(rf"\b{code}\b") for code in ERROR_CODES
|
||
}
|
||
|
||
|
||
def find_log_files() -> List[Path]:
|
||
if not LOG_ROOT.exists():
|
||
return []
|
||
|
||
candidates: List[Tuple[float, Path]] = []
|
||
for root, dirs, files in os.walk(LOG_ROOT):
|
||
if LOG_FILENAME in files:
|
||
p = Path(root) / LOG_FILENAME
|
||
try:
|
||
mtime = p.stat().st_mtime
|
||
except OSError:
|
||
continue
|
||
candidates.append((mtime, p))
|
||
|
||
candidates.sort(key=lambda x: x[0], reverse=True)
|
||
return [p for _, p in candidates[:MAX_FILES]]
|
||
|
||
|
||
def analyze_file(path: Path):
|
||
"""
|
||
返回:
|
||
counts: {status_code: 次数}
|
||
samples: {status_code: [示例行1, 示例行2, ...]}
|
||
"""
|
||
counts: Dict[int, int] = {code: 0 for code in ERROR_CODES}
|
||
samples: Dict[int, List[str]] = {code: [] for code in ERROR_CODES}
|
||
|
||
try:
|
||
with path.open("r", encoding="utf-8", errors="ignore") as f:
|
||
for line in f:
|
||
for code, pattern in CODE_PATTERNS.items():
|
||
if pattern.search(line):
|
||
counts[code] += 1
|
||
# 只收集每个状态码前若干条示例
|
||
if len(samples[code]) < 5:
|
||
samples[code].append(line.strip()[:300])
|
||
break
|
||
except OSError as e:
|
||
print(f"[!] 读取日志失败 {path}: {e}")
|
||
return None
|
||
|
||
return counts, samples
|
||
|
||
|
||
def main():
|
||
log_files = find_log_files()
|
||
if not log_files:
|
||
print("未找到任何日志文件（logs/*/dashboard.log）。")
|
||
return
|
||
|
||
overall_counts: Dict[int, int] = {code: 0 for code in ERROR_CODES}
|
||
overall_samples: Dict[int, List[str]] = {code: [] for code in ERROR_CODES}
|
||
|
||
print(f"=== HTTP 错误日志分析（最近最多 {MAX_FILES} 个日志文件）===\n")
|
||
|
||
for idx, path in enumerate(log_files, start=1):
|
||
print(f"[{idx}] 日志文件: {path}")
|
||
result = analyze_file(path)
|
||
if result is None:
|
||
print(" 无法读取该日志文件。\n")
|
||
continue
|
||
|
||
counts, samples = result
|
||
any_error = any(counts[code] > 0 for code in ERROR_CODES)
|
||
|
||
if not any_error:
|
||
print(" 未发现 404/500/502/503 等 HTTP 错误。")
|
||
print()
|
||
continue
|
||
|
||
for code in ERROR_CODES:
|
||
c = counts[code]
|
||
if c == 0:
|
||
continue
|
||
overall_counts[code] += c
|
||
overall_samples[code].extend(samples[code])
|
||
print(f" 状态码 {code}: {c} 次")
|
||
if samples[code]:
|
||
print(" 示例：")
|
||
for line in samples[code]:
|
||
print(f" {line}")
|
||
|
||
print()
|
||
|
||
print("=== 汇总统计 ===")
|
||
any_overall = any(overall_counts[code] > 0 for code in ERROR_CODES)
|
||
if not any_overall:
|
||
print(" 未在最近的日志文件中发现任何配置的 HTTP 错误状态码。")
|
||
return
|
||
|
||
for code in ERROR_CODES:
|
||
c = overall_counts[code]
|
||
if c == 0:
|
||
continue
|
||
print(f" 状态码 {code}: {c} 次")
|
||
|
||
print("\n提示：以上为按状态码汇总的次数与部分示例行，")
|
||
print("你可以结合 URL 路径、接口名称、堆栈片段来推断可能的路由配置问题、权限问题或后端异常。")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|