move python clide to legacy/

Clide is being rebuilt as a Flutter desktop app. The Python Textual
implementation moves wholesale into legacy/ rather than being deleted:
its pane model, panel set, git skills, and panel communication design
are real thought that should remain readable next to the new code
while the rebuild finds its shape. Git's rename tracking preserves
history, so `git log -- legacy/` still works.

The Flutter rebuild lives at the repo root alongside a Go sidecar
(the architecture claudian was heading toward, which folds into
clide as a core component rather than a separate plugin project).
Bootstrap of the new shape lands in subsequent commits.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-20 20:30:51 +02:00
co-authored by Claude Opus 4.7
parent 30b8cf7db9
commit a355751437
163 changed files with 0 additions and 0 deletions
+20
View File
@@ -0,0 +1,20 @@
"""Business logic services for Clide."""
from clide.services.git_service import GitService
from clide.services.linter_service import LinterService
from clide.services.process_service import ProcessService
from clide.services.settings_service import SettingsService, UserSettings, get_settings_service
from clide.services.skill_installer import SkillInstaller, get_skill_installer
from clide.services.todo_scanner import TodoScanner
__all__ = [
"GitService",
"LinterService",
"ProcessService",
"SettingsService",
"SkillInstaller",
"TodoScanner",
"UserSettings",
"get_settings_service",
"get_skill_installer",
]
+238
View File
@@ -0,0 +1,238 @@
"""Claude Code event detection and parsing.
This module provides event infrastructure for detecting Claude Code actions
from terminal output, enabling tight IDE integration.
"""
from __future__ import annotations
import re
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from textual.message import Message
# Event Types
# -----------
@dataclass
class ClaudeEvent:
"""Base class for Claude Code events."""
pass
@dataclass
class FileReadEvent(ClaudeEvent):
"""Emitted when Claude reads a file."""
path: Path
@dataclass
class FileEditEvent(ClaudeEvent):
"""Emitted when Claude edits a file."""
path: Path
@dataclass
class FileWriteEvent(ClaudeEvent):
"""Emitted when Claude creates/writes a file."""
path: Path
@dataclass
class GlobEvent(ClaudeEvent):
"""Emitted when Claude searches for files."""
pattern: str
@dataclass
class GrepEvent(ClaudeEvent):
"""Emitted when Claude searches file contents."""
pattern: str
@dataclass
class ToolStartEvent(ClaudeEvent):
"""Emitted when Claude starts using a tool."""
tool_name: str
@dataclass
class ToolEndEvent(ClaudeEvent):
"""Emitted when Claude finishes using a tool."""
tool_name: str
@dataclass
class DiffProposedEvent(ClaudeEvent):
"""Emitted when Claude proposes a diff."""
content: str
# Textual Messages
# ----------------
class ClaudeEventMessage(Message):
"""Textual message wrapper for Claude events."""
def __init__(self, event: ClaudeEvent) -> None:
self.event = event
super().__init__()
# Pattern Matching
# ----------------
# Patterns for detecting Claude Code output
PATTERNS = {
# Tool invocations - Claude Code shows these with bullet points
"tool_read": re.compile(r"● Read\(([^)]+)\)"),
"tool_edit": re.compile(r"● Edit\(([^)]+)\)"),
"tool_write": re.compile(r"● Write\(([^)]+)\)"),
"tool_glob": re.compile(r"● Glob\(([^)]+)\)"),
"tool_grep": re.compile(r"● Grep\(([^)]+)\)"),
# Generic tool pattern
"tool_start": re.compile(r"● (\w+)\("),
"tool_end": re.compile(r"└─"),
# Diff headers
"diff_header": re.compile(r"^@@\s*-\d+(?:,\d+)?\s+\+\d+(?:,\d+)?\s*@@", re.MULTILINE),
"diff_file": re.compile(r"^(?:---|\+\+\+)\s+([^\s]+)", re.MULTILINE),
}
class ClaudeEventParser:
"""Parses Claude Code terminal output to detect events.
This parser is designed to work with raw terminal data fed
through the pyte event callback.
"""
def __init__(self, callback: Callable[[ClaudeEvent], None] | None = None) -> None:
"""Initialize the event parser.
Args:
callback: Optional callback invoked for each detected event.
"""
self._callback = callback
self._buffer = ""
self._current_tool: str | None = None
def set_callback(self, callback: Callable[[ClaudeEvent], None] | None) -> None:
"""Set the event callback."""
self._callback = callback
def feed(self, data: str) -> list[ClaudeEvent]:
"""Feed terminal data and return detected events.
Args:
data: Raw terminal data from Claude Code.
Returns:
List of detected events.
"""
events: list[ClaudeEvent] = []
# Add to buffer for multi-line matching
self._buffer += data
# Limit buffer size to prevent memory issues
if len(self._buffer) > 10000:
self._buffer = self._buffer[-5000:]
# Check for tool invocations
for match in PATTERNS["tool_read"].finditer(data):
path = Path(match.group(1).strip())
events.append(FileReadEvent(path=path))
for match in PATTERNS["tool_edit"].finditer(data):
path = Path(match.group(1).strip())
events.append(FileEditEvent(path=path))
for match in PATTERNS["tool_write"].finditer(data):
path = Path(match.group(1).strip())
events.append(FileWriteEvent(path=path))
for match in PATTERNS["tool_glob"].finditer(data):
pattern = match.group(1).strip()
events.append(GlobEvent(pattern=pattern))
for match in PATTERNS["tool_grep"].finditer(data):
pattern = match.group(1).strip()
events.append(GrepEvent(pattern=pattern))
# Check for generic tool start/end
for match in PATTERNS["tool_start"].finditer(data):
tool_name = match.group(1)
# Don't emit for tools we handle specifically
if tool_name not in ("Read", "Edit", "Write", "Glob", "Grep"):
events.append(ToolStartEvent(tool_name=tool_name))
self._current_tool = tool_name
if PATTERNS["tool_end"].search(data) and self._current_tool:
events.append(ToolEndEvent(tool_name=self._current_tool))
self._current_tool = None
# Check for diff content
if PATTERNS["diff_header"].search(self._buffer):
# Extract diff content (simplified - real impl would be more sophisticated)
events.append(DiffProposedEvent(content=self._buffer))
# Clear buffer after detecting diff
self._buffer = ""
# Invoke callback for each event
if self._callback:
for evt in events:
try:
self._callback(evt)
except Exception:
pass # Don't let callback errors propagate
return events
def reset(self) -> None:
"""Reset parser state."""
self._buffer = ""
self._current_tool = None
# Global parser instance for convenience
_event_parser: ClaudeEventParser | None = None
def get_event_parser() -> ClaudeEventParser:
"""Get the global event parser instance."""
global _event_parser
if _event_parser is None:
_event_parser = ClaudeEventParser()
return _event_parser
def setup_event_parsing(callback: Callable[[ClaudeEvent], None]) -> None:
"""Set up event parsing with the given callback.
This should be called during app initialization to wire up
the event parser with the terminal stream.
"""
from clide.vendor import pyte
parser = get_event_parser()
parser.set_callback(callback)
# Wire up to pyte's event callback
# The parser.feed returns events but pyte expects None return
def _feed_wrapper(data: str) -> None:
parser.feed(data)
pyte.set_event_callback(_feed_wrapper)
+48
View File
@@ -0,0 +1,48 @@
"""SQLite database engine and session management.
Used by both standalone Clide and clide-web. The database file
defaults to ~/.clide/clide.db but is configurable.
"""
from __future__ import annotations
from collections.abc import Generator
from pathlib import Path
from sqlmodel import Session as DBSession
from sqlmodel import SQLModel, create_engine
_engine = None
DEFAULT_DB_PATH = Path.home() / ".clide" / "clide.db"
def get_engine(db_path: Path | None = None):
"""Create or return the SQLAlchemy engine."""
global _engine
if _engine is None:
path = db_path or DEFAULT_DB_PATH
path.parent.mkdir(parents=True, exist_ok=True)
_engine = create_engine(
f"sqlite:///{path}",
echo=False,
connect_args={"check_same_thread": False},
)
return _engine
def init_db(db_path: Path | None = None) -> None:
"""Create all tables if they don't exist."""
# Import models so SQLModel registers them
import clide.models.db # noqa: F401
engine = get_engine(db_path)
SQLModel.metadata.create_all(engine)
def get_db() -> Generator[DBSession, None, None]:
"""Yield a database session. Usable as a FastAPI dependency or context manager."""
if _engine is None:
raise RuntimeError("Database not initialized — call init_db() first")
with DBSession(_engine) as session:
yield session
+204
View File
@@ -0,0 +1,204 @@
"""File operations service."""
from pathlib import Path
# Language extension mapping for syntax highlighting
# Maps file extensions to tree-sitter language identifiers
LANGUAGE_MAP: dict[str, str] = {
# Python
".py": "python",
".pyi": "python",
".pyw": "python",
# JavaScript/TypeScript
".js": "javascript",
".mjs": "javascript",
".cjs": "javascript",
".jsx": "javascript",
".ts": "typescript",
".tsx": "typescript",
".mts": "typescript",
".cts": "typescript",
# Web
".html": "html",
".htm": "html",
".css": "css",
".scss": "css",
".sass": "css",
".less": "css",
# Dart/Flutter
".dart": "dart",
# Data formats
".json": "json",
".jsonc": "json",
".yaml": "yaml",
".yml": "yaml",
".toml": "toml",
".xml": "xml",
# Markdown
".md": "markdown",
".markdown": "markdown",
# Shell
".sh": "bash",
".bash": "bash",
".zsh": "bash",
".fish": "bash",
# SQL
".sql": "sql",
# Other languages
".rs": "rust",
".go": "go",
".java": "java",
".c": "c",
".h": "c",
".cpp": "cpp",
".hpp": "cpp",
".cc": "cpp",
".cxx": "cpp",
".rb": "ruby",
".php": "php",
".vue": "vue",
".svelte": "svelte",
".lua": "lua",
".r": "r",
".R": "r",
".swift": "swift",
".kt": "kotlin",
".kts": "kotlin",
".scala": "scala",
".ex": "elixir",
".exs": "elixir",
}
class FileService:
"""Service for file I/O operations."""
def __init__(self, project_path: Path) -> None:
self.project_path = project_path
# Static methods for simple sync operations (used by EditorPane)
@staticmethod
def read_file(path: Path) -> str:
"""Read file contents synchronously.
Args:
path: Path to file
Returns:
File contents as string
"""
return path.read_text(encoding="utf-8")
@staticmethod
def write_file(path: Path, content: str) -> bool:
"""Write content to file synchronously.
Args:
path: Path to file
content: Content to write
Returns:
True if successful
"""
try:
path.write_text(content, encoding="utf-8")
return True
except OSError:
return False
@staticmethod
def detect_language(path: Path) -> str | None:
"""Detect language from file extension.
Args:
path: File path
Returns:
Language identifier for tree-sitter or None
"""
return LANGUAGE_MAP.get(path.suffix.lower())
# Instance methods for async operations
async def read_file_async(self, path: Path) -> str:
"""Read file contents asynchronously.
Args:
path: Path to file (relative or absolute)
Returns:
File contents as string
"""
full_path = self._resolve_path(path)
return full_path.read_text(encoding="utf-8")
async def write_file_async(self, path: Path, content: str) -> None:
"""Write content to file asynchronously.
Args:
path: Path to file
content: Content to write
"""
full_path = self._resolve_path(path)
full_path.write_text(content, encoding="utf-8")
async def file_exists(self, path: Path) -> bool:
"""Check if file exists.
Args:
path: Path to check
Returns:
True if file exists
"""
full_path = self._resolve_path(path)
return full_path.exists() and full_path.is_file()
async def get_language(self, path: Path) -> str | None:
"""Detect language from file extension.
Args:
path: File path
Returns:
Language identifier or None
"""
return LANGUAGE_MAP.get(path.suffix.lower())
def _resolve_path(self, path: Path) -> Path:
"""Resolve path relative to project root.
Args:
path: Path to resolve
Returns:
Absolute path
"""
if path.is_absolute():
return path
return self.project_path / path
def list_directory(self, path: Path | None = None) -> list[Path]:
"""List directory contents.
Args:
path: Directory path (defaults to project root)
Returns:
List of paths in directory
"""
dir_path = self._resolve_path(path) if path else self.project_path
if not dir_path.is_dir():
return []
entries = []
for entry in sorted(dir_path.iterdir()):
# Skip hidden files and common excludes
if entry.name.startswith("."):
continue
if entry.name in ("__pycache__", "node_modules", ".git"):
continue
entries.append(entry)
# Sort: directories first, then files
entries.sort(key=lambda p: (not p.is_dir(), p.name.lower()))
return entries
+287
View File
@@ -0,0 +1,287 @@
"""File system watching service for real-time sync.
This module provides file system monitoring capabilities for the Clide IDE,
enabling reactive updates when files change on disk.
"""
from __future__ import annotations
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Literal
if TYPE_CHECKING:
from collections.abc import Callable
from pydantic import BaseModel, ConfigDict
from textual.message import Message
try:
from watchdog.events import (
DirCreatedEvent,
DirDeletedEvent,
DirModifiedEvent,
DirMovedEvent,
FileCreatedEvent,
FileDeletedEvent,
FileModifiedEvent,
FileMovedEvent,
)
from watchdog.events import (
FileSystemEventHandler as WatchdogHandler,
)
from watchdog.observers import Observer as WatchdogObserver
WATCHDOG_AVAILABLE = True
except ImportError:
WATCHDOG_AVAILABLE = False
WatchdogObserver = None # type: ignore[misc, assignment]
WatchdogHandler = object # type: ignore[misc, assignment]
FileCreatedEvent = None # type: ignore[misc, assignment]
FileModifiedEvent = None # type: ignore[misc, assignment]
FileDeletedEvent = None # type: ignore[misc, assignment]
FileMovedEvent = None # type: ignore[misc, assignment]
DirCreatedEvent = None # type: ignore[misc, assignment]
DirModifiedEvent = None # type: ignore[misc, assignment]
DirDeletedEvent = None # type: ignore[misc, assignment]
DirMovedEvent = None # type: ignore[misc, assignment]
class FileEvent(BaseModel):
"""A file system event.
Attributes:
path: The path to the file/directory that changed.
event_type: The type of change that occurred.
timestamp: When the event occurred.
is_directory: Whether this is a directory event.
old_path: For move events, the original path.
"""
model_config = ConfigDict(strict=True, frozen=True)
path: Path
event_type: Literal["created", "modified", "deleted", "moved"]
timestamp: datetime
is_directory: bool = False
old_path: Path | None = None
class FileEventMessage(Message):
"""Textual message for file events."""
def __init__(self, event: FileEvent) -> None:
self.event = event
super().__init__()
class FileWatcher:
"""Watches a directory for file system changes.
Uses watchdog for efficient cross-platform file monitoring.
Emits FileEvent objects to registered handlers.
Example:
watcher = FileWatcher(Path.cwd())
watcher.register_handler(my_handler)
watcher.start()
# ... later ...
watcher.stop()
"""
def __init__(self, root: Path, ignore_patterns: list[str] | None = None) -> None:
"""Initialize the file watcher.
Args:
root: The root directory to watch.
ignore_patterns: Glob patterns to ignore (e.g., ["*.pyc", "__pycache__"]).
"""
self._root = root.resolve()
self._ignore_patterns = ignore_patterns or [
"*.pyc",
"__pycache__",
".git",
".venv",
"venv",
"node_modules",
".mypy_cache",
".ruff_cache",
".pytest_cache",
"*.egg-info",
".clide",
".claude",
]
self._handlers: list[Callable[[FileEvent], None]] = []
self._observer: WatchdogObserver | None = None # type: ignore[valid-type]
self._running = False
@property
def is_available(self) -> bool:
"""Check if watchdog is available."""
return WATCHDOG_AVAILABLE
@property
def is_running(self) -> bool:
"""Check if the watcher is currently running."""
return self._running
@property
def root(self) -> Path:
"""Get the root directory being watched."""
return self._root
def register_handler(self, handler: Callable[[FileEvent], None]) -> None:
"""Register a handler for file events.
Args:
handler: A callable that accepts a FileEvent.
"""
if handler not in self._handlers:
self._handlers.append(handler)
def unregister_handler(self, handler: Callable[[FileEvent], None]) -> None:
"""Unregister a handler.
Args:
handler: The handler to remove.
"""
if handler in self._handlers:
self._handlers.remove(handler)
def _should_ignore(self, path: Path) -> bool:
"""Check if a path should be ignored based on patterns."""
path_str = str(path)
for pattern in self._ignore_patterns:
# Simple pattern matching - could be enhanced with fnmatch
if pattern.startswith("*"):
if path_str.endswith(pattern[1:]):
return True
elif pattern in path_str:
return True
return False
def _emit_event(self, event: FileEvent) -> None:
"""Emit an event to all handlers."""
if self._should_ignore(event.path):
return
for handler in self._handlers:
try:
handler(event)
except Exception:
pass # Don't let handler errors affect other handlers
def start(self) -> bool:
"""Start watching for file changes.
Returns:
True if started successfully, False if watchdog is not available.
"""
if not WATCHDOG_AVAILABLE:
return False
if self._running:
return True
event_handler = _WatchdogHandler(self)
self._observer = WatchdogObserver()
self._observer.schedule(event_handler, str(self._root), recursive=True)
self._observer.start()
self._running = True
return True
def stop(self) -> None:
"""Stop watching for file changes."""
if self._observer is not None:
self._observer.stop()
self._observer.join(timeout=5)
self._observer = None
self._running = False
class _WatchdogHandler(WatchdogHandler): # type: ignore[misc, valid-type]
"""Internal handler for watchdog events."""
def __init__(self, watcher: FileWatcher) -> None:
super().__init__()
self._watcher = watcher
def _create_event(
self,
src_path: str | bytes,
event_type: Literal["created", "modified", "deleted", "moved"],
is_directory: bool,
dest_path: str | bytes | None = None,
) -> FileEvent:
"""Create a FileEvent from watchdog event data."""
# Watchdog can return bytes or str depending on platform
src = src_path.decode() if isinstance(src_path, bytes) else src_path
dest = dest_path.decode() if isinstance(dest_path, bytes) else dest_path
return FileEvent(
path=Path(dest if dest else src),
event_type=event_type,
timestamp=datetime.now(),
is_directory=is_directory,
old_path=Path(src) if dest else None,
)
def on_created(self, event) -> None: # type: ignore[no-untyped-def]
file_event = self._create_event(event.src_path, "created", event.is_directory)
self._watcher._emit_event(file_event)
def on_modified(self, event) -> None: # type: ignore[no-untyped-def]
file_event = self._create_event(event.src_path, "modified", event.is_directory)
self._watcher._emit_event(file_event)
def on_deleted(self, event) -> None: # type: ignore[no-untyped-def]
file_event = self._create_event(event.src_path, "deleted", event.is_directory)
self._watcher._emit_event(file_event)
def on_moved(self, event) -> None: # type: ignore[no-untyped-def]
file_event = self._create_event(
event.src_path, "moved", event.is_directory, event.dest_path
)
self._watcher._emit_event(file_event)
# Global watcher instance
_file_watcher: FileWatcher | None = None
def get_file_watcher(root: Path | None = None) -> FileWatcher:
"""Get or create the global file watcher.
Args:
root: The root directory to watch. Only used on first call.
Returns:
The FileWatcher instance.
"""
global _file_watcher
if _file_watcher is None:
_file_watcher = FileWatcher(root or Path.cwd())
return _file_watcher
def setup_file_watching(
root: Path,
handlers: list[Callable[[FileEvent], None]] | None = None,
) -> FileWatcher:
"""Set up file watching with optional initial handlers.
Args:
root: The root directory to watch.
handlers: Optional list of handlers to register.
Returns:
The configured FileWatcher.
"""
global _file_watcher
_file_watcher = FileWatcher(root)
if handlers:
for handler in handlers:
_file_watcher.register_handler(handler)
_file_watcher.start()
return _file_watcher
+273
View File
@@ -0,0 +1,273 @@
"""Git operations service."""
from pathlib import Path
from clide.models.git import (
ChangeStatus,
GitBranch,
GitChange,
GitCommit,
GitStatus,
)
from clide.services.process_service import ProcessService
class GitService:
"""Service for git operations."""
def __init__(self, repo_path: Path) -> None:
self.repo_path = repo_path
self._process = ProcessService(cwd=repo_path)
async def get_status(self) -> GitStatus:
"""Get current git status.
Returns:
GitStatus with staged/unstaged changes
"""
# Get porcelain status
result = await self._process.run("git", "status", "--porcelain", "-z")
staged: list[GitChange] = []
unstaged: list[GitChange] = []
untracked: list[str] = []
if result.success and result.stdout:
entries = result.stdout.split("\0")
for entry in entries:
if not entry or len(entry) < 3:
continue
index_status = entry[0]
worktree_status = entry[1]
path = entry[3:]
# Parse status
if index_status == "?":
# Untracked files go in both untracked list and unstaged
untracked.append(path)
unstaged.append(
GitChange(
path=path,
status=ChangeStatus.UNTRACKED,
staged=False,
)
)
else:
if index_status != " ":
staged.append(
GitChange(
path=path,
status=self._parse_status(index_status),
staged=True,
)
)
if worktree_status != " ":
unstaged.append(
GitChange(
path=path,
status=self._parse_status(worktree_status),
staged=False,
)
)
# Get current branch
branch_result = await self._process.run("git", "branch", "--show-current")
branch = branch_result.stdout.strip() if branch_result.success else "HEAD"
# Get ahead/behind
ahead, behind = await self._get_ahead_behind(branch)
return GitStatus(
branch=branch,
ahead=ahead,
behind=behind,
staged=tuple(staged),
unstaged=tuple(unstaged),
untracked=tuple(untracked),
)
async def _get_ahead_behind(self, branch: str) -> tuple[int, int]:
"""Get commits ahead/behind upstream."""
result = await self._process.run(
"git", "rev-list", "--left-right", "--count", f"{branch}...@{{upstream}}"
)
if result.success:
parts = result.stdout.strip().split()
if len(parts) == 2:
return int(parts[0]), int(parts[1])
return 0, 0
def _parse_status(self, char: str) -> ChangeStatus:
"""Parse git status character to ChangeStatus."""
mapping = {
"A": ChangeStatus.ADDED,
"M": ChangeStatus.MODIFIED,
"D": ChangeStatus.DELETED,
"R": ChangeStatus.RENAMED,
"C": ChangeStatus.COPIED,
"?": ChangeStatus.UNTRACKED,
"!": ChangeStatus.IGNORED,
"U": ChangeStatus.UNMERGED,
}
return mapping.get(char, ChangeStatus.MODIFIED)
async def stage_file(self, path: str) -> bool:
"""Stage a file.
Args:
path: File path to stage
Returns:
True if successful
"""
result = await self._process.run("git", "add", path)
return result.success
async def unstage_file(self, path: str) -> bool:
"""Unstage a file.
Args:
path: File path to unstage
Returns:
True if successful
"""
result = await self._process.run("git", "restore", "--staged", path)
return result.success
async def discard_changes(self, path: str) -> bool:
"""Discard changes to a file.
Args:
path: File path to discard
Returns:
True if successful
"""
result = await self._process.run("git", "restore", path)
return result.success
async def get_branches(self) -> list[GitBranch]:
"""Get list of branches.
Returns:
List of GitBranch objects
"""
result = await self._process.run(
"git",
"branch",
"-a",
"--format",
"%(HEAD)%(refname:short)|%(upstream:short)|%(objectname:short)|%(subject)",
)
branches: list[GitBranch] = []
if result.success:
for line in result.stdout.strip().split("\n"):
if not line:
continue
is_current = line.startswith("*")
parts = line[1:].split("|")
if len(parts) >= 4:
name = parts[0].strip()
branches.append(
GitBranch(
name=name,
is_current=is_current,
is_remote=name.startswith("remotes/"),
tracking=parts[1] or None,
commit_hash=parts[2],
commit_message=parts[3],
)
)
return branches
async def checkout_branch(self, branch: str) -> bool:
"""Checkout a branch.
Args:
branch: Branch name to checkout
Returns:
True if successful
"""
result = await self._process.run("git", "checkout", branch)
return result.success
async def create_branch(self, name: str, start_point: str | None = None) -> bool:
"""Create a new branch.
Args:
name: New branch name
start_point: Optional starting commit/branch
Returns:
True if successful
"""
args = ["git", "checkout", "-b", name]
if start_point:
args.append(start_point)
result = await self._process.run(*args)
return result.success
async def get_diff(self, path: str, staged: bool = False) -> str:
"""Get diff for a file.
Args:
path: File path
staged: Whether to get staged diff
Returns:
Diff output string
"""
args = ["git", "diff"]
if staged:
args.append("--cached")
args.append("--")
args.append(path)
result = await self._process.run(*args)
return result.stdout if result.success else ""
async def get_log(self, max_count: int = 50) -> list[GitCommit]:
"""Get commit log.
Args:
max_count: Maximum number of commits to return
Returns:
List of GitCommit objects
"""
result = await self._process.run(
"git",
"log",
f"--max-count={max_count}",
"--format=%H|%h|%s|%an|%ar|%P|%D",
"--all",
)
commits: list[GitCommit] = []
if result.success:
for line in result.stdout.strip().split("\n"):
if not line:
continue
parts = line.split("|")
if len(parts) >= 7:
parents = tuple(parts[5].split()) if parts[5] else ()
refs = tuple(r.strip() for r in parts[6].split(",")) if parts[6] else ()
commits.append(
GitCommit(
hash=parts[0],
short_hash=parts[1],
message=parts[2],
author=parts[3],
date=parts[4],
is_merge=len(parents) > 1,
parents=parents,
refs=refs,
)
)
return commits
+127
View File
@@ -0,0 +1,127 @@
"""Linter integration service."""
import json
from pathlib import Path
from clide.models.problems import Problem, ProblemsSummary, Severity
from clide.services.process_service import ProcessService
class LinterService:
"""Service for running linters and parsing output."""
def __init__(self, project_path: Path) -> None:
self.project_path = project_path
self._process = ProcessService(cwd=project_path)
async def run_ruff(self) -> list[Problem]:
"""Run ruff linter.
Returns:
List of problems found
"""
result = await self._process.run(
"ruff", "check", "--output-format=json", "."
)
problems: list[Problem] = []
if result.stdout:
try:
data = json.loads(result.stdout)
for item in data:
severity = self._ruff_severity(item.get("code", ""))
problems.append(Problem(
file_path=Path(item["filename"]),
line=item["location"]["row"],
column=item["location"]["column"],
end_line=item.get("end_location", {}).get("row"),
end_column=item.get("end_location", {}).get("column"),
severity=severity,
message=item["message"],
source="ruff",
code=item.get("code"),
))
except json.JSONDecodeError:
pass
return problems
def _ruff_severity(self, code: str) -> Severity:
"""Map ruff code to severity."""
if code.startswith("E") or code.startswith("F"):
return Severity.ERROR
if code.startswith("W"):
return Severity.WARNING
return Severity.INFO
async def run_mypy(self) -> list[Problem]:
"""Run mypy type checker.
Returns:
List of problems found
"""
result = await self._process.run(
"mypy", "--output=json", "."
)
problems: list[Problem] = []
for line in result.stdout.strip().split("\n"):
if not line:
continue
try:
data = json.loads(line)
severity = self._mypy_severity(data.get("severity", "error"))
problems.append(Problem(
file_path=Path(data["file"]),
line=data["line"],
column=data.get("column", 1),
severity=severity,
message=data["message"],
source="mypy",
code=data.get("code"),
))
except (json.JSONDecodeError, KeyError):
continue
return problems
def _mypy_severity(self, severity: str) -> Severity:
"""Map mypy severity to Severity enum."""
mapping = {
"error": Severity.ERROR,
"warning": Severity.WARNING,
"note": Severity.INFO,
}
return mapping.get(severity, Severity.ERROR)
async def run_all(self, linters: list[str]) -> tuple[list[Problem], ProblemsSummary]:
"""Run all configured linters.
Args:
linters: List of linter names to run
Returns:
Tuple of (problems list, summary)
"""
all_problems: list[Problem] = []
for linter in linters:
if linter == "ruff":
all_problems.extend(await self.run_ruff())
elif linter == "mypy":
all_problems.extend(await self.run_mypy())
# Create summary
errors = sum(1 for p in all_problems if p.severity == Severity.ERROR)
warnings = sum(1 for p in all_problems if p.severity == Severity.WARNING)
infos = sum(1 for p in all_problems if p.severity == Severity.INFO)
hints = sum(1 for p in all_problems if p.severity == Severity.HINT)
summary = ProblemsSummary(
errors=errors,
warnings=warnings,
infos=infos,
hints=hints,
)
return all_problems, summary
+124
View File
@@ -0,0 +1,124 @@
"""Generic subprocess management service."""
import asyncio
import subprocess
from dataclasses import dataclass
from pathlib import Path
@dataclass
class CommandResult:
"""Result of a command execution."""
returncode: int
stdout: str
stderr: str
@property
def success(self) -> bool:
"""Check if command succeeded."""
return self.returncode == 0
class ProcessService:
"""Service for running subprocess commands."""
def __init__(self, cwd: Path | None = None) -> None:
self.cwd = cwd or Path.cwd()
async def run(
self,
*args: str,
cwd: Path | None = None,
timeout: float | None = 30.0,
env: dict[str, str] | None = None,
) -> CommandResult:
"""Run a command asynchronously.
Args:
*args: Command and arguments
cwd: Working directory (defaults to service cwd)
timeout: Timeout in seconds
env: Environment variables to add
Returns:
CommandResult with stdout, stderr, and returncode
"""
working_dir = cwd or self.cwd
try:
process = await asyncio.create_subprocess_exec(
*args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=working_dir,
env=env,
)
stdout, stderr = await asyncio.wait_for(
process.communicate(),
timeout=timeout,
)
return CommandResult(
returncode=process.returncode or 0,
stdout=stdout.decode("utf-8", errors="replace"),
stderr=stderr.decode("utf-8", errors="replace"),
)
except TimeoutError:
process.kill()
return CommandResult(
returncode=-1,
stdout="",
stderr="Command timed out",
)
except Exception as e:
return CommandResult(
returncode=-1,
stdout="",
stderr=str(e),
)
def run_sync(
self,
*args: str,
cwd: Path | None = None,
timeout: float | None = 30.0,
) -> CommandResult:
"""Run a command synchronously (for use in threads).
Args:
*args: Command and arguments
cwd: Working directory
timeout: Timeout in seconds
Returns:
CommandResult with stdout, stderr, and returncode
"""
working_dir = cwd or self.cwd
try:
result = subprocess.run(
args,
capture_output=True,
cwd=working_dir,
timeout=timeout,
text=True,
)
return CommandResult(
returncode=result.returncode,
stdout=result.stdout,
stderr=result.stderr,
)
except subprocess.TimeoutExpired:
return CommandResult(
returncode=-1,
stdout="",
stderr="Command timed out",
)
except Exception as e:
return CommandResult(
returncode=-1,
stdout="",
stderr=str(e),
)
+162
View File
@@ -0,0 +1,162 @@
"""Settings persistence service for user preferences."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from pydantic import BaseModel
class UserSettings(BaseModel):
"""User settings that persist across sessions.
Stored in ~/.clide/settings.json
"""
# Appearance
theme: str = "summer-night"
# Panel state
sidebar_visible: bool = True
context_visible: bool = True
# Window
compact_mode: bool = False
# Behavior
auto_save: bool = True
confirm_exit: bool = True
# Integrations
jira_enabled: bool = False
# Debug
terminal_debug: bool = False # Verbose terminal/pyte logging to ~/.clide/terminal_debug.log
class SettingsService:
"""Service for loading and saving user settings.
Settings are stored in ~/.clide/settings.json
"""
def __init__(self, settings_dir: Path | None = None) -> None:
"""Initialize the settings service.
Args:
settings_dir: Override the settings directory (default: ~/.clide)
"""
self._settings_dir = settings_dir or Path.home() / ".clide"
self._settings_file = self._settings_dir / "settings.json"
self._settings: UserSettings | None = None
@property
def settings_dir(self) -> Path:
"""Get the settings directory path."""
return self._settings_dir
@property
def settings_file(self) -> Path:
"""Get the settings file path."""
return self._settings_file
def load(self) -> UserSettings:
"""Load settings from disk, creating defaults if needed.
Returns:
The loaded or default UserSettings
"""
if self._settings is not None:
return self._settings
if self._settings_file.exists():
try:
data = json.loads(self._settings_file.read_text())
self._settings = UserSettings.model_validate(data)
except (json.JSONDecodeError, ValueError):
# Invalid JSON or schema, use defaults
self._settings = UserSettings()
else:
self._settings = UserSettings()
return self._settings
def save(self) -> None:
"""Save current settings to disk."""
if self._settings is None:
return
# Ensure directory exists
self._settings_dir.mkdir(parents=True, exist_ok=True)
# Write settings as formatted JSON
data = self._settings.model_dump(mode="json")
self._settings_file.write_text(
json.dumps(data, indent=2, sort_keys=True) + "\n"
)
def get(self, key: str, default: Any = None) -> Any:
"""Get a setting value.
Args:
key: The setting key (attribute name)
default: Default value if key doesn't exist
Returns:
The setting value or default
"""
settings = self.load()
return getattr(settings, key, default)
def set(self, key: str, value: Any, *, save: bool = True) -> None:
"""Set a setting value.
Args:
key: The setting key (attribute name)
value: The value to set
save: Whether to save immediately (default: True)
"""
settings = self.load()
if hasattr(settings, key):
# Create new settings with updated value
data = settings.model_dump()
data[key] = value
self._settings = UserSettings.model_validate(data)
if save:
self.save()
def update(self, **kwargs: Any) -> None:
"""Update multiple settings at once.
Args:
**kwargs: Key-value pairs to update
"""
settings = self.load()
data = settings.model_dump()
for key, value in kwargs.items():
if hasattr(settings, key):
data[key] = value
self._settings = UserSettings.model_validate(data)
self.save()
def reset(self) -> None:
"""Reset settings to defaults."""
self._settings = UserSettings()
self.save()
# Global instance for convenience
_settings_service: SettingsService | None = None
def get_settings_service() -> SettingsService:
"""Get the global settings service instance."""
global _settings_service
if _settings_service is None:
_settings_service = SettingsService()
return _settings_service
+285
View File
@@ -0,0 +1,285 @@
"""Skill installer service for Claude Code skills.
This module provides functionality to install skill templates
into the user's Claude Code configuration.
"""
from __future__ import annotations
import shutil
from pathlib import Path
from typing import Literal
# Path to bundled skill templates within Clide package
TEMPLATES_DIR = Path(__file__).parent.parent / "templates" / "skills"
# Default installation locations
USER_SKILLS_DIR = Path.home() / ".claude" / "skills"
class SkillInstaller:
"""Installs Claude Code skills from templates.
Skills can be installed to:
- User level: ~/.claude/skills/ (available globally)
- Project level: .claude/skills/ (available in project only)
Example:
installer = SkillInstaller()
# Check if skill exists
if not installer.is_installed("git-workflow"):
installer.install("git-workflow")
# Install to project instead of user
installer.install("git-workflow", scope="project")
"""
def __init__(
self,
templates_dir: Path | None = None,
project_dir: Path | None = None,
) -> None:
"""Initialize the skill installer.
Args:
templates_dir: Override the templates directory.
project_dir: Project directory for project-scoped skills.
"""
self._templates_dir = templates_dir or TEMPLATES_DIR
self._project_dir = project_dir or Path.cwd()
@property
def templates_dir(self) -> Path:
"""Get the templates directory."""
return self._templates_dir
@property
def user_skills_dir(self) -> Path:
"""Get the user skills directory."""
return USER_SKILLS_DIR
@property
def project_skills_dir(self) -> Path:
"""Get the project skills directory."""
return self._project_dir / ".claude" / "skills"
def list_available_templates(self) -> list[str]:
"""List all available skill templates.
Returns:
List of skill names that can be installed.
"""
if not self._templates_dir.exists():
return []
return [
d.name
for d in self._templates_dir.iterdir()
if d.is_dir() and (d / "SKILL.md").exists()
]
def list_installed_skills(
self,
scope: Literal["user", "project", "all"] = "all",
) -> list[dict[str, str]]:
"""List installed skills.
Args:
scope: Which skills to list - user, project, or all.
Returns:
List of dicts with 'name', 'scope', and 'path' keys.
"""
skills = []
if scope in ("user", "all"):
if self.user_skills_dir.exists():
for d in self.user_skills_dir.iterdir():
if d.is_dir() and (d / "SKILL.md").exists():
skills.append(
{
"name": d.name,
"scope": "user",
"path": str(d),
}
)
if scope in ("project", "all"):
if self.project_skills_dir.exists():
for d in self.project_skills_dir.iterdir():
if d.is_dir() and (d / "SKILL.md").exists():
skills.append(
{
"name": d.name,
"scope": "project",
"path": str(d),
}
)
return skills
def is_installed(
self,
skill_name: str,
scope: Literal["user", "project", "any"] = "any",
) -> bool:
"""Check if a skill is installed.
Args:
skill_name: The skill name to check.
scope: Where to check - user, project, or any.
Returns:
True if the skill is installed.
"""
if scope in ("user", "any"):
user_skill = self.user_skills_dir / skill_name / "SKILL.md"
if user_skill.exists():
return True
if scope in ("project", "any"):
project_skill = self.project_skills_dir / skill_name / "SKILL.md"
if project_skill.exists():
return True
return False
def get_skill_path(
self,
skill_name: str,
scope: Literal["user", "project", "any"] = "any",
) -> Path | None:
"""Get the path to an installed skill.
Args:
skill_name: The skill name.
scope: Where to look - user, project, or any (project takes priority).
Returns:
Path to the skill directory, or None if not found.
"""
# Project scope takes priority when scope is "any"
if scope in ("project", "any"):
project_skill = self.project_skills_dir / skill_name
if (project_skill / "SKILL.md").exists():
return project_skill
if scope in ("user", "any"):
user_skill = self.user_skills_dir / skill_name
if (user_skill / "SKILL.md").exists():
return user_skill
return None
def install(
self,
skill_name: str,
scope: Literal["user", "project"] = "user",
overwrite: bool = False,
) -> Path:
"""Install a skill from templates.
Args:
skill_name: The skill name to install.
scope: Where to install - user or project level.
overwrite: Whether to overwrite existing installation.
Returns:
Path to the installed skill.
Raises:
ValueError: If skill template doesn't exist.
FileExistsError: If skill exists and overwrite is False.
"""
# Check template exists
template_dir = self._templates_dir / skill_name
if not template_dir.exists() or not (template_dir / "SKILL.md").exists():
raise ValueError(f"Skill template '{skill_name}' not found")
# Determine target directory
if scope == "user":
target_dir = self.user_skills_dir / skill_name
else:
target_dir = self.project_skills_dir / skill_name
# Check if already exists
if target_dir.exists():
if not overwrite:
raise FileExistsError(f"Skill '{skill_name}' already installed at {target_dir}")
shutil.rmtree(target_dir)
# Create parent directory
target_dir.parent.mkdir(parents=True, exist_ok=True)
# Copy template
shutil.copytree(template_dir, target_dir)
return target_dir
def uninstall(
self,
skill_name: str,
scope: Literal["user", "project"] = "user",
) -> bool:
"""Uninstall a skill.
Args:
skill_name: The skill name to uninstall.
scope: Where to uninstall from - user or project level.
Returns:
True if skill was uninstalled, False if it wasn't installed.
"""
if scope == "user":
skill_dir = self.user_skills_dir / skill_name
else:
skill_dir = self.project_skills_dir / skill_name
if skill_dir.exists():
shutil.rmtree(skill_dir)
return True
return False
def ensure_installed(
self,
skill_name: str,
scope: Literal["user", "project"] = "project",
) -> Path:
"""Ensure a skill is installed, installing if needed.
Args:
skill_name: The skill name.
scope: Where to install if not present (default: project).
Returns:
Path to the skill directory.
Raises:
ValueError: If skill template doesn't exist.
"""
existing = self.get_skill_path(skill_name)
if existing:
return existing
return self.install(skill_name, scope=scope)
# Global instance
_skill_installer: SkillInstaller | None = None
def get_skill_installer(project_dir: Path | None = None) -> SkillInstaller:
"""Get or create the global skill installer.
Args:
project_dir: Project directory (only used on first call).
Returns:
The SkillInstaller instance.
"""
global _skill_installer
if _skill_installer is None:
_skill_installer = SkillInstaller(project_dir=project_dir)
return _skill_installer
+166
View File
@@ -0,0 +1,166 @@
"""Syntax highlighting service for additional language support.
Textual 7.x includes built-in support for many languages when tree-sitter
packages are installed. This module provides utilities for checking and
registering additional languages.
"""
from __future__ import annotations
import logging
logger = logging.getLogger(__name__)
# Languages supported by Textual's TextArea with tree-sitter packages
SUPPORTED_LANGUAGES = {
# Core web languages
"python",
"javascript",
"typescript",
"html",
"css",
"json",
# Markup/config
"markdown",
"yaml",
"toml",
"xml",
# Shell
"bash",
# SQL
"sql",
# Systems languages
"rust",
"go",
"java",
# Regex
"regex",
}
# Additional languages that may be registered if packages are available
OPTIONAL_LANGUAGES = [
"dart",
"kotlin",
"swift",
"scala",
"ruby",
"php",
"lua",
"c",
"cpp",
"csharp",
"elixir",
"haskell",
"ocaml",
"zig",
"nim",
"vue",
"svelte",
]
def register_languages() -> list[str]:
"""Register additional languages with Textual's TextArea.
In Textual 7.x, languages are automatically registered when tree-sitter
packages are installed. This function registers additional languages
that need special handling (like TypeScript which has separate functions).
Returns:
List of successfully registered language names
"""
try:
from textual.widgets import TextArea
except ImportError:
logger.warning("Textual not available")
return []
registered = []
# Register TypeScript and TSX (they have special language function names)
try:
import tree_sitter_typescript as tst
# Register TypeScript
try:
TextArea.register_language(tst.language_typescript(), "typescript")
registered.append("typescript")
logger.debug("Registered language: typescript")
except Exception as e:
logger.debug(f"Could not register typescript: {e}")
# Register TSX
try:
TextArea.register_language(tst.language_tsx(), "tsx")
registered.append("tsx")
logger.debug("Registered language: tsx")
except Exception as e:
logger.debug(f"Could not register tsx: {e}")
except ImportError:
logger.debug("tree-sitter-typescript not installed")
# Register other optional languages with standard API
for lang_name in OPTIONAL_LANGUAGES:
try:
# Try to import the tree-sitter package for this language
module_name = f"tree_sitter_{lang_name}"
module = __import__(module_name)
# Get the language function
if hasattr(module, "language"):
language = module.language()
# Try to get a highlight query if available
highlight_query = None
if hasattr(module, "HIGHLIGHTS_QUERY"):
highlight_query = module.HIGHLIGHTS_QUERY
# Register with Textual
try:
TextArea.register_language(language, lang_name, highlight_query)
registered.append(lang_name)
logger.debug(f"Registered language: {lang_name}")
except Exception as e:
logger.debug(f"Could not register language '{lang_name}': {e}")
except ImportError:
# Package not installed, skip
pass
except Exception as e:
logger.debug(f"Error processing language '{lang_name}': {e}")
return registered
def get_available_languages() -> list[str]:
"""Get list of all available languages for syntax highlighting.
Returns:
List of language names that can be used with TextArea
"""
try:
from textual.widgets import TextArea
# Create a temporary instance to check available languages
ta = TextArea()
return sorted(ta.available_languages)
except ImportError:
return sorted(SUPPORTED_LANGUAGES)
except Exception:
return sorted(SUPPORTED_LANGUAGES)
def is_syntax_highlighting_available() -> bool:
"""Check if syntax highlighting is available.
Returns:
True if tree-sitter is installed and syntax highlighting works
"""
try:
from textual.widgets import TextArea
ta = TextArea("test", language="python")
return ta.is_syntax_aware
except Exception:
return False
+244
View File
@@ -0,0 +1,244 @@
"""TODO comment scanner service."""
import re
from pathlib import Path
from clide.models.todos import ProjectTodoItem, TodoItem, TodosSummary, TodoType
from clide.services.process_service import ProcessService
class TodoScanner:
"""Service for scanning TODO/FIXME comments in code."""
# Pattern to match TODO-style comments
TODO_PATTERN = re.compile(
r"(?:#|//|/\*|\*|<!--)\s*(TODO|FIXME|HACK|XXX|NOTE|BUG|OPTIMIZE|REVIEW)\s*:?\s*(.+?)(?:\*/|-->)?$",
re.IGNORECASE,
)
# Pattern to match markdown checkboxes: - [ ] or - [x]
CHECKBOX_PATTERN = re.compile(r"^(\s*)-\s*\[([ xX])\]\s*(.+)$")
# File extensions to scan
SCAN_EXTENSIONS = {
".py",
".js",
".ts",
".jsx",
".tsx",
".java",
".c",
".cpp",
".h",
".go",
".rs",
".rb",
".php",
".css",
".scss",
".html",
".vue",
".svelte",
".md",
".sh",
".bash",
".yaml",
".yml",
".toml",
}
def __init__(self, project_path: Path) -> None:
self.project_path = project_path
self._process = ProcessService(cwd=project_path)
async def scan(
self,
) -> tuple[list[TodoItem], list[ProjectTodoItem], TodosSummary]:
"""Scan project for TODO comments and TODO.md items.
Returns:
Tuple of (code todo items, project todo items, summary)
"""
items: list[TodoItem] = []
# Use ripgrep if available for speed
result = await self._process.run(
"rg",
"--line-number",
"--no-heading",
"-e",
r"\b(TODO|FIXME|HACK|XXX|NOTE|BUG|OPTIMIZE|REVIEW)\b",
"--type-add",
"code:*.py",
"--type-add",
"code:*.js",
"--type-add",
"code:*.ts",
"--type",
"code",
".",
)
if result.success:
items = self._parse_ripgrep_output(result.stdout)
else:
# Fallback to Python-based scanning
items = await self._scan_with_python()
# Parse TODO.md if it exists
project_items = self._parse_todo_md()
# Create summary
todo_count = sum(1 for i in items if i.todo_type == TodoType.TODO)
fixme_count = sum(1 for i in items if i.todo_type == TodoType.FIXME)
hack_count = sum(1 for i in items if i.todo_type == TodoType.HACK)
other_count = len(items) - todo_count - fixme_count - hack_count
project_todo_count = sum(1 for i in project_items if not i.checked)
project_done_count = sum(1 for i in project_items if i.checked)
summary = TodosSummary(
todo_count=todo_count,
fixme_count=fixme_count,
hack_count=hack_count,
other_count=other_count,
project_todo_count=project_todo_count,
project_done_count=project_done_count,
)
return items, project_items, summary
def _parse_todo_md(self) -> list[ProjectTodoItem]:
"""Parse TODO.md file for checkbox items.
Returns:
List of project TODO items
"""
todo_md_path = self.project_path / "TODO.md"
if not todo_md_path.exists():
return []
items: list[ProjectTodoItem] = []
current_section = "General"
current_subsection: str | None = None
try:
content = todo_md_path.read_text(encoding="utf-8")
for line_num, line in enumerate(content.split("\n"), 1):
# Check for section headers (## Section)
if line.startswith("## "):
current_section = line[3:].strip()
current_subsection = None
continue
# Check for subsection headers (### Subsection)
if line.startswith("### "):
current_subsection = line[4:].strip()
continue
# Check for checkbox items
match = self.CHECKBOX_PATTERN.match(line)
if match:
checkbox_state = match.group(2)
text = match.group(3).strip()
checked = checkbox_state.lower() == "x"
items.append(
ProjectTodoItem(
text=text,
section=current_section,
subsection=current_subsection,
line=line_num,
checked=checked,
)
)
except (OSError, UnicodeDecodeError):
pass
return items
def _parse_ripgrep_output(self, output: str) -> list[TodoItem]:
"""Parse ripgrep output into TodoItems."""
items: list[TodoItem] = []
for line in output.strip().split("\n"):
if not line:
continue
# Format: path:line:content
parts = line.split(":", 2)
if len(parts) < 3:
continue
file_path = Path(parts[0])
try:
line_num = int(parts[1])
except ValueError:
continue
content = parts[2]
# Parse the TODO type and text
match = self.TODO_PATTERN.search(content)
if match:
todo_type_str = match.group(1).upper()
todo_text = match.group(2).strip()
try:
todo_type = TodoType(todo_type_str)
except ValueError:
todo_type = TodoType.TODO
items.append(
TodoItem(
file_path=file_path,
line=line_num,
column=content.find(todo_type_str) + 1,
todo_type=todo_type,
text=todo_text,
context_line=content.strip(),
)
)
return items
async def _scan_with_python(self) -> list[TodoItem]:
"""Fallback Python-based scanning."""
items: list[TodoItem] = []
for ext in self.SCAN_EXTENSIONS:
for file_path in self.project_path.rglob(f"*{ext}"):
# Skip hidden directories and common excludes
if any(part.startswith(".") for part in file_path.parts):
continue
if "node_modules" in file_path.parts:
continue
if "__pycache__" in file_path.parts:
continue
try:
content = file_path.read_text(encoding="utf-8", errors="ignore")
for line_num, line in enumerate(content.split("\n"), 1):
match = self.TODO_PATTERN.search(line)
if match:
todo_type_str = match.group(1).upper()
todo_text = match.group(2).strip()
try:
todo_type = TodoType(todo_type_str)
except ValueError:
todo_type = TodoType.TODO
items.append(
TodoItem(
file_path=file_path.relative_to(self.project_path),
line=line_num,
column=line.find(todo_type_str) + 1,
todo_type=todo_type,
text=todo_text,
context_line=line.strip(),
)
)
except (OSError, UnicodeDecodeError):
continue
return items
+359
View File
@@ -0,0 +1,359 @@
"""Auto-update service for Clide.
Checks for updates from the release server and handles self-updating.
User settings in ~/.clide/ are preserved across updates.
"""
from __future__ import annotations
import json
import logging
import os
import platform
import shutil
import stat
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from urllib.error import URLError
from urllib.request import Request, urlopen
logger = logging.getLogger(__name__)
# Update server configuration
UPDATE_SERVER = "https://git.schweitz.net"
REPO_OWNER = "jeroen" # TODO: Update with actual owner
REPO_NAME = "clide" # TODO: Update with actual repo name
# Current version (injected at build time or read from package)
try:
from clide import __version__ as CURRENT_VERSION
except ImportError:
CURRENT_VERSION = "0.0.0"
@dataclass
class ReleaseInfo:
"""Information about a release."""
version: str
tag_name: str
download_url: str
release_notes: str
published_at: str
@dataclass
class UpdateCheckResult:
"""Result of checking for updates."""
update_available: bool
current_version: str
latest_version: str | None
release_info: ReleaseInfo | None
error: str | None = None
def get_platform_asset_name() -> str:
"""Get the expected asset name for the current platform."""
system = platform.system().lower()
machine = platform.machine().lower()
if system == "darwin":
return "macos.dmg"
elif system == "linux":
# Normalize architecture names
if machine in ("x86_64", "amd64"):
arch = "x86_64"
elif machine in ("aarch64", "arm64"):
arch = "aarch64"
else:
arch = machine
return f"linux-{arch}.AppImage"
elif system == "windows":
return "windows-setup.exe"
else:
raise RuntimeError(f"Unsupported platform: {system}")
def parse_version(version: str) -> tuple[int, ...]:
"""Parse a version string into a tuple for comparison."""
# Remove 'v' prefix if present
version = version.lstrip("v")
# Split and convert to integers
parts = []
for part in version.split("."):
# Handle versions like "1.0.0-beta"
num_part = ""
for char in part:
if char.isdigit():
num_part += char
else:
break
parts.append(int(num_part) if num_part else 0)
return tuple(parts)
def is_newer_version(current: str, latest: str) -> bool:
"""Check if latest version is newer than current."""
return parse_version(latest) > parse_version(current)
def check_for_updates() -> UpdateCheckResult:
"""Check the release server for available updates.
Returns:
UpdateCheckResult with update status and release info.
"""
try:
# Gitea API endpoint for releases
api_url = f"{UPDATE_SERVER}/api/v1/repos/{REPO_OWNER}/{REPO_NAME}/releases/latest"
request = Request(api_url)
request.add_header("Accept", "application/json")
request.add_header("User-Agent", f"Clide/{CURRENT_VERSION}")
with urlopen(request, timeout=10) as response:
data = json.loads(response.read().decode("utf-8"))
tag_name = data.get("tag_name", "")
latest_version = tag_name.lstrip("v")
# Find the download URL for current platform
platform_asset = get_platform_asset_name()
download_url = None
for asset in data.get("assets", []):
if platform_asset in asset.get("name", ""):
download_url = asset.get("browser_download_url")
break
if not download_url:
# Try constructing URL from release
download_url = f"{UPDATE_SERVER}/{REPO_OWNER}/{REPO_NAME}/releases/download/{tag_name}/Clide-{tag_name}-{platform_asset}"
release_info = ReleaseInfo(
version=latest_version,
tag_name=tag_name,
download_url=download_url,
release_notes=data.get("body", ""),
published_at=data.get("published_at", ""),
)
update_available = is_newer_version(CURRENT_VERSION, latest_version)
return UpdateCheckResult(
update_available=update_available,
current_version=CURRENT_VERSION,
latest_version=latest_version,
release_info=release_info,
)
except URLError as e:
logger.error(f"Failed to check for updates: {e}")
return UpdateCheckResult(
update_available=False,
current_version=CURRENT_VERSION,
latest_version=None,
release_info=None,
error=f"Network error: {e.reason}",
)
except json.JSONDecodeError as e:
logger.error(f"Failed to parse update response: {e}")
return UpdateCheckResult(
update_available=False,
current_version=CURRENT_VERSION,
latest_version=None,
release_info=None,
error="Invalid response from update server",
)
except Exception as e:
logger.error(f"Unexpected error checking for updates: {e}")
return UpdateCheckResult(
update_available=False,
current_version=CURRENT_VERSION,
latest_version=None,
release_info=None,
error=str(e),
)
def download_update(release_info: ReleaseInfo, progress_callback=None) -> Path:
"""Download the update to a temporary location.
Args:
release_info: Release information with download URL.
progress_callback: Optional callback(bytes_downloaded, total_bytes).
Returns:
Path to the downloaded file.
Raises:
RuntimeError: If download fails.
"""
try:
request = Request(release_info.download_url)
request.add_header("User-Agent", f"Clide/{CURRENT_VERSION}")
# Create temp file with appropriate extension
suffix = Path(release_info.download_url).suffix or ""
fd, temp_path = tempfile.mkstemp(suffix=suffix, prefix="clide_update_")
os.close(fd)
with urlopen(request, timeout=300) as response:
total_size = int(response.headers.get("Content-Length", 0))
downloaded = 0
chunk_size = 8192
with open(temp_path, "wb") as f:
while True:
chunk = response.read(chunk_size)
if not chunk:
break
f.write(chunk)
downloaded += len(chunk)
if progress_callback:
progress_callback(downloaded, total_size)
return Path(temp_path)
except Exception as e:
logger.error(f"Failed to download update: {e}")
raise RuntimeError(f"Download failed: {e}") from e
def get_executable_path() -> Path:
"""Get the path to the current executable."""
if getattr(sys, "frozen", False):
# Running as compiled executable
return Path(sys.executable)
else:
# Running as Python script
return Path(sys.argv[0]).resolve()
def apply_update(downloaded_file: Path) -> bool:
"""Apply the downloaded update.
This replaces the current executable with the new version.
On macOS/Linux, this can happen while the app is running.
On Windows, we need to use a helper script.
Args:
downloaded_file: Path to the downloaded update file.
Returns:
True if update was applied successfully.
"""
system = platform.system().lower()
current_exe = get_executable_path()
try:
if system == "darwin":
# macOS: For DMG, just inform user to install manually
# For direct binary updates, we can replace in-place
if downloaded_file.suffix == ".dmg":
logger.info(f"DMG downloaded to: {downloaded_file}")
return True # User needs to install manually
# Direct binary replacement
backup_path = current_exe.with_suffix(".backup")
shutil.copy2(current_exe, backup_path)
shutil.copy2(downloaded_file, current_exe)
current_exe.chmod(
current_exe.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH,
)
backup_path.unlink()
return True
elif system == "linux":
# Linux: AppImage can be replaced directly
if downloaded_file.suffix == ".AppImage":
backup_path = current_exe.with_suffix(".backup")
shutil.copy2(current_exe, backup_path)
shutil.copy2(downloaded_file, current_exe)
current_exe.chmod(
current_exe.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH,
)
backup_path.unlink()
return True
return False
elif system == "windows":
# Windows: Can't replace running executable directly
# Create a batch script to replace after exit
batch_script = current_exe.parent / "update_clide.bat"
new_exe = downloaded_file
script_content = f"""@echo off
echo Updating Clide...
timeout /t 2 /nobreak >nul
copy /y "{new_exe}" "{current_exe}"
del "{new_exe}"
del "%~f0"
echo Update complete!
"""
batch_script.write_text(script_content)
logger.info(f"Update script created: {batch_script}")
logger.info("Please restart Clide to complete the update.")
return True
return False
except Exception as e:
logger.error(f"Failed to apply update: {e}")
return False
finally:
# Clean up downloaded file if it still exists and wasn't moved
if downloaded_file.exists() and system != "windows":
try:
downloaded_file.unlink()
except Exception:
pass
def perform_update(progress_callback=None) -> tuple[bool, str]:
"""Check for and perform an update.
Args:
progress_callback: Optional callback for download progress.
Returns:
Tuple of (success, message).
"""
# Check for updates
result = check_for_updates()
if result.error:
return False, f"Failed to check for updates: {result.error}"
if not result.update_available:
return True, f"Already running the latest version ({result.current_version})"
if not result.release_info:
return False, "No release information available"
# Download update
try:
downloaded_file = download_update(result.release_info, progress_callback)
except RuntimeError as e:
return False, str(e)
# Apply update
if apply_update(downloaded_file):
system = platform.system().lower()
if system == "windows":
return True, f"Update to {result.latest_version} downloaded. Restart Clide to complete."
elif system == "darwin" and downloaded_file.suffix == ".dmg":
return (
True,
f"Update {result.latest_version} downloaded to {downloaded_file}. Please install manually.",
)
else:
return (
True,
f"Updated to {result.latest_version}. Restart Clide to use the new version.",
)
else:
return False, "Failed to apply update"