Spaces:
Sleeping
Sleeping
| """Logging utilities. | |
| Logs API call metadata only — timestamps, model, token counts, and error | |
| codes. Never raw dossier text, screenshot content, or generated | |
| proposals. Logs rotate after 7 days. | |
| Redaction is *enforced at the sink*, not left to caller discipline: every | |
| log record passes through :class:`_RedactionFilter`, which scrubs | |
| API-key- and organization-ID-shaped tokens from the fully-rendered | |
| message regardless of which call site produced it. So even a careless | |
| ``logger.info("...%s", api_key)`` cannot write a secret to disk. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import threading | |
| from logging.handlers import TimedRotatingFileHandler | |
| from app.config import LOG_DIR, LOG_RETENTION_DAYS | |
| from app.utils.redaction import redact | |
| class _RedactionFilter(logging.Filter): | |
| """Scrub secrets from every record before it is formatted/written. | |
| The filter rewrites ``record.msg`` to the fully-interpolated, redacted | |
| message and clears ``record.args`` so the handler's formatter cannot | |
| re-introduce an unscrubbed value from the original args tuple. | |
| """ | |
| def filter(self, record: logging.LogRecord) -> bool: | |
| try: | |
| rendered = record.getMessage() | |
| except Exception: # noqa: BLE001 - never let logging raise | |
| return True | |
| cleaned = redact(rendered) | |
| if cleaned != rendered: | |
| record.msg = cleaned | |
| record.args = () | |
| return True | |
| _APP_LOGGER = "upwork_strategist" | |
| # Stops two sessions importing modules at once from each creating an | |
| # app.log handler. | |
| _setup_lock = threading.Lock() | |
| def _shared_file_handler() -> TimedRotatingFileHandler: | |
| """Return the one ``app.log`` handler, creating it on first use. | |
| Every logger must share it: two rotating handlers on the same file each | |
| roll it over at midnight, and the second rollover deletes the first | |
| one's dated backup (the previous day's log). It lives on the | |
| ``upwork_strategist`` logger, so a re-imported copy of this module still | |
| finds it. The caller must hold ``_setup_lock``. | |
| """ | |
| app_logger = logging.getLogger(_APP_LOGGER) | |
| for existing in app_logger.handlers: | |
| if isinstance(existing, TimedRotatingFileHandler): | |
| return existing | |
| LOG_DIR.mkdir(parents=True, exist_ok=True) | |
| handler = TimedRotatingFileHandler( | |
| LOG_DIR / "app.log", | |
| when="D", | |
| interval=1, | |
| backupCount=LOG_RETENTION_DAYS, | |
| encoding="utf-8", | |
| ) | |
| handler.setFormatter( | |
| logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s") | |
| ) | |
| handler.addFilter(_RedactionFilter()) | |
| app_logger.addHandler(handler) | |
| return handler | |
| def get_logger(name: str = _APP_LOGGER) -> logging.Logger: | |
| """Return a logger configured for redacted, metadata-only logging. | |
| ``upwork_strategist.*`` loggers reach the shared ``app.log`` handler by | |
| propagation, so every record is written once. | |
| """ | |
| logger = logging.getLogger(name) | |
| with _setup_lock: | |
| if any(isinstance(f, _RedactionFilter) for f in logger.filters): | |
| return logger | |
| handler = _shared_file_handler() | |
| in_app_tree = name == _APP_LOGGER or name.startswith(_APP_LOGGER + ".") | |
| if not in_app_tree and handler not in logger.handlers: | |
| logger.addHandler(handler) | |
| # Enforce redaction at the sink so no caller can leak a secret, even | |
| # by mistake. Attached to both the handler and the logger so records | |
| # are scrubbed regardless of propagation path. | |
| logger.addFilter(_RedactionFilter()) | |
| logger.setLevel(logging.INFO) | |
| return logger | |