111 lines
4.0 KiB
Python
111 lines
4.0 KiB
Python
import base64
|
|
import binascii
|
|
import json
|
|
import logging
|
|
import re
|
|
from typing import Any
|
|
|
|
|
|
class Logger:
|
|
_DATA_URL_RE = re.compile(r"data:image/[^;]+;base64,[A-Za-z0-9+/=]+")
|
|
_JSON_B64_RE = re.compile(r'("b64_json"\s*:\s*")([A-Za-z0-9+/=]+)(")')
|
|
|
|
def __init__(self, name: str = "chatgpt2api") -> None:
|
|
self._logger = logging.getLogger(name)
|
|
if not self._logger.handlers:
|
|
handler = logging.StreamHandler()
|
|
handler.setFormatter(logging.Formatter("[%(levelname)s] %(message)s"))
|
|
self._logger.addHandler(handler)
|
|
self._logger.setLevel(logging.DEBUG)
|
|
self._logger.propagate = False
|
|
|
|
def _enabled(self, level: str) -> bool:
|
|
try:
|
|
from services.config import config
|
|
levels = set(config.log_levels)
|
|
except Exception:
|
|
levels = set()
|
|
return level in (levels or {"info", "warning", "error"})
|
|
|
|
def _mask_string(self, value: str, keep: int = 10) -> str:
|
|
if len(value) <= keep:
|
|
return value
|
|
return value[:keep] + "..."
|
|
|
|
def _mask_base64(self, value: str) -> str:
|
|
if value.startswith("data:") and ";base64," in value:
|
|
header, _, data = value.partition(",")
|
|
return f"{header},{self._mask_string(data, 24)} (base64 len={len(data)})"
|
|
return f"{self._mask_string(value, 24)} (base64 len={len(value)})"
|
|
|
|
def _is_base64_string(self, value: str) -> bool:
|
|
if len(value) < 64 or len(value) % 4 != 0:
|
|
return False
|
|
if not any(char in value for char in "+/="):
|
|
return False
|
|
try:
|
|
base64.b64decode(value, validate=True)
|
|
return True
|
|
except (binascii.Error, ValueError):
|
|
return False
|
|
|
|
def _sanitize_string(self, value: str) -> str:
|
|
stripped = value.strip()
|
|
if stripped.startswith("data:") and ";base64," in stripped:
|
|
return self._mask_base64(stripped)
|
|
if self._is_base64_string(stripped):
|
|
return self._mask_base64(stripped)
|
|
sanitized = self._DATA_URL_RE.sub(lambda match: self._mask_base64(match.group(0)), value)
|
|
sanitized = self._JSON_B64_RE.sub(
|
|
lambda match: f'{match.group(1)}{self._mask_base64(match.group(2))}{match.group(3)}',
|
|
sanitized,
|
|
)
|
|
if sanitized != value:
|
|
return sanitized
|
|
return value
|
|
|
|
def _sanitize(self, value: Any) -> Any:
|
|
if isinstance(value, dict):
|
|
sanitized = {}
|
|
for key, item in value.items():
|
|
lowered_key = key.lower()
|
|
if isinstance(item, str) and ("token" in lowered_key or lowered_key == "dx"):
|
|
sanitized[key] = self._mask_string(item)
|
|
elif isinstance(item, str) and ("base64" in lowered_key or lowered_key == "b64_json"):
|
|
sanitized[key] = self._mask_base64(item)
|
|
else:
|
|
sanitized[key] = self._sanitize(item)
|
|
return sanitized
|
|
if isinstance(value, list):
|
|
return [self._sanitize(item) for item in value]
|
|
if isinstance(value, tuple):
|
|
return tuple(self._sanitize(item) for item in value)
|
|
if isinstance(value, str):
|
|
return self._sanitize_string(value)
|
|
return value
|
|
|
|
def _message(self, value: Any) -> str:
|
|
sanitized = self._sanitize(value)
|
|
if isinstance(sanitized, str):
|
|
return sanitized
|
|
return json.dumps(sanitized, ensure_ascii=False, default=str)
|
|
|
|
def debug(self, message: Any) -> None:
|
|
if self._enabled("debug"):
|
|
self._logger.debug(self._message(message))
|
|
|
|
def info(self, message: Any) -> None:
|
|
if self._enabled("info"):
|
|
self._logger.info(self._message(message))
|
|
|
|
def warning(self, message: Any) -> None:
|
|
if self._enabled("warning"):
|
|
self._logger.warning(self._message(message))
|
|
|
|
def error(self, message: Any) -> None:
|
|
if self._enabled("error"):
|
|
self._logger.error(self._message(message))
|
|
|
|
|
|
logger = Logger()
|