Spaces:
Paused
Paused
| 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() | |