Skip to content

core.logger

from core.logger import *

core.logger

CLR_FAIL = '\x1b[91m' module-attribute

CLR_INFO = '\x1b[94m' module-attribute

CLR_MAGENTA = '\x1b[95m' module-attribute

CLR_RESET = '\x1b[0m' module-attribute

CLR_SUCCESS = '\x1b[92m' module-attribute

CLR_WARN = '\x1b[93m' module-attribute

AppLogger

Source code in core/logger.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
class AppLogger:
    _log_dir = "logs"
    _structured_log_file: str | None = None

    @classmethod
    def initialize(cls, log_dir: str = "logs") -> None:
        cls._log_dir = log_dir
        if not os.path.exists(log_dir):
            os.makedirs(log_dir)

        log_filename = f"{log_dir}/{datetime.now().strftime('%Y-%m-%d')}.log"
        cls._structured_log_file = f"{log_dir}/{datetime.now().strftime('%Y-%m-%d')}.jsonl"

        logging.basicConfig(
            level=logging.INFO,
            format="%(asctime)s [%(levelname)s] %(message)s",
            handlers=[
                logging.FileHandler(log_filename, encoding="utf-8"),
                logging.NullHandler(),
            ],
        )

    @classmethod
    def _write_structured(cls, record: StructuredLogRecord) -> None:
        if cls._structured_log_file:
            try:
                with open(cls._structured_log_file, "a", encoding="utf-8") as f:
                    f.write(json.dumps(record.to_dict()) + "\n")
            except Exception:
                pass

    @classmethod
    def info(cls, msg: str, **meta: Any) -> None:
        safe = redact_secrets(msg)
        record = StructuredLogRecord("INFO", safe, **meta)
        logging.info(safe)
        cls._write_structured(record)
        print(f"{CLR_INFO}[INFO] [{datetime.now().strftime('%H:%M:%S')}] {safe}{CLR_RESET}")

    @classmethod
    def success(cls, msg: str, **meta: Any) -> None:
        safe = redact_secrets(msg)
        record = StructuredLogRecord("SUCCESS", safe, **meta)
        logging.info(f"SUCCESS: {safe}")
        cls._write_structured(record)
        print(f"{CLR_SUCCESS}[SUCC] [{datetime.now().strftime('%H:%M:%S')}] {safe}{CLR_RESET}")

    @classmethod
    def warn(cls, msg: str, **meta: Any) -> None:
        safe = redact_secrets(msg)
        record = StructuredLogRecord("WARN", safe, **meta)
        logging.warning(safe)
        cls._write_structured(record)
        print(f"{CLR_WARN}[WARN] [{datetime.now().strftime('%H:%M:%S')}] {safe}{CLR_RESET}")

    @classmethod
    def error(cls, msg: str, **meta: Any) -> None:
        safe = redact_secrets(msg)
        record = StructuredLogRecord("ERROR", safe, **meta)
        logging.error(safe, exc_info=True)
        cls._write_structured(record)
        print(f"{CLR_FAIL}[ERR!] [{datetime.now().strftime('%H:%M:%S')}] {safe}{CLR_RESET}")

    @classmethod
    def debug(cls, msg: str) -> None:
        print(f"{CLR_MAGENTA}[DEBG] {redact_secrets(msg)}{CLR_RESET}")

debug(msg) classmethod

Source code in core/logger.py 
158
159
160
@classmethod
def debug(cls, msg: str) -> None:
    print(f"{CLR_MAGENTA}[DEBG] {redact_secrets(msg)}{CLR_RESET}")

error(msg, **meta) classmethod

Source code in core/logger.py 
150
151
152
153
154
155
156
@classmethod
def error(cls, msg: str, **meta: Any) -> None:
    safe = redact_secrets(msg)
    record = StructuredLogRecord("ERROR", safe, **meta)
    logging.error(safe, exc_info=True)
    cls._write_structured(record)
    print(f"{CLR_FAIL}[ERR!] [{datetime.now().strftime('%H:%M:%S')}] {safe}{CLR_RESET}")

info(msg, **meta) classmethod

Source code in core/logger.py 
126
127
128
129
130
131
132
@classmethod
def info(cls, msg: str, **meta: Any) -> None:
    safe = redact_secrets(msg)
    record = StructuredLogRecord("INFO", safe, **meta)
    logging.info(safe)
    cls._write_structured(record)
    print(f"{CLR_INFO}[INFO] [{datetime.now().strftime('%H:%M:%S')}] {safe}{CLR_RESET}")

initialize(log_dir='logs') classmethod

Source code in core/logger.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@classmethod
def initialize(cls, log_dir: str = "logs") -> None:
    cls._log_dir = log_dir
    if not os.path.exists(log_dir):
        os.makedirs(log_dir)

    log_filename = f"{log_dir}/{datetime.now().strftime('%Y-%m-%d')}.log"
    cls._structured_log_file = f"{log_dir}/{datetime.now().strftime('%Y-%m-%d')}.jsonl"

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
        handlers=[
            logging.FileHandler(log_filename, encoding="utf-8"),
            logging.NullHandler(),
        ],
    )

success(msg, **meta) classmethod

Source code in core/logger.py 
134
135
136
137
138
139
140
@classmethod
def success(cls, msg: str, **meta: Any) -> None:
    safe = redact_secrets(msg)
    record = StructuredLogRecord("SUCCESS", safe, **meta)
    logging.info(f"SUCCESS: {safe}")
    cls._write_structured(record)
    print(f"{CLR_SUCCESS}[SUCC] [{datetime.now().strftime('%H:%M:%S')}] {safe}{CLR_RESET}")

warn(msg, **meta) classmethod

Source code in core/logger.py 
142
143
144
145
146
147
148
@classmethod
def warn(cls, msg: str, **meta: Any) -> None:
    safe = redact_secrets(msg)
    record = StructuredLogRecord("WARN", safe, **meta)
    logging.warning(safe)
    cls._write_structured(record)
    print(f"{CLR_WARN}[WARN] [{datetime.now().strftime('%H:%M:%S')}] {safe}{CLR_RESET}")

StructuredLogRecord

Source code in core/logger.py 
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
class StructuredLogRecord:
    __slots__ = ("timestamp", "level", "message", "recipient", "account", "status", "latency", "correlation_id", "extra")

    def __init__(
        self,
        level: str,
        message: str,
        recipient: str | None = None,
        account: str | None = None,
        status: str | None = None,
        latency: float | None = None,
        correlation_id: str | None = None,
        extra: dict[str, Any] | None = None,
    ) -> None:
        self.timestamp = datetime.now().isoformat()
        self.level = level
        self.message = message
        self.recipient = recipient
        self.account = account
        self.status = status
        self.latency = latency
        self.correlation_id = correlation_id
        self.extra = extra or {}

    def to_dict(self) -> dict[str, Any]:
        d: dict[str, Any] = {
            "timestamp": self.timestamp,
            "level": self.level,
            "message": self.message,
        }
        if self.recipient:
            d["recipient"] = self.recipient
        if self.account:
            d["account"] = self.account
        if self.status:
            d["status"] = self.status
        if self.latency is not None:
            d["latency"] = self.latency
        if self.correlation_id:
            d["correlation_id"] = self.correlation_id
        if self.extra:
            d["extra"] = self.extra
        return d

__slots__ = ('timestamp', 'level', 'message', 'recipient', 'account', 'status', 'latency', 'correlation_id', 'extra') class-attribute instance-attribute

account = account instance-attribute

correlation_id = correlation_id instance-attribute

extra = extra or {} instance-attribute

latency = latency instance-attribute

level = level instance-attribute

message = message instance-attribute

recipient = recipient instance-attribute

status = status instance-attribute

timestamp = datetime.now().isoformat() instance-attribute

__init__(level, message, recipient=None, account=None, status=None, latency=None, correlation_id=None, extra=None)

Source code in core/logger.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def __init__(
    self,
    level: str,
    message: str,
    recipient: str | None = None,
    account: str | None = None,
    status: str | None = None,
    latency: float | None = None,
    correlation_id: str | None = None,
    extra: dict[str, Any] | None = None,
) -> None:
    self.timestamp = datetime.now().isoformat()
    self.level = level
    self.message = message
    self.recipient = recipient
    self.account = account
    self.status = status
    self.latency = latency
    self.correlation_id = correlation_id
    self.extra = extra or {}

to_dict()

Source code in core/logger.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def to_dict(self) -> dict[str, Any]:
    d: dict[str, Any] = {
        "timestamp": self.timestamp,
        "level": self.level,
        "message": self.message,
    }
    if self.recipient:
        d["recipient"] = self.recipient
    if self.account:
        d["account"] = self.account
    if self.status:
        d["status"] = self.status
    if self.latency is not None:
        d["latency"] = self.latency
    if self.correlation_id:
        d["correlation_id"] = self.correlation_id
    if self.extra:
        d["extra"] = self.extra
    return d

redact_secrets(text)

Source code in core/logger.py 
43
44
45
46
47
def redact_secrets(text: str) -> str:
    _init_secret_patterns()
    for pattern, replacement in _SECRET_PATTERNS:
        text = pattern.sub(replacement, text)
    return text