cmshoppe/app/diagnostics.py

"""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 = appconfig.diagnostic_log_dir()
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")