Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import json | |
| import os | |
| import time | |
| import uuid | |
| from typing import Any, Literal | |
| from pydantic import BaseModel, Field | |
| ActivityStatus = Literal["ok", "running", "warning", "error"] | |
| _DEFAULT_ROOT = os.path.expanduser("~/.studybuddy/projects/activity") | |
| class ProjectActivityEvent(BaseModel): | |
| event_id: str = Field(default_factory=lambda: uuid.uuid4().hex) | |
| project_id: str | |
| event_type: str | |
| message: str | |
| status: ActivityStatus = "ok" | |
| created_at: float = Field(default_factory=time.time) | |
| metadata: dict[str, Any] = Field(default_factory=dict) | |
| class ProjectActivityService: | |
| def __init__(self, root: str | None = None) -> None: | |
| self.root = root or _DEFAULT_ROOT | |
| os.makedirs(self.root, exist_ok=True) | |
| def _path(self, project_id: str) -> str: | |
| safe = project_id.replace("/", "_").replace("\\", "_").replace("..", "_") | |
| return os.path.join(self.root, f"{safe}.jsonl") | |
| def record( | |
| self, | |
| project_id: str, | |
| event_type: str, | |
| message: str, | |
| status: ActivityStatus = "ok", | |
| metadata: dict[str, Any] | None = None, | |
| ) -> ProjectActivityEvent: | |
| event = ProjectActivityEvent( | |
| project_id=project_id, | |
| event_type=event_type, | |
| message=message, | |
| status=status, | |
| metadata=metadata or {}, | |
| ) | |
| with open(self._path(project_id), "a", encoding="utf-8") as f: | |
| f.write(json.dumps(event.model_dump(), ensure_ascii=True) + "\n") | |
| return event | |
| def list(self, project_id: str, limit: int = 50) -> list[ProjectActivityEvent]: | |
| path = self._path(project_id) | |
| if not os.path.exists(path): | |
| return [] | |
| rows: list[tuple[int, ProjectActivityEvent]] = [] | |
| with open(path, encoding="utf-8") as f: | |
| for index, line in enumerate(f): | |
| if not line.strip(): | |
| continue | |
| try: | |
| rows.append((index, ProjectActivityEvent(**json.loads(line)))) | |
| except Exception: | |
| continue | |
| rows.sort(key=lambda row: (row[1].created_at, row[0]), reverse=True) | |
| return [row for _, row in rows[:limit]] | |
| def record_llm_usage( | |
| self, | |
| project_id: str, | |
| operation: str, | |
| tokens: int | float, | |
| elapsed_seconds: float, | |
| model: str, | |
| ) -> ProjectActivityEvent: | |
| tokens_per_second = round(float(tokens) / elapsed_seconds, 1) if elapsed_seconds > 0 else 0 | |
| return self.record( | |
| project_id, | |
| "llm_usage", | |
| f"{operation} generated {int(tokens)} tokens", | |
| metadata={ | |
| "operation": operation, | |
| "tokens": int(tokens), | |
| "elapsed_seconds": round(elapsed_seconds, 3), | |
| "tokens_per_second": tokens_per_second, | |
| "model": model, | |
| }, | |
| ) | |