79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
"""
|
|
AutoDev - Action Logger & Worklog
|
|
Logs every action to console and persistent worklog for resumability.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime
|
|
|
|
from . import config
|
|
|
|
|
|
class Logger:
|
|
def __init__(self, workdir: str):
|
|
self.workdir = workdir
|
|
self.log_path = os.path.join(workdir, config.WORKLOG_FILE)
|
|
self.entries: list[dict] = []
|
|
self._load()
|
|
|
|
def _load(self):
|
|
if os.path.exists(self.log_path):
|
|
try:
|
|
with open(self.log_path, "r") as f:
|
|
self.entries = json.load(f)
|
|
except (json.JSONDecodeError, IOError):
|
|
self.entries = []
|
|
|
|
def _save(self):
|
|
try:
|
|
os.makedirs(os.path.dirname(self.log_path) or ".", exist_ok=True)
|
|
with open(self.log_path, "w") as f:
|
|
json.dump(self.entries, f, indent=2)
|
|
except OSError:
|
|
pass # If we truly can't write, don't crash the whole process
|
|
|
|
def log(self, action: str, detail: str = "", status: str = "ok"):
|
|
entry = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"action": action,
|
|
"detail": detail[:3000],
|
|
"status": status,
|
|
}
|
|
self.entries.append(entry)
|
|
self._save()
|
|
# Console output with visual indicators
|
|
icons = {"ok": "✓", "error": "✗", "warn": "⚠"}
|
|
icon = icons.get(status, "…")
|
|
# Color: green for ok, red for error, yellow for warn
|
|
colors = {"ok": "\033[32m", "error": "\033[31m", "warn": "\033[33m"}
|
|
reset = "\033[0m"
|
|
color = colors.get(status, "")
|
|
print(f" {color}[{icon}]{reset} {action}: {detail[:200]}")
|
|
# Push to web UI if running
|
|
try:
|
|
from .web import push_event
|
|
push_event("log", {"action": action, "detail": detail[:500], "status": status})
|
|
except Exception:
|
|
pass
|
|
|
|
def get_recent(self, n: int = 20) -> list[dict]:
|
|
return self.entries[-n:]
|
|
|
|
def get_all(self) -> list[dict]:
|
|
return list(self.entries)
|
|
|
|
def get_last_phase(self) -> str | None:
|
|
for e in reversed(self.entries):
|
|
if e["action"] == "phase_complete":
|
|
return e["detail"]
|
|
return None
|
|
|
|
def get_errors_since(self, n_entries_back: int = 50) -> list[str]:
|
|
"""Get recent error details for debugging context."""
|
|
errors = []
|
|
for e in self.entries[-n_entries_back:]:
|
|
if e["status"] == "error":
|
|
errors.append(f"[{e['action']}] {e['detail']}")
|
|
return errors
|