Spaces:
Sleeping
Sleeping
| """Simple structured logger.""" | |
| import json | |
| import sys | |
| from datetime import datetime | |
| from pathlib import Path | |
| class Logger: | |
| """Structured logging for cloud coder agents.""" | |
| def __init__(self, log_file: Path = None): | |
| self.log_file = log_file or Path("/tmp/cloud-coder.log") | |
| self.ensure_log_dir() | |
| def ensure_log_dir(self): | |
| """Ensure log directory exists.""" | |
| self.log_file.parent.mkdir(parents=True, exist_ok=True) | |
| def log(self, level: str, message: str, **kwargs): | |
| """Write a structured log entry.""" | |
| entry = { | |
| "timestamp": datetime.utcnow().isoformat() + "Z", | |
| "level": level, | |
| "message": message, | |
| **kwargs, | |
| } | |
| line = json.dumps(entry) | |
| print(line, file=sys.stderr) | |
| with open(self.log_file, "a") as f: | |
| f.write(line + "\n") | |
| def info(self, message: str, **kwargs): | |
| self.log("INFO", message, **kwargs) | |
| def warning(self, message: str, **kwargs): | |
| self.log("WARNING", message, **kwargs) | |
| def error(self, message: str, **kwargs): | |
| self.log("ERROR", message, **kwargs) | |
| def task(self, task_id: str, status: str, **kwargs): | |
| self.log( | |
| "TASK", f"Task {task_id} status: {status}", task_id=task_id, status=status, **kwargs | |
| ) | |
| # Global logger instance | |
| logger = Logger() | |