Spaces:
Sleeping
Sleeping
File size: 3,942 Bytes
4eafa75 b90c16e 4eafa75 b90c16e 4eafa75 | 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 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 | """
Centralized logger for Job Automation Agent.
Writes to data/logs/run_YYYY-MM-DD_HH-MM-SS.log
Also keeps last N lines in memory for the UI.
"""
import logging
import os
import sys
from datetime import datetime
from pathlib import Path
from collections import deque
LOG_DIR = Path("data/logs")
_memory_buffer: deque = deque(maxlen=500) # last 500 lines in memory
_current_log_file: str = ""
def setup(run_id: str = "") -> str:
"""
Set up file + console logging. Returns path to log file.
Call once at the start of each pipeline run.
"""
global _current_log_file, _memory_buffer
_memory_buffer.clear()
LOG_DIR.mkdir(parents=True, exist_ok=True)
tag = run_id or datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
log_path = str(LOG_DIR / f"run_{tag}.log")
_current_log_file = log_path
# Root logger
root = logging.getLogger()
root.setLevel(logging.DEBUG)
# Remove old handlers
for h in root.handlers[:]:
root.removeHandler(h)
fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(name)s — %(message)s",
datefmt="%H:%M:%S")
# File handler
fh = logging.FileHandler(log_path, encoding="utf-8")
fh.setLevel(logging.DEBUG)
fh.setFormatter(fmt)
root.addHandler(fh)
# Memory handler (for UI display)
mh = _MemoryHandler()
mh.setLevel(logging.DEBUG)
mh.setFormatter(fmt)
root.addHandler(mh)
# Force UTF-8 on stdout/stderr so unicode chars (✓ → etc.) don't crash on Windows cp1252
try:
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
except AttributeError:
pass # already replaced by Streamlit or not a TextIOWrapper
# Redirect stdout/stderr so print() and Playwright output are captured in log file
sys.stdout = _TeeStream(sys.stdout, log_path)
sys.stderr = _TeeStream(sys.stderr, log_path)
return log_path
def get_log_file() -> str:
return _current_log_file
def get_memory_lines() -> list[str]:
return list(_memory_buffer)
def read_log_file(tail: int = 200) -> str:
"""Read last N lines of current log file."""
if not _current_log_file or not os.path.exists(_current_log_file):
return ""
try:
with open(_current_log_file, "r", encoding="utf-8", errors="replace") as f:
lines = f.readlines()
return "".join(lines[-tail:])
except Exception:
return ""
def list_log_files() -> list[str]:
"""Return all log files, newest first."""
if not LOG_DIR.exists():
return []
return sorted(
[str(p) for p in LOG_DIR.glob("run_*.log")],
reverse=True,
)
class _MemoryHandler(logging.Handler):
def emit(self, record):
try:
_memory_buffer.append(self.format(record))
except Exception:
pass
class _TeeStream:
"""Writes to both original stream and log file."""
def __init__(self, original, log_path: str):
self._orig = original
self._log = log_path
def write(self, text):
try:
self._orig.write(text)
except (UnicodeEncodeError, UnicodeDecodeError):
# Windows cp1252 console can't handle unicode — write safe version
try:
self._orig.write(text.encode("utf-8", errors="replace").decode("ascii", errors="replace"))
except Exception:
pass
except Exception:
pass
if text.strip():
try:
with open(self._log, "a", encoding="utf-8", errors="replace") as f:
f.write(text)
except Exception:
pass
def flush(self):
try:
self._orig.flush()
except Exception:
pass
def fileno(self):
return self._orig.fileno()
def isatty(self):
return False
|