Spaces:
Sleeping
Sleeping
File size: 2,972 Bytes
2e818da | 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 | 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,
},
)
|