Add CLI and services modules

src/codexchange/cli/commands.py

@@ -0,0 +1,321 @@
+"""CLI commands for CodeXchange CLI."""
+
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from pathlib import Path
+from typing import Optional
+
+import typer
+from rich import print
+from rich.panel import Panel
+from rich.table import Table
+
+from codexchange.config import get_config
+from codexchange.models import (
+    ConversionRequest,
+    Language,
+)
+from codexchange.services.conversion_service import ConversionService
+from codexchange.services.ollama_service import connect, list_models
+from codexchange.utils.syntax_check import verify_syntax
+
+
+app = typer.Typer()
+
+
+def show_supported_languages() -> None:
+    """Display supported languages and conversions."""
+    table = Table(title="Supported Languages")
+    table.add_column("Language", style="cyan")
+    table.add_column("Extensions", style="green")
+    
+    for lang in Language:
+        extensions = {
+            Language.JAVASCRIPT: ".js, .jsx",
+            Language.TYPESCRIPT: ".ts, .tsx",
+            Language.PYTHON: ".py",
+            Language.JAVA: ".java",
+        }
+        table.add_row(lang.value, extensions[lang])
+    
+    print(table)
+    
+    print("\n[bold]Supported Conversions:[/bold]")
+    conversions = [
+        "JavaScript ↔ TypeScript",
+        "JavaScript ↔ Python",
+        "JavaScript ↔ Java",
+        "TypeScript ↔ Python",
+        "TypeScript ↔ Java",
+        "Python ↔ Java",
+    ]
+    for conv in conversions:
+        print(f"  • {conv}")
+
+
+def list_languages_cmd() -> None:
+    """List supported programming languages."""
+    show_supported_languages()
+
+
+def list_models_cmd() -> None:
+    """List available Ollama models."""
+    config = get_config()
+    
+    print(f"Connecting to Ollama at {config.ollama_host}...")
+    
+    try:
+        models = list_models(host=config.ollama_host, timeout=config.timeout)
+        
+        if not models:
+            print("[yellow]No models found. Make sure Ollama is running and has models pulled.[/yellow]")
+            return
+        
+        table = Table(title="Available Ollama Models")
+        table.add_column("Model Name", style="cyan")
+        table.add_column("Size", style="green")
+        table.add_column("Modified", style="yellow")
+        
+        for model in models:
+            table.add_row(
+                model.name,
+                model.size or "N/A",
+                model.modified_at or "N/A"
+            )
+        
+        print(table)
+        
+    except ConnectionError as e:
+        print(f"[red]Error: {e}[/red]")
+        raise typer.Exit(1)
+
+
+def convert_cmd(
+    input_file: str = typer.Argument(..., help="Input file path"),
+    output_file: Optional[str] = typer.Option(None, "--output", "-o", help="Output file path"),
+    source_language: str = typer.Option(..., "--from", "-f", help="Source language"),
+    target_language: str = typer.Option(..., "--to", "-t", help="Target language"),
+    model: Optional[str] = typer.Option(None, "--model", "-m", help="Ollama model to use"),
+    verify: bool = typer.Option(False, "--verify", "-v", help="Verify syntax after conversion"),
+) -> None:
+    """Convert a single file from one language to another."""
+    config = get_config()
+    
+    try:
+        source_lang = Language(source_language.lower())
+    except ValueError:
+        print(f"[red]Error: Unsupported source language '{source_language}'[/red]")
+        print("Use 'codexchange list-languages' to see supported languages.")
+        raise typer.Exit(1)
+    
+    try:
+        target_lang = Language(target_language.lower())
+    except ValueError:
+        print(f"[red]Error: Unsupported target language '{target_language}'[/red]")
+        print("Use 'codexchange list-languages' to see supported languages.")
+        raise typer.Exit(1)
+    
+    input_path = Path(input_file)
+    if not input_path.exists():
+        print(f"[red]Error: Input file '{input_file}' not found[/red]")
+        raise typer.Exit(1)
+    
+    try:
+        with open(input_path, "r") as f:
+            source_code = f.read()
+    except Exception as e:
+        print(f"[red]Error reading input file: {e}[/red]")
+        raise typer.Exit(1)
+    
+    model_name = model or config.default_model
+    
+    print(f"Converting {input_file} from {source_lang.value} to {target_lang.value}...")
+    print(f"Using model: {model_name}")
+    
+    request = ConversionRequest(
+        source_code=source_code,
+        source_language=source_lang,
+        target_language=target_lang,
+        model=model_name
+    )
+    
+    try:
+        ollama = connect(host=config.ollama_host, timeout=config.timeout)
+        service = ConversionService(ollama_service=ollama)
+        result = service.convert(request)
+        
+        if result.success:
+            print("[green]Conversion successful![/green]")
+            
+            if verify:
+                print("Verifying syntax...")
+                if result.converted_code:
+                    is_valid, warnings = verify_syntax(result.converted_code, target_lang)
+                    result.syntax_verified = is_valid
+                    result.syntax_warnings = warnings
+                    
+                    if is_valid:
+                        print("[green]Syntax verification passed![/green]")
+                    else:
+                        print("[yellow]Syntax verification warnings:[/yellow]")
+                        for warning in warnings:
+                            print(f"  • {warning}")
+            
+            if output_file:
+                output_path = Path(output_file)
+                try:
+                    output_path.parent.mkdir(parents=True, exist_ok=True)
+                    if result.converted_code:
+                        with open(output_path, "w") as f:
+                            f.write(result.converted_code)
+                    print(f"Output written to: {output_file}")
+                except Exception as e:
+                    print(f"[red]Error writing output file: {e}[/red]")
+                    raise typer.Exit(1)
+            else:
+                print("\n[bold]Converted code:[/bold]")
+                if result.converted_code:
+                    print(Panel(result.converted_code, title="Output"))
+        else:
+            print(f"[red]Conversion failed: {result.error_message}[/red]")
+            raise typer.Exit(1)
+            
+    except ConnectionError as e:
+        print(f"[red]Error: {e}[/red]")
+        raise typer.Exit(1)
+
+
+def batch_convert_cmd(
+    input_directory: str = typer.Argument(..., help="Input directory path"),
+    output_directory: Optional[str] = typer.Option(None, "--output", "-o", help="Output directory path"),
+    source_language: str = typer.Option(..., "--from", "-f", help="Source language"),
+    target_language: str = typer.Option(..., "--to", "-t", help="Target language"),
+    model: Optional[str] = typer.Option(None, "--model", "-m", help="Ollama model to use"),
+    recursive: bool = typer.Option(True, "--recursive", "-r", help="Recursively process subdirectories"),
+    concurrency: int = typer.Option(1, "--concurrency", "-c", help="Number of concurrent conversions"),
+    verify: bool = typer.Option(False, "--verify", "-v", help="Verify syntax after conversion"),
+) -> None:
+    """Convert multiple files in a directory."""
+    config = get_config()
+    
+    try:
+        source_lang = Language(source_language.lower())
+    except ValueError:
+        print(f"[red]Error: Unsupported source language '{source_language}'[/red]")
+        raise typer.Exit(1)
+    
+    try:
+        target_lang = Language(target_language.lower())
+    except ValueError:
+        print(f"[red]Error: Unsupported target language '{target_language}'[/red]")
+        raise typer.Exit(1)
+    
+    input_path = Path(input_directory)
+    if not input_path.exists():
+        print(f"[red]Error: Input directory '{input_directory}' not found[/red]")
+        raise typer.Exit(1)
+    
+    if not input_path.is_dir():
+        print(f"[red]Error: '{input_directory}' is not a directory[/red]")
+        raise typer.Exit(1)
+    
+    extensions = {
+        Language.JAVASCRIPT: [".js", ".jsx"],
+        Language.TYPESCRIPT: [".ts", ".tsx"],
+        Language.PYTHON: [".py"],
+        Language.JAVA: [".java"],
+    }
+    
+    source_extensions = extensions[source_lang]
+    
+    pattern = "**/*" if recursive else "*"
+    file_pattern = input_path.glob(pattern)
+    
+    source_files = [
+        f for f in file_pattern
+        if f.is_file() and f.suffix.lower() in source_extensions
+    ]
+    
+    if not source_files:
+        print(f"[yellow]No {source_lang.value} files found in {input_directory}[/yellow]")
+        return
+    
+    print(f"Found {len(source_files)} {source_lang.value} file(s) to convert")
+    
+    output_dir = None
+    if output_directory:
+        output_dir = Path(output_directory)
+        output_dir.mkdir(parents=True, exist_ok=True)
+    
+    model_name = model or config.default_model
+    
+    try:
+        ollama = connect(host=config.ollama_host, timeout=config.timeout)
+        service = ConversionService(ollama_service=ollama)
+    except ConnectionError as e:
+        print(f"[red]Error: {e}[/red]")
+        raise typer.Exit(1)
+    
+    results = []
+    
+    def convert_file(file_path: Path) -> tuple:
+        try:
+            with open(file_path, "r") as f:
+                source_code = f.read()
+            
+            request = ConversionRequest(
+                source_code=source_code,
+                source_language=source_lang,
+                target_language=target_lang,
+                model=model_name
+            )
+            
+            result = service.convert(request)
+            
+            output_path = None
+            if result.success and output_dir:
+                relative_path = file_path.relative_to(input_path)
+                output_path = output_dir / relative_path.with_suffix(target_lang.extension)
+                output_path.parent.mkdir(parents=True, exist_ok=True)
+                
+                if result.converted_code:
+                    with open(output_path, "w") as f:
+                        f.write(result.converted_code)
+                
+                if verify and result.converted_code:
+                    is_valid, warnings = verify_syntax(result.converted_code, target_lang)
+                    result.syntax_verified = is_valid
+                    result.syntax_warnings = warnings
+            
+            return (str(file_path), result)
+        except Exception:
+            return (str(file_path), None)
+    
+    print(f"Starting conversion with concurrency={concurrency}...")
+    
+    with ThreadPoolExecutor(max_workers=concurrency) as executor:
+        futures = {
+            executor.submit(convert_file, f): f for f in source_files
+        }
+        
+        for future in as_completed(futures):
+            file_path, result = future.result()
+            results.append((file_path, result))
+            
+            if result:
+                if result.success:
+                    print(f"[green]✓ {file_path}[/green]")
+                else:
+                    print(f"[red]✗ {file_path}: {result.error_message}[/red]")
+            else:
+                print(f"[red]✗ {file_path}: Unexpected error[/red]")
+    
+    success_count = sum(1 for _, r in results if r and r.success)
+    fail_count = len(results) - success_count
+    
+    print("\n[bold]Summary:[/bold]")
+    print(f"  Successfully converted: {success_count}")
+    print(f"  Failed: {fail_count}")
+    print(f"  Total: {len(results)}")
+    
+    if fail_count > 0:
+        raise typer.Exit(1)