61 lines
1.6 KiB
Python
61 lines
1.6 KiB
Python
"""KEFU_WS_DEBUG=1 时把 IM 收/发原始 content 写入 backend/ws_media_debug.log。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
_ENABLED: bool | None = None
|
|
_LOG_PATH: Path | None = None
|
|
|
|
|
|
def _is_enabled() -> bool:
|
|
global _ENABLED
|
|
if _ENABLED is None:
|
|
_ENABLED = os.getenv("KEFU_WS_DEBUG", "").strip().lower() in ("1", "true", "yes")
|
|
return _ENABLED
|
|
|
|
|
|
def _log_path() -> Path:
|
|
global _LOG_PATH
|
|
if _LOG_PATH is None:
|
|
backend_dir = Path(__file__).resolve().parents[2]
|
|
_LOG_PATH = backend_dir / "ws_media_debug.log"
|
|
return _LOG_PATH
|
|
|
|
|
|
def log_im_message(
|
|
*,
|
|
direction: str,
|
|
message_type: int | str,
|
|
conversation_id: str = "",
|
|
content: Any = None,
|
|
fields: dict[str, Any] | None = None,
|
|
) -> None:
|
|
if not _is_enabled():
|
|
return
|
|
try:
|
|
ts = datetime.now().isoformat()
|
|
if isinstance(content, (dict, list)):
|
|
content_s = json.dumps(content, ensure_ascii=False, separators=(",", ":"))
|
|
elif content is None:
|
|
content_s = ""
|
|
else:
|
|
content_s = str(content)
|
|
fields_s = ""
|
|
if fields:
|
|
fields_s = " | fields=" + json.dumps(fields, ensure_ascii=False, separators=(",", ":"))
|
|
line = (
|
|
f"{ts} {direction} type={message_type} conv={conversation_id} "
|
|
f"content={content_s}{fields_s}\n"
|
|
)
|
|
path = _log_path()
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(path, "a", encoding="utf-8") as f:
|
|
f.write(line)
|
|
except Exception:
|
|
pass
|