118 lines
3.5 KiB
Python
118 lines
3.5 KiB
Python
"""Diagnostic logging helpers for local debug logs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as _dt
|
|
import json
|
|
import os
|
|
import re
|
|
import traceback
|
|
|
|
from . import appconfig
|
|
|
|
DEFAULT_LOG_DIR = "logs"
|
|
DEFAULT_LOG_FILE = "cmshopee.log"
|
|
DEFAULT_MAX_BYTES = 2 * 1024 * 1024
|
|
DEFAULT_BACKUPS = 3
|
|
|
|
_SECRET_ASSIGNMENT_RE = re.compile(
|
|
r"(?i)\b(api[_-]?key|apikey|token|password|cookie|authorization)\b\s*[:=]\s*([^\s,;]+)"
|
|
)
|
|
_BEARER_RE = re.compile(r"(?i)\bBearer\s+[A-Za-z0-9._\-]+")
|
|
|
|
|
|
def _redact_text(value):
|
|
text = str(value)
|
|
text = _SECRET_ASSIGNMENT_RE.sub(lambda match: f"{match.group(1)}=***", text)
|
|
return _BEARER_RE.sub("Bearer ***", text)
|
|
|
|
|
|
def _sanitize_log_value(value):
|
|
value = appconfig.sanitize_for_log(value)
|
|
if isinstance(value, dict):
|
|
return {key: _sanitize_log_value(child) for key, child in value.items()}
|
|
if isinstance(value, list):
|
|
return [_sanitize_log_value(item) for item in value]
|
|
if isinstance(value, tuple):
|
|
return tuple(_sanitize_log_value(item) for item in value)
|
|
if isinstance(value, str):
|
|
return _redact_text(value)
|
|
return value
|
|
|
|
|
|
def redact_log_text(value):
|
|
"""Return free-form diagnostic text with common secret assignments redacted."""
|
|
|
|
return _redact_text(value)
|
|
|
|
|
|
def diagnostic_log_path(log_dir=None, filename=DEFAULT_LOG_FILE):
|
|
directory = os.path.abspath(log_dir or DEFAULT_LOG_DIR)
|
|
return os.path.join(directory, filename)
|
|
|
|
|
|
def write_diagnostic_log(
|
|
message,
|
|
*,
|
|
level="INFO",
|
|
step=None,
|
|
task_id=None,
|
|
alias=None,
|
|
item_id=None,
|
|
elapsed_ms=None,
|
|
payload=None,
|
|
exc=None,
|
|
log_dir=None,
|
|
log_path=None,
|
|
max_bytes=DEFAULT_MAX_BYTES,
|
|
backups=DEFAULT_BACKUPS,
|
|
):
|
|
"""Append a sanitized diagnostic log entry and return the log path."""
|
|
|
|
path = os.path.abspath(log_path or diagnostic_log_path(log_dir))
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
_rotate_if_needed(path, max_bytes=max_bytes, backups=backups)
|
|
|
|
entry = {
|
|
"time": _dt.datetime.now().isoformat(timespec="seconds"),
|
|
"level": str(level or "INFO").upper(),
|
|
"message": _redact_text(message),
|
|
}
|
|
if step is not None:
|
|
entry["step"] = _redact_text(step)
|
|
if task_id is not None:
|
|
entry["task_id"] = task_id
|
|
if alias is not None:
|
|
entry["alias"] = _redact_text(alias)
|
|
if item_id is not None:
|
|
entry["item_id"] = _redact_text(item_id)
|
|
if elapsed_ms is not None:
|
|
entry["elapsed_ms"] = int(elapsed_ms)
|
|
if payload is not None:
|
|
entry["payload"] = _sanitize_log_value(payload)
|
|
if exc is not None:
|
|
entry["exception"] = exc.__class__.__name__
|
|
tb = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
|
|
entry["traceback"] = _redact_text(appconfig.redact_secrets(tb))
|
|
|
|
safe_entry = _sanitize_log_value(entry)
|
|
with open(path, "a", encoding="utf-8") as fh:
|
|
fh.write(json.dumps(safe_entry, ensure_ascii=False, sort_keys=True))
|
|
fh.write("\n")
|
|
return path
|
|
|
|
|
|
def _rotate_if_needed(path, max_bytes=DEFAULT_MAX_BYTES, backups=DEFAULT_BACKUPS):
|
|
if max_bytes <= 0 or backups <= 0:
|
|
return
|
|
if not os.path.exists(path) or os.path.getsize(path) < max_bytes:
|
|
return
|
|
oldest = f"{path}.{int(backups)}"
|
|
if os.path.exists(oldest):
|
|
os.remove(oldest)
|
|
for index in range(int(backups) - 1, 0, -1):
|
|
source = f"{path}.{index}"
|
|
target = f"{path}.{index + 1}"
|
|
if os.path.exists(source):
|
|
os.replace(source, target)
|
|
os.replace(path, f"{path}.1") |