移除HTTP详情请求日志
This commit is contained in:
yml2213
@@ -1,241 +0,0 @@
|
||||
"""HTTP 请求/响应详情日志记录器
|
||||
|
||||
将关键 HTTP 请求的完整详情（method、url、headers、body、status、response、耗时）
|
||||
以 JSON Lines 格式写入日志文件，便于在 Web 界面查看和排查问题。
|
||||
"""
|
||||
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
from loguru import logger
|
||||
|
||||
# 日志文件路径
|
||||
_LOG_DIR = Path("logs")
|
||||
_LOG_FILE = _LOG_DIR / "http_detail.jsonl"
|
||||
_MAX_BODY_LEN = 2000 # 单个 body 最大记录长度，避免过大
|
||||
|
||||
# 线程安全写锁
|
||||
_write_lock = threading.Lock()
|
||||
|
||||
# 确保日志目录存在
|
||||
_LOG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
def _truncate(text: Any, max_len: int = _MAX_BODY_LEN) -> str:
|
||||
"""截断过长的文本"""
|
||||
if text is None:
|
||||
return ""
|
||||
s = text if isinstance(text, str) else str(text)
|
||||
if len(s) > max_len:
|
||||
return s[:max_len] + f"...[truncated {len(s) - max_len} chars]"
|
||||
return s
|
||||
|
||||
|
||||
def _safe_headers(headers: Any) -> dict:
|
||||
"""清理 headers 中的敏感信息"""
|
||||
if not headers:
|
||||
return {}
|
||||
if hasattr(headers, 'items'):
|
||||
headers = dict(headers)
|
||||
safe = {}
|
||||
sensitive = {'authorization', 'cookie', 'set-cookie', 'password'}
|
||||
for k, v in headers.items():
|
||||
if k.lower() in sensitive:
|
||||
safe[k] = '***'
|
||||
else:
|
||||
safe[k] = v
|
||||
return safe
|
||||
|
||||
|
||||
# 请求体中需要脱敏的字段名（小写匹配，包含即遮罩）
|
||||
_SENSITIVE_BODY_KEYS = {
|
||||
'password', 'pwd', 'passwd', 'secret', 'token', 'apikey', 'api_key',
|
||||
'email_password', 'mm', 'authorization', 'credential',
|
||||
}
|
||||
|
||||
|
||||
def _safe_body(body: Any) -> str:
|
||||
"""清理请求体中的敏感字段值。
|
||||
|
||||
支持 dict、JSON 字符串、其他类型。
|
||||
敏感字段的值会被替换为 ***，其余内容保留（仍受 _truncate 限制）。
|
||||
"""
|
||||
if body is None:
|
||||
return ""
|
||||
|
||||
# 尝试解析 JSON 字符串
|
||||
parsed = body
|
||||
if isinstance(body, str):
|
||||
try:
|
||||
parsed = json.loads(body)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
# 非 JSON 字符串，直接截断
|
||||
return _truncate(body)
|
||||
|
||||
# dict 类型：遮罩敏感字段
|
||||
if isinstance(parsed, dict):
|
||||
safe = {}
|
||||
for k, v in parsed.items():
|
||||
if any(s in k.lower() for s in _SENSITIVE_BODY_KEYS):
|
||||
safe[k] = '***'
|
||||
elif isinstance(v, (dict, list)):
|
||||
safe[k] = _safe_body(v) if isinstance(v, dict) else _truncate(str(v))
|
||||
else:
|
||||
safe[k] = v
|
||||
return _truncate(json.dumps(safe, ensure_ascii=False))
|
||||
|
||||
# 其他类型：直接截断
|
||||
return _truncate(str(body))
|
||||
|
||||
|
||||
def log_http(
|
||||
category: str,
|
||||
method: str,
|
||||
url: str,
|
||||
*,
|
||||
request_headers: Any = None,
|
||||
request_body: Any = None,
|
||||
status_code: Optional[int] = None,
|
||||
response_headers: Any = None,
|
||||
response_body: Any = None,
|
||||
duration: Optional[float] = None,
|
||||
error: Optional[str] = None,
|
||||
proxy: Optional[str] = None,
|
||||
tag: str = "",
|
||||
) -> None:
|
||||
"""
|
||||
记录一条 HTTP 请求/响应详情日志。
|
||||
|
||||
Args:
|
||||
category: 分类（如 'douyu_login', 'geetest', 'proxy_verify', 'whitelist'）
|
||||
method: HTTP 方法
|
||||
url: 请求 URL（会被脱敏，移除敏感查询参数）
|
||||
request_headers: 请求头
|
||||
request_body: 请求体
|
||||
status_code: 响应状态码
|
||||
response_headers: 响应头
|
||||
response_body: 响应体
|
||||
duration: 耗时（秒）
|
||||
error: 错误信息
|
||||
proxy: 使用的代理
|
||||
tag: 额外标签（如账号名）
|
||||
"""
|
||||
entry = {
|
||||
"timestamp": datetime.now().isoformat(timespec="milliseconds"),
|
||||
"ts": time.time(),
|
||||
"category": category,
|
||||
"tag": tag,
|
||||
"method": method.upper(),
|
||||
"url": _truncate(url, 500),
|
||||
"proxy": proxy,
|
||||
"request": {
|
||||
"headers": _safe_headers(request_headers),
|
||||
"body": _safe_body(request_body),
|
||||
},
|
||||
"response": {
|
||||
"status_code": status_code,
|
||||
"headers": _safe_headers(response_headers),
|
||||
"body": _truncate(response_body),
|
||||
},
|
||||
"duration_ms": round(duration * 1000, 1) if duration is not None else None,
|
||||
"error": _truncate(error, 500) if error else None,
|
||||
}
|
||||
|
||||
# 判断级别
|
||||
if error or (status_code and status_code >= 400):
|
||||
entry["level"] = "error"
|
||||
elif status_code and status_code >= 300:
|
||||
entry["level"] = "warning"
|
||||
else:
|
||||
entry["level"] = "info"
|
||||
|
||||
try:
|
||||
line = json.dumps(entry, ensure_ascii=False)
|
||||
with _write_lock:
|
||||
with open(_LOG_FILE, "a", encoding="utf-8") as f:
|
||||
f.write(line + "\n")
|
||||
except Exception as e:
|
||||
logger.debug(f"写入HTTP详情日志失败: {e}")
|
||||
|
||||
|
||||
def read_http_logs(
|
||||
*,
|
||||
limit: int = 100,
|
||||
offset: int = 0,
|
||||
category: Optional[str] = None,
|
||||
level: Optional[str] = None,
|
||||
keyword: Optional[str] = None,
|
||||
) -> tuple[list[dict], int]:
|
||||
"""
|
||||
读取 HTTP 详情日志，支持筛选和分页。
|
||||
|
||||
Args:
|
||||
limit: 返回条数上限
|
||||
offset: 偏移量（从最新往前数）
|
||||
category: 按分类筛选
|
||||
level: 按级别筛选（info/warning/error）
|
||||
keyword: 关键词搜索（url、tag、error）
|
||||
|
||||
Returns:
|
||||
(日志条目列表, 总匹配条数)，列表按时间倒序（最新在前）
|
||||
"""
|
||||
if not _LOG_FILE.exists():
|
||||
return [], 0
|
||||
|
||||
entries: list[dict] = []
|
||||
try:
|
||||
with open(_LOG_FILE, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
entry = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
|
||||
# 筛选
|
||||
if category and entry.get("category") != category:
|
||||
continue
|
||||
if level and entry.get("level") != level:
|
||||
continue
|
||||
if keyword:
|
||||
kw = keyword.lower()
|
||||
searchable = " ".join([
|
||||
str(entry.get("url", "")),
|
||||
str(entry.get("tag", "")),
|
||||
str(entry.get("error", "")),
|
||||
str(entry.get("method", "")),
|
||||
]).lower()
|
||||
if kw not in searchable:
|
||||
continue
|
||||
|
||||
entries.append(entry)
|
||||
except Exception as e:
|
||||
logger.error(f"读取HTTP详情日志失败: {e}")
|
||||
return [], 0
|
||||
|
||||
# 按时间倒序
|
||||
entries.sort(key=lambda x: x.get("ts", 0), reverse=True)
|
||||
total = len(entries)
|
||||
# 分页（offset 从最新开始算）
|
||||
page = entries[offset:offset + limit]
|
||||
return page, total
|
||||
|
||||
|
||||
def clear_http_logs() -> int:
|
||||
"""清空 HTTP 详情日志，返回清空的条数"""
|
||||
count = 0
|
||||
with _write_lock:
|
||||
if _LOG_FILE.exists():
|
||||
try:
|
||||
with open(_LOG_FILE, "r", encoding="utf-8") as f:
|
||||
count = sum(1 for line in f if line.strip())
|
||||
except Exception:
|
||||
pass
|
||||
_LOG_FILE.write_text("", encoding="utf-8")
|
||||
return count
|
||||
Reference in New Issue
Block a user