| from __future__ import annotations |
|
|
| from dataclasses import dataclass |
| from datetime import datetime |
| import json |
| from pathlib import Path |
| from typing import Any |
|
|
| from .paths import CLAUDE_EVENTS_DIR, CLAUDE_PROJECTS_DIR, CLAUDE_TASKS_DIR |
|
|
|
|
| @dataclass(frozen=True) |
| class ClaudeTask: |
| group_id: str |
| id: str |
| subject: str |
| description: str |
| status: str |
| blocks: tuple[str, ...] |
| blocked_by: tuple[str, ...] |
| path: Path |
| updated_ms: int |
|
|
|
|
| @dataclass(frozen=True) |
| class ClaudeTaskGroup: |
| id: str |
| label: str |
| tasks: tuple[ClaudeTask, ...] |
| updated_ms: int |
|
|
|
|
| @dataclass(frozen=True) |
| class ClaudeStats: |
| task_count: int |
| pending_count: int |
| active_count: int |
| completed_count: int |
| blocked_count: int |
| latest_task: ClaudeTask | None |
|
|
|
|
| @dataclass(frozen=True) |
| class ClaudeEvent: |
| ts_ms: int |
| event_type: str |
| tool_name: str |
| cwd: str |
| session_id: str |
| summary: str |
|
|
|
|
| @dataclass(frozen=True) |
| class ClaudeSession: |
| id: str |
| project: str |
| path: Path |
| updated_ms: int |
|
|
|
|
| def _shorten(text: str | None, limit: int = 160) -> str: |
| value = (text or "").strip().replace("\n", " ") |
| if len(value) <= limit: |
| return value |
| return value[: limit - 3].rstrip() + "..." |
|
|
|
|
| def _as_str_list(value: Any) -> tuple[str, ...]: |
| if not isinstance(value, list): |
| return () |
| return tuple(str(item) for item in value if str(item).strip()) |
|
|
|
|
| def _load_json(path: Path) -> dict[str, Any] | None: |
| try: |
| data = json.loads(path.read_text(encoding="utf-8")) |
| except (OSError, json.JSONDecodeError, UnicodeDecodeError): |
| return None |
| return data if isinstance(data, dict) else None |
|
|
|
|
| def _safe_mtime_ms(path: Path) -> int: |
| try: |
| return int(path.stat().st_mtime * 1000) |
| except OSError: |
| return 0 |
|
|
|
|
| def _task_from_file(path: Path, group_id: str) -> ClaudeTask | None: |
| data = _load_json(path) |
| if data is None: |
| return None |
| task_id = str(data.get("id") or path.stem) |
| subject = _shorten(str(data.get("subject") or "(untitled task)"), 180) |
| description = _shorten(str(data.get("description") or ""), 260) |
| status = str(data.get("status") or "unknown").strip().lower() or "unknown" |
| return ClaudeTask( |
| group_id=group_id, |
| id=task_id, |
| subject=subject, |
| description=description, |
| status=status, |
| blocks=_as_str_list(data.get("blocks")), |
| blocked_by=_as_str_list(data.get("blockedBy")), |
| path=path, |
| updated_ms=_safe_mtime_ms(path), |
| ) |
|
|
|
|
| def _project_label_from_dir(name: str) -> str: |
| label = name |
| for prefix in ("-home-smlflg-Projekte-", "-home-smlflg-"): |
| if label.startswith(prefix): |
| label = label[len(prefix) :] |
| break |
| label = label.replace("--", "/").replace("-", " / ") |
| return label or name |
|
|
|
|
| def _build_group_labels(projects_dir: Path) -> dict[str, str]: |
| if not projects_dir.is_dir(): |
| return {} |
| labels: dict[str, str] = {} |
| try: |
| files = projects_dir.glob("*/*.jsonl") |
| for path in files: |
| labels.setdefault(path.stem, _project_label_from_dir(path.parent.name)) |
| except OSError: |
| return {} |
| return labels |
|
|
|
|
| def list_claude_tasks( |
| *, |
| tasks_dir: Path = CLAUDE_TASKS_DIR, |
| projects_dir: Path = CLAUDE_PROJECTS_DIR, |
| query: str = "", |
| limit: int = 200, |
| ) -> list[ClaudeTask]: |
| if not tasks_dir.is_dir(): |
| return [] |
|
|
| needle = query.strip().lower() |
| tasks: list[ClaudeTask] = [] |
| try: |
| group_dirs = [path for path in tasks_dir.iterdir() if path.is_dir()] |
| except OSError: |
| return [] |
|
|
| for group_dir in group_dirs: |
| try: |
| files = sorted(group_dir.glob("*.json")) |
| except OSError: |
| continue |
| for path in files: |
| task = _task_from_file(path, group_dir.name) |
| if task is None: |
| continue |
| searchable = " ".join( |
| [ |
| task.group_id, |
| task.id, |
| task.subject, |
| task.description, |
| task.status, |
| ] |
| ).lower() |
| if needle and needle not in searchable: |
| continue |
| tasks.append(task) |
|
|
| tasks.sort(key=lambda item: (-item.updated_ms, item.group_id, item.id)) |
| return tasks[:limit] |
|
|
|
|
| def group_claude_tasks( |
| tasks: list[ClaudeTask], |
| *, |
| projects_dir: Path = CLAUDE_PROJECTS_DIR, |
| ) -> list[ClaudeTaskGroup]: |
| labels = _build_group_labels(projects_dir) |
| grouped: dict[str, list[ClaudeTask]] = {} |
| for task in tasks: |
| grouped.setdefault(task.group_id, []).append(task) |
|
|
| groups = [ |
| ClaudeTaskGroup( |
| id=group_id, |
| label=labels.get(group_id, group_id), |
| tasks=tuple(sorted(items, key=lambda item: (item.status, item.id))), |
| updated_ms=max((item.updated_ms for item in items), default=0), |
| ) |
| for group_id, items in grouped.items() |
| ] |
| groups.sort(key=lambda item: (-item.updated_ms, item.label)) |
| return groups |
|
|
|
|
| def get_claude_stats( |
| *, |
| tasks_dir: Path = CLAUDE_TASKS_DIR, |
| projects_dir: Path = CLAUDE_PROJECTS_DIR, |
| ) -> ClaudeStats: |
| tasks = list_claude_tasks(tasks_dir=tasks_dir, projects_dir=projects_dir, limit=10_000) |
| pending = 0 |
| active = 0 |
| completed = 0 |
| blocked = 0 |
| for task in tasks: |
| status = task.status |
| if status in {"completed", "done", "closed"}: |
| completed += 1 |
| elif status in {"active", "running", "in_progress", "in-progress"}: |
| active += 1 |
| else: |
| pending += 1 |
| if status == "blocked" or task.blocked_by: |
| blocked += 1 |
| return ClaudeStats( |
| task_count=len(tasks), |
| pending_count=pending, |
| active_count=active, |
| completed_count=completed, |
| blocked_count=blocked, |
| latest_task=tasks[0] if tasks else None, |
| ) |
|
|
|
|
| def list_claude_sessions( |
| *, |
| projects_dir: Path = CLAUDE_PROJECTS_DIR, |
| limit: int = 20, |
| ) -> list[ClaudeSession]: |
| if not projects_dir.is_dir(): |
| return [] |
| sessions: list[ClaudeSession] = [] |
| try: |
| files = projects_dir.glob("*/*.jsonl") |
| for path in files: |
| sessions.append( |
| ClaudeSession( |
| id=path.stem, |
| project=_project_label_from_dir(path.parent.name), |
| path=path, |
| updated_ms=_safe_mtime_ms(path), |
| ) |
| ) |
| except OSError: |
| return [] |
| sessions.sort(key=lambda item: -item.updated_ms) |
| return sessions[:limit] |
|
|
|
|
| def _event_summary(data: dict[str, Any]) -> str: |
| tool_input = data.get("tool_input") |
| if isinstance(tool_input, dict): |
| for key in ("description", "command", "file_path", "query"): |
| if tool_input.get(key): |
| return _shorten(str(tool_input[key]), 180) |
| tool_response = data.get("tool_response") |
| if isinstance(tool_response, dict): |
| for key in ("summary", "stdout", "stderr"): |
| if tool_response.get(key): |
| return _shorten(str(tool_response[key]), 180) |
| return "" |
|
|
|
|
| def list_recent_claude_events( |
| *, |
| events_dir: Path = CLAUDE_EVENTS_DIR, |
| limit: int = 20, |
| ) -> list[ClaudeEvent]: |
| if not events_dir.is_dir(): |
| return [] |
| try: |
| files = sorted(events_dir.glob("*.jsonl"), key=lambda path: path.name, reverse=True) |
| except OSError: |
| return [] |
|
|
| events: list[ClaudeEvent] = [] |
| for path in files: |
| if len(events) >= limit: |
| break |
| try: |
| lines = path.read_text(encoding="utf-8", errors="replace").splitlines() |
| except OSError: |
| continue |
| for line in reversed(lines): |
| if len(events) >= limit: |
| break |
| try: |
| data = json.loads(line) |
| except json.JSONDecodeError: |
| continue |
| if not isinstance(data, dict): |
| continue |
| logged_ts = data.get("logged_ts") |
| try: |
| ts_ms = int(float(logged_ts) * 1000) |
| except (TypeError, ValueError): |
| ts_ms = _safe_mtime_ms(path) |
| events.append( |
| ClaudeEvent( |
| ts_ms=ts_ms, |
| event_type=str(data.get("event_type") or data.get("hook_event_name") or ""), |
| tool_name=str(data.get("tool_name") or ""), |
| cwd=str(data.get("cwd") or ""), |
| session_id=str(data.get("session_id") or ""), |
| summary=_event_summary(data), |
| ) |
| ) |
|
|
| events.sort(key=lambda item: -item.ts_ms) |
| return events[:limit] |
|
|
|
|
| def format_claude_time(ms: int) -> str: |
| if not ms: |
| return "-" |
| return datetime.fromtimestamp(ms / 1000).strftime("%Y-%m-%d %H:%M") |
|
|
