From 424b2e7573c57468ae514c12a011e1136249466c Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Fri, 30 Jan 2026 18:58:56 +0000 Subject: [PATCH] Add CLI and services modules --- src/codexchange/cli/commands.py | 321 ++++++++++++++++++++++++++++++++ 1 file changed, 321 insertions(+) create mode 100644 src/codexchange/cli/commands.py diff --git a/src/codexchange/cli/commands.py b/src/codexchange/cli/commands.py new file mode 100644 index 0000000..6ec1cbc --- /dev/null +++ b/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)