Spaces:
Paused
Paused
File size: 4,121 Bytes
9d7ddb9 | 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 | 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()
|