115 lines
3.1 KiB
Python
115 lines
3.1 KiB
Python
import logging
|
|
import json
|
|
import pathlib
|
|
from functools import cache
|
|
from enum import Enum
|
|
from logging.handlers import RotatingFileHandler
|
|
|
|
from monitor_utils.constants import (
|
|
PROXIES_PATH,
|
|
PROJECT_ROOT,
|
|
PRODUCTION_BOT_TOKEN,
|
|
DEVELOPMENT_BOT_TOKEN,
|
|
LOGGING_LEVEL,
|
|
)
|
|
|
|
RESERVED = frozenset(logging.LogRecord("", 0, "", 0, "", (), None).__dict__)
|
|
|
|
|
|
class DevelopmentMode(str, Enum):
|
|
PRODUCTION = "PRODUCTION"
|
|
DEVELOPMENT = "DEVELOPMENT"
|
|
|
|
|
|
class JSONFormatter(logging.Formatter):
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
payload = {
|
|
"time": self.formatTime(record),
|
|
"level": record.levelname,
|
|
"logger": record.name,
|
|
"function": record.funcName,
|
|
"message": record.getMessage(),
|
|
}
|
|
for key, value in record.__dict__.items():
|
|
if key not in RESERVED and not key.startswith("_"):
|
|
payload[key] = value
|
|
|
|
if record.exc_info:
|
|
payload["exc"] = self.formatException(record.exc_info)
|
|
|
|
return json.dumps(payload)
|
|
|
|
|
|
def configure_logging(monitor_name: str, log_dir: str = "/logs"):
|
|
handler = logging.StreamHandler()
|
|
handler.setFormatter(JSONFormatter())
|
|
|
|
def inject_monitor(record: logging.LogRecord) -> bool:
|
|
record.monitor_name = monitor_name
|
|
return True
|
|
|
|
handler.addFilter(inject_monitor)
|
|
|
|
file_handler = RotatingFileHandler(
|
|
f"{log_dir}/{monitor_name}.jsonl",
|
|
maxBytes=10 * 1024 * 1024,
|
|
backupCount=5,
|
|
)
|
|
file_handler.setFormatter(JSONFormatter())
|
|
file_handler.addFilter(inject_monitor)
|
|
|
|
root = logging.getLogger()
|
|
root.handlers.clear()
|
|
root.setLevel(LOGGING_LEVEL)
|
|
root.addHandler(handler)
|
|
root.addHandler(file_handler)
|
|
|
|
logging.getLogger("aiohttp").setLevel(logging.WARNING)
|
|
logging.getLogger("pymongo").setLevel(logging.WARNING)
|
|
|
|
|
|
def is_production(development_mode: DevelopmentMode) -> bool:
|
|
return development_mode == DevelopmentMode.PRODUCTION
|
|
|
|
|
|
def get_bot_token(is_production: bool) -> str:
|
|
if is_production:
|
|
return PRODUCTION_BOT_TOKEN
|
|
|
|
return DEVELOPMENT_BOT_TOKEN
|
|
|
|
|
|
@cache
|
|
def get_proxies() -> list[str]:
|
|
with PROXIES_PATH.open() as f:
|
|
proxies = []
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
parts = line.split(":", 3)
|
|
if len(parts) == 4 and not parts[1].startswith("//"):
|
|
host, port, user, password = parts
|
|
proxies.append(f"http://{user}:{password}@{host}:{port}")
|
|
else:
|
|
proxies.append(f"http://{line}")
|
|
|
|
if not proxies:
|
|
raise ValueError("Proxy list cannot be empty")
|
|
|
|
return proxies
|
|
|
|
|
|
def get_config(
|
|
monitor_name: str, use_production_path: bool = False
|
|
) -> tuple[dict, pathlib.Path]:
|
|
monitor_name = monitor_name.replace(".json", "")
|
|
if use_production_path:
|
|
config_path = PROJECT_ROOT / "configs" / f"{monitor_name}.json"
|
|
else:
|
|
config_path = PROJECT_ROOT / "tests" / "configs" / f"{monitor_name}.json"
|
|
|
|
with open(config_path) as f:
|
|
return json.load(f), config_path
|