""" logging_setup.py Central logging configuration for WickBot. Provides: - configure_logging(): sets up the root "wickbot" logger with a console handler (existing format), a rotating file handler (logs/wickbot.log), and a SQLite-backed handler (logs/wickbot.db, table `logs`). - get_logger(name): returns a child logger under the "wickbot" namespace. - profile(label, ...): a context manager AND decorator that times a block of work, logs the elapsed time, and (optionally) records a row in the `profiling` table. All handlers are defensive: if the DB or file sink fails, logging never crashes the application (errors are swallowed and, at most, printed to stderr once). Stdlib only: logging, logging.handlers, sqlite3, threading, time, json. """ import io import json import logging import logging.handlers import os import sqlite3 import sys import threading import time from contextlib import contextmanager from datetime import datetime, timezone from typing import Optional # --------------------------------------------------------------------------- # Paths # --------------------------------------------------------------------------- _BASE_DIR = os.path.dirname(os.path.abspath(__file__)) LOG_DIR = os.path.join(_BASE_DIR, "logs") LOG_FILE = os.path.join(LOG_DIR, "wickbot.log") DB_PATH = os.path.join(LOG_DIR, "wickbot.db") FORMAT = "%(asctime)s | %(levelname)s | %(name)s | %(message)s" FORMATTER = logging.Formatter(FORMAT) ROOT_LOGGER_NAME = "wickbot" # Module-level flag so configure_logging() is idempotent. _configured = False _config_lock = threading.Lock() # Shared SQLite connection for the DB log handler + profiling table. # Guarded by _db_lock for thread safety (sqlite3 connections are not # thread-safe by default). _db_conn: Optional[sqlite3.Connection] = None _db_lock = threading.Lock() _db_failed = False # latch: stop trying if the DB is permanently broken # --------------------------------------------------------------------------- # Schema # --------------------------------------------------------------------------- SCHEMA = """ CREATE TABLE IF NOT EXISTS logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT, level TEXT, logger TEXT, message TEXT, module TEXT, funcName TEXT, lineno INTEGER, elapsed_ms REAL, extra TEXT ); CREATE TABLE IF NOT EXISTS profiling ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT, label TEXT, elapsed_ms REAL, success INTEGER, detail TEXT ); """ def _ensure_db() -> Optional[sqlite3.Connection]: """Open (and lazily create) the SQLite DB + schema. Returns None on failure so callers can degrade gracefully.""" global _db_conn, _db_failed if _db_failed: return None with _db_lock: if _db_conn is not None: return _db_conn try: os.makedirs(LOG_DIR, exist_ok=True) conn = sqlite3.connect(DB_PATH, check_same_thread=False) conn.execute("PRAGMA journal_mode=WAL") conn.executescript(SCHEMA) conn.commit() _db_conn = conn return _db_conn except Exception as e: # noqa: BLE001 - never crash on logging setup _db_failed = True try: print(f"[logging_setup] DB handler disabled: {e}", file=sys.stderr) except Exception: pass return None # --------------------------------------------------------------------------- # DB log handler # --------------------------------------------------------------------------- class DatabaseLogHandler(logging.Handler): """Writes each log record as a row in the `logs` table. Inserts are committed per-record (lightweight, safe). If the DB is unavailable the handler silently drops records rather than raising. """ def emit(self, record: logging.LogRecord) -> None: global _db_failed if _db_failed: return conn = _ensure_db() if conn is None: return try: # elapsed_ms: time since the record's relative creation, if set # by the profiler via LogRecord factory; otherwise None. elapsed_ms = getattr(record, "elapsed_ms", None) extra = getattr(record, "extra", None) extra_text = json.dumps(extra, default=str) if isinstance(extra, (dict, list)) else (extra if isinstance(extra, str) else None) ts = datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat() conn.execute( """ INSERT INTO logs (timestamp, level, logger, message, module, funcName, lineno, elapsed_ms, extra) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( ts, record.levelname, record.name, record.getMessage(), record.module, record.funcName, record.lineno, elapsed_ms, extra_text, ), ) conn.commit() except Exception: # noqa: BLE001 - never let logging crash the app # Disable further DB attempts to avoid spamming on every record. _db_failed = True # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- def configure_logging(level: int = logging.INFO) -> None: """Configure the root "wickbot" logger with console + rotating file + SQLite handlers. Idempotent — safe to call multiple times.""" global _configured with _config_lock: if _configured: return os.makedirs(LOG_DIR, exist_ok=True) root = logging.getLogger(ROOT_LOGGER_NAME) root.setLevel(level) # Don't let records bubble to the root Python logger's default # handler (which would duplicate to stderr via basicConfig elsewhere). root.propagate = False # --- Console handler (existing format) --- # On Windows the default stdout is cp1252, which cannot encode emoji # (e.g. the red circle used in notifications) and raises # UnicodeEncodeError. Force UTF-8 with errors='replace' so logging # never crashes on non-ASCII content. try: console_stream = open( sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1, errors="replace", ) except Exception: # noqa: BLE001 - fall back to a safe wrapper console_stream = io.TextIOWrapper( sys.stdout.buffer, encoding="utf-8", errors="replace" ) console = logging.StreamHandler(console_stream) console.setFormatter(FORMATTER) root.addHandler(console) # --- Rotating file handler (10 MB, 5 backups) --- try: file_handler = logging.handlers.RotatingFileHandler( LOG_FILE, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8" ) file_handler.setFormatter(FORMATTER) root.addHandler(file_handler) except Exception as e: # noqa: BLE001 try: print(f"[logging_setup] File handler disabled: {e}", file=sys.stderr) except Exception: pass # --- Database handler --- db_handler = DatabaseLogHandler() db_handler.setFormatter(FORMATTER) # DB handler stores everything; keep it at the same level. db_handler.setLevel(level) root.addHandler(db_handler) _configured = True # Touch the DB so tables exist + file is created on first import. _ensure_db() get_logger("wickbot.logging_setup").info( "Logging configured: console + file (%s) + db (%s)", LOG_FILE, DB_PATH ) def get_logger(name: str) -> logging.Logger: """Return a logger under the wickbot namespace. Accepts either a bare name ("mt5") or a fully-qualified one ("wickbot.mt5"); both resolve under the root "wickbot" logger so they inherit the configured handlers. """ if not name.startswith(ROOT_LOGGER_NAME): name = f"{ROOT_LOGGER_NAME}.{name}" if name else ROOT_LOGGER_NAME return logging.getLogger(name) # --------------------------------------------------------------------------- # Profiling # --------------------------------------------------------------------------- def _record_profiling(label: str, elapsed_ms: float, success: bool, detail: Optional[str]) -> None: global _db_failed if _db_failed: return conn = _ensure_db() if conn is None: return try: ts = datetime.now(timezone.utc).isoformat() conn.execute( """ INSERT INTO profiling (timestamp, label, elapsed_ms, success, detail) VALUES (?, ?, ?, ?, ?) """, (ts, label, elapsed_ms, 1 if success else 0, detail), ) conn.commit() except Exception: # noqa: BLE001 _db_failed = True @contextmanager def profile(label: str, log_level: int = logging.DEBUG, detail: Optional[str] = None): """Context manager that times a block and logs + records the elapsed time. Usage: with profile("main_cycle"): ... do work ... On normal exit it logs the elapsed milliseconds and writes a profiling row (success=1). On exception it still records the elapsed time (success=0) and re-raises. """ log = get_logger("wickbot.profile") start = time.perf_counter() ok = True try: yield except Exception: ok = False elapsed_ms = (time.perf_counter() - start) * 1000.0 log.log(log_level, "profile[%s] failed after %.2f ms", label, elapsed_ms) _record_profiling(label, elapsed_ms, False, detail) raise else: elapsed_ms = (time.perf_counter() - start) * 1000.0 log.log(log_level, "profile[%s] elapsed=%.2f ms", label, elapsed_ms) _record_profiling(label, elapsed_ms, True, detail) def profile_call(label: str, log_level: int = logging.INFO, detail: Optional[str] = None): """Decorator variant of profile() for functions/methods. Usage: @profile_call("mt5_order_send") def send_market_order(self, ...): ... """ def decorator(func): def wrapper(*args, **kwargs): with profile(label or func.__name__, log_level=log_level, detail=detail): return func(*args, **kwargs) wrapper.__name__ = func.__name__ wrapper.__doc__ = func.__doc__ return wrapper return decorator # Auto-configure on first import so any module importing logging_setup # immediately gets file + DB persistence without an explicit call. configure_logging()