"""Simple shared log storage for Gradle build output."""
from collections import deque
from datetime import datetime
from threading import Lock
from typing import Any
# Module-level shared storage - this will be shared across all imports
_logs: deque = deque(maxlen=5000)
_lock = Lock()
# Track active builds
_active_builds: dict[str, dict[str, Any]] = {} # build_id -> {task, start_time, ...}
_build_counter = 0
def start_build(task: str) -> str:
"""Mark a build as started.
Args:
task: The task(s) being run.
Returns:
Build ID for tracking.
"""
global _build_counter
with _lock:
_build_counter += 1
build_id = f"build_{_build_counter}"
_active_builds[build_id] = {
"id": build_id,
"task": task,
"start_time": datetime.now().isoformat(),
}
# Notify dashboard
_emit_builds_update()
return build_id
def end_build(build_id: str) -> None:
"""Mark a build as completed.
Args:
build_id: The build ID from start_build.
"""
with _lock:
if build_id in _active_builds:
del _active_builds[build_id]
# Notify dashboard
_emit_builds_update()
def get_active_builds() -> list[dict[str, Any]]:
"""Get list of active builds.
Returns:
List of active build dictionaries.
"""
with _lock:
return list(_active_builds.values())
def _emit_builds_update() -> None:
"""Emit builds update via WebSocket."""
try:
from .app import emit_builds_changed
emit_builds_changed()
except Exception:
pass
def add_log(message: str, level: str = "INFO") -> None:
"""Add a log entry.
Args:
message: Log message.
level: Log level (INFO, ERROR, WARN, DEBUG).
"""
if not message.strip():
return
entry = {
"timestamp": datetime.now().isoformat(),
"level": level,
"message": message,
}
with _lock:
_logs.append(entry)
# Emit WebSocket event for real-time update
try:
from .app import emit_log
emit_log(entry)
except Exception:
pass # Ignore if socketio not available
def get_logs(limit: int = 500) -> list[dict[str, Any]]:
"""Get recent logs.
Args:
limit: Maximum number of log entries to return.
Returns:
List of log entry dictionaries.
"""
with _lock:
logs = list(_logs)
return logs[-limit:] if len(logs) > limit else logs
def clear_logs() -> None:
"""Clear all logs."""
with _lock:
_logs.clear()
def log_count() -> int:
"""Get the number of logs."""
with _lock:
return len(_logs)
