Spaces:
Running
Running
File size: 3,968 Bytes
0157ac7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | """Loguru-based structured logging configuration.
All logs are written to server.log as JSON lines for full traceability.
Stdlib logging is intercepted and funneled to loguru.
Context vars (request_id, node_id, chat_id) from contextualize() are
included at top level for easy grep/filter.
"""
import json
import logging
import re
from pathlib import Path
from loguru import logger
_configured = False
# Context keys we promote to top-level JSON for traceability
_CONTEXT_KEYS = ("request_id", "node_id", "chat_id")
_TELEGRAM_BOT_RE = re.compile(
r"(https?://api\.telegram\.org/)bot([0-9]+:[A-Za-z0-9_-]+)(/?)",
re.IGNORECASE,
)
# Authorization: Bearer <token> (HTTP client / proxy debug lines)
_AUTH_BEARER_RE = re.compile(
r"(\bAuthorization\s*:\s*Bearer\s+)([^\s'\"]+)",
re.IGNORECASE,
)
def _redact_sensitive_substrings(message: str) -> str:
"""Remove obvious API tokens and secrets before JSON log line emission."""
text = _TELEGRAM_BOT_RE.sub(r"\1bot<redacted>\3", message)
return _AUTH_BEARER_RE.sub(r"\1<redacted>", text)
def _serialize_with_context(record) -> str:
"""Format record as JSON with context vars at top level.
Returns a format template; we inject _json into record for output.
"""
extra = record.get("extra", {})
out = {
"time": str(record["time"]),
"level": record["level"].name,
"message": _redact_sensitive_substrings(str(record["message"])),
"module": record["name"],
"function": record["function"],
"line": record["line"],
}
for key in _CONTEXT_KEYS:
if key in extra and extra[key] is not None:
out[key] = extra[key]
record["_json"] = json.dumps(out, default=str)
return "{_json}\n"
class InterceptHandler(logging.Handler):
"""Redirect stdlib logging to loguru."""
def emit(self, record: logging.LogRecord) -> None:
try:
level = logger.level(record.levelname).name
except ValueError:
level = record.levelno
frame, depth = logging.currentframe(), 2
while frame is not None and frame.f_code.co_filename == logging.__file__:
frame = frame.f_back
depth += 1
logger.opt(depth=depth, exception=record.exc_info).log(
level, record.getMessage()
)
def configure_logging(
log_file: str, *, force: bool = False, verbose_third_party: bool = False
) -> None:
"""Configure loguru with JSON output to log_file and intercept stdlib logging.
Idempotent: skips if already configured (e.g. hot reload).
Use force=True to reconfigure (e.g. in tests with a different log path).
When ``verbose_third_party`` is false, noisy HTTP and Telegram loggers are capped
at WARNING unless explicitly configured otherwise.
"""
global _configured
if _configured and not force:
return
_configured = True
# Remove default loguru handler (writes to stderr)
logger.remove()
# Truncate log file on fresh start for clean debugging if possible
try:
Path(log_file).write_text("")
except PermissionError:
# File might be open by another process (e.g. redirection)
pass
# Add file sink: JSON lines, DEBUG level, context vars at top level
logger.add(
log_file,
level="DEBUG",
format=_serialize_with_context,
encoding="utf-8",
mode="a",
rotation="50 MB",
)
# Intercept stdlib logging: route all root logger output to loguru
intercept = InterceptHandler()
logging.root.handlers = [intercept]
logging.root.setLevel(logging.DEBUG)
third_party = (
"httpx",
"httpcore",
"httpcore.http11",
"httpcore.connection",
"telegram",
"telegram.ext",
)
for name in third_party:
logging.getLogger(name).setLevel(
logging.WARNING if not verbose_third_party else logging.NOTSET
)
|