Spaces:
Sleeping
Sleeping
File size: 3,651 Bytes
6303ae6 357ee36 6303ae6 357ee36 6303ae6 357ee36 6303ae6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 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 93 94 95 96 97 98 99 100 101 | """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
|