Initial commit: CodeMap v0.1.0 - CLI tool for code analysis and diagram generation

codemap/cli/watch.py

@@ -0,0 +1,124 @@
+import sys
+import time
+from pathlib import Path
+from typing import Optional
+from watchdog.observers import Observer
+from watchdog.events import FileSystemEventHandler
+import typer
+from rich.console import Console
+from codemap.cli.analyze import _collect_files, _display_results
+from codemap.parsers import PythonParser, JavaScriptParser, GoParser
+from codemap.core import GraphBuilder, MermaidGenerator
+from codemap.templates import render_html
+
+console = Console()
+
+class DiagramChangeHandler(FileSystemEventHandler):
+    def __init__(self, path: str, output: str, format: str):
+        self.path = Path(path)
+        self.output = output
+        self.format = format
+        self.debounce_time = 1.0
+        self.last_event = 0
+        self.processing = False
+
+    def on_any_event(self, event):
+        if self.processing:
+            return
+
+        current_time = time.time()
+        if current_time - self.last_event < self.debounce_time:
+            return
+
+        if event.is_directory:
+            return
+
+        if not any(event.src_path.endswith(ext) for ext in [".py", ".js", ".ts", ".jsx", ".tsx", ".go", ".mjs"]):
+            return
+
+        self.last_event = current_time
+        self.processing = True
+
+        try:
+            console.clear()
+            console.print(f"[bold green]Change detected: {event.event_type}[/bold green]")
+            console.print(f"[bold green]File: {event.src_path}[/bold green]")
+            console.print("")
+
+            self._regenerate()
+        finally:
+            self.processing = False
+
+    def _regenerate(self):
+        files = _collect_files(self.path)
+        if not files:
+            return
+
+        parsers = [PythonParser(), JavaScriptParser(), GoParser()]
+        parsed_files = []
+        errors = []
+
+        for file_path in files:
+            for parser in parsers:
+                if parser.can_parse(file_path):
+                    try:
+                        parsed = parser.parse(file_path)
+                        parsed_files.append(parsed)
+                    except Exception as e:
+                        errors.append((file_path, str(e)))
+                    break
+
+        builder = GraphBuilder()
+        builder.build_from_files(parsed_files)
+
+        graph_data = builder.get_graph_data()
+        mermaid_gen = MermaidGenerator(graph_data)
+
+        if self.format == "html":
+            output_content = render_html(mermaid_gen.generate_flowchart(), title="Code Map (Live)", auto_refresh=True)
+        else:
+            output_content = mermaid_gen.generate_flowchart()
+
+        if self.output:
+            Path(self.output).write_text(output_content, encoding="utf-8")
+            console.print(f"[green]Updated: {self.output}[/green]")
+        else:
+            _display_results(builder, errors, str(self.path))
+
+        console.print("")
+        console.print("[dim]Watching for changes... (Ctrl+C to stop)[/dim]")
+
+
+def watch(
+    path: str = typer.Argument(..., help="Directory to watch"),
+    output: Optional[str] = typer.Option(None, "--output", "-o", help="Output file path"),
+    format: str = typer.Option("mermaid", "--format", "-f", help="Output format (mermaid, html)"),
+    delay: float = typer.Option(1.0, "--delay", help="Debounce delay in seconds"),
+) -> None:
+    target_path = Path(path)
+    if not target_path.exists():
+        console.print(f"[red]Error: Path '{path}' does not exist[/red]")
+        sys.exit(1)
+
+    if not target_path.is_dir():
+        console.print(f"[red]Error: '{path}' must be a directory[/red]")
+        sys.exit(1)
+
+    console.print(f"[bold green]Starting watch mode for: {path}[/bold green]")
+    console.print("[dim]Press Ctrl+C to stop[/dim]")
+    console.print("")
+
+    event_handler = DiagramChangeHandler(path, output, format)
+    observer = Observer()
+    observer.schedule(event_handler, str(target_path), recursive=True)
+    observer.start()
+
+    try:
+        event_handler._regenerate()
+        while True:
+            time.sleep(0.5)
+    except KeyboardInterrupt:
+        console.print("\n[yellow]Stopping watcher...[/yellow]")
+        observer.stop()
+
+    observer.join()