logger_manager.py 1.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455
  1. from typing import Any
  2. from core.utils.log.aliyun_log import AliyunLogger
  3. from core.utils.log.local_log import Local
  4. from loguru._logger import Logger as LoguruLogger
  5. class LoggerManager:
  6. """
  7. 日志管理器
  8. """
  9. _local_loggers: dict[str, LoguruLogger] = {}
  10. _aliyun_loggers: dict[str, AliyunLogger] = {}
  11. @staticmethod
  12. def get_logger(
  13. platform: str="system",
  14. mode: str="crawler",
  15. log_to_console: bool = True
  16. ) -> LoguruLogger:
  17. key = f"{platform}_{mode}"
  18. if key not in LoggerManager._local_loggers:
  19. LoggerManager._local_loggers[key] = Local.init_logger(
  20. platform=platform,
  21. mode=mode,
  22. log_to_console=log_to_console
  23. )
  24. return LoggerManager._local_loggers[key]
  25. @staticmethod
  26. def get_aliyun_logger(
  27. platform: str = "system",
  28. mode: str = "crawler",
  29. env: str = "prod",
  30. project: str | None = None,
  31. logstore: str | None = None,
  32. endpoint: str | None = None,
  33. ) -> AliyunLogger:
  34. """
  35. :rtype: AliyunLogger
  36. """
  37. # 目的地属于缓存键,避免同一platform/mode的不同业务误用同一个Logstore。
  38. key = f"{platform}_{mode}_{env}_{project}_{logstore}_{endpoint}"
  39. if key not in LoggerManager._aliyun_loggers:
  40. LoggerManager._aliyun_loggers[key] = AliyunLogger(
  41. platform=platform,
  42. mode=mode,
  43. env=env,
  44. project=project,
  45. logstore=logstore,
  46. endpoint=endpoint,
  47. )
  48. return LoggerManager._aliyun_loggers[key]