feat: project directory management — create, log, read, list
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,111 @@
|
||||
"""Project directory management — create, log, read, list."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
DEFAULT_WORKBENCH_DIR = Path.home() / "Kitty-Workbench"
|
||||
|
||||
|
||||
def _now_iso() -> str:
|
||||
return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
|
||||
|
||||
|
||||
def project_path(name: str, workbench_dir: Path = DEFAULT_WORKBENCH_DIR) -> Path:
|
||||
return workbench_dir / name
|
||||
|
||||
|
||||
def project_exists(name: str, workbench_dir: Path = DEFAULT_WORKBENCH_DIR) -> bool:
|
||||
return project_path(name, workbench_dir).is_dir()
|
||||
|
||||
|
||||
def create_project(
|
||||
name: str, title: str, workbench_dir: Path = DEFAULT_WORKBENCH_DIR
|
||||
) -> Path:
|
||||
"""Create a project directory with log files. Idempotent — won't overwrite existing logs."""
|
||||
pdir = project_path(name, workbench_dir)
|
||||
pdir.mkdir(parents=True, exist_ok=True)
|
||||
(pdir / "assets").mkdir(exist_ok=True)
|
||||
|
||||
md_path = pdir / "session.md"
|
||||
if not md_path.exists():
|
||||
md_path.write_text(f"# {title} — Session Log\n\nStarted: {_now_iso()}\n\n")
|
||||
|
||||
jsonl_path = pdir / "session.jsonl"
|
||||
if not jsonl_path.exists():
|
||||
jsonl_path.write_text("")
|
||||
|
||||
cost_path = pdir / "cost-log.jsonl"
|
||||
if not cost_path.exists():
|
||||
cost_path.write_text("")
|
||||
|
||||
return pdir
|
||||
|
||||
|
||||
def append_log(
|
||||
name: str,
|
||||
entry: str,
|
||||
data: Optional[dict] = None,
|
||||
level: str = "info",
|
||||
workbench_dir: Path = DEFAULT_WORKBENCH_DIR,
|
||||
) -> dict:
|
||||
"""Append a log entry to session.md and session.jsonl. Returns the entry dict."""
|
||||
pdir = project_path(name, workbench_dir)
|
||||
ts = _now_iso()
|
||||
|
||||
with open(pdir / "session.md", "a") as f:
|
||||
f.write(f"\n### {ts}\n{entry}\n")
|
||||
|
||||
obj = {"ts": ts, "entry": entry, "level": level}
|
||||
if data:
|
||||
obj.update(data)
|
||||
with open(pdir / "session.jsonl", "a") as f:
|
||||
f.write(json.dumps(obj) + "\n")
|
||||
|
||||
return obj
|
||||
|
||||
|
||||
def read_log(
|
||||
name: str, tail: int = 20, workbench_dir: Path = DEFAULT_WORKBENCH_DIR
|
||||
) -> list[dict]:
|
||||
"""Read recent log entries from session.jsonl."""
|
||||
jsonl_path = project_path(name, workbench_dir) / "session.jsonl"
|
||||
if not jsonl_path.exists():
|
||||
return []
|
||||
|
||||
lines = jsonl_path.read_text().strip().split("\n")
|
||||
lines = [l for l in lines if l.strip()]
|
||||
recent = lines[-tail:] if len(lines) > tail else lines
|
||||
|
||||
entries = []
|
||||
for line in recent:
|
||||
try:
|
||||
entries.append(json.loads(line))
|
||||
except json.JSONDecodeError:
|
||||
entries.append({"raw": line})
|
||||
return entries
|
||||
|
||||
|
||||
def list_projects(workbench_dir: Path = DEFAULT_WORKBENCH_DIR) -> list[dict]:
|
||||
"""List all project directories."""
|
||||
if not workbench_dir.exists():
|
||||
return []
|
||||
projects = []
|
||||
for d in sorted(workbench_dir.iterdir()):
|
||||
if d.is_dir() and not d.name.startswith(".") and d.name != "START.md":
|
||||
projects.append({"name": d.name})
|
||||
return projects
|
||||
|
||||
|
||||
def log_session_event(
|
||||
name: str, event: str, workbench_dir: Path = DEFAULT_WORKBENCH_DIR, **extra
|
||||
) -> None:
|
||||
"""Append to cost-log.jsonl."""
|
||||
pdir = project_path(name, workbench_dir)
|
||||
obj = {"ts": _now_iso(), "event": event, "project": name}
|
||||
obj.update(extra)
|
||||
with open(pdir / "cost-log.jsonl", "a") as f:
|
||||
f.write(json.dumps(obj) + "\n")
|
||||
Reference in New Issue
Block a user