This commit is contained in:
403
src/cli.py
Normal file
403
src/cli.py
Normal file
@@ -0,0 +1,403 @@
|
||||
"""CLI interface for Code Pattern Search."""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional, cast
|
||||
|
||||
import click
|
||||
from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
|
||||
from .config import Config
|
||||
from .github_client import GitHubClient, RateLimitExceededError
|
||||
from .pattern_matcher import PatternMatcher, PatternLibrary, MatchLocation
|
||||
from .cache_manager import CacheManager
|
||||
from .exporter import Exporter
|
||||
from .models import SearchResult
|
||||
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
@click.group()
|
||||
@click.option(
|
||||
"--cache-dir",
|
||||
type=click.Path(path_type=Path),
|
||||
help="Custom cache directory",
|
||||
)
|
||||
@click.option(
|
||||
"--cache-ttl",
|
||||
type=int,
|
||||
help="Cache TTL in seconds",
|
||||
)
|
||||
@click.option(
|
||||
"--token",
|
||||
envvar="GITHUB_TOKEN",
|
||||
help="GitHub API token",
|
||||
)
|
||||
@click.option(
|
||||
"--verbose",
|
||||
"-v",
|
||||
is_flag=True,
|
||||
help="Enable verbose output",
|
||||
)
|
||||
@click.pass_context
|
||||
def main(
|
||||
ctx: click.Context,
|
||||
cache_dir: Optional[Path],
|
||||
cache_ttl: Optional[int],
|
||||
token: Optional[str],
|
||||
verbose: bool,
|
||||
) -> None:
|
||||
"""Search GitHub repositories by code patterns."""
|
||||
ctx.ensure_object(dict)
|
||||
ctx.obj["config"] = Config(
|
||||
cache_dir=cache_dir,
|
||||
cache_ttl=cache_ttl or 3600,
|
||||
token=token,
|
||||
verbose=verbose,
|
||||
)
|
||||
ctx.obj["verbose"] = verbose
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.argument("pattern", type=str, default=None, required=False)
|
||||
@click.option(
|
||||
"--language", "-l",
|
||||
type=str,
|
||||
help="Filter by programming language",
|
||||
)
|
||||
@click.option(
|
||||
"--stars-min",
|
||||
type=int,
|
||||
default=0,
|
||||
help="Minimum star count",
|
||||
)
|
||||
@click.option(
|
||||
"--stars-max",
|
||||
type=int,
|
||||
default=None,
|
||||
help="Maximum star count",
|
||||
)
|
||||
@click.option(
|
||||
"--repos",
|
||||
type=int,
|
||||
default=10,
|
||||
help="Number of repositories to search",
|
||||
)
|
||||
@click.option(
|
||||
"--output", "-o",
|
||||
type=click.Path(path_type=Path),
|
||||
help="Export results to JSON file",
|
||||
)
|
||||
@click.option(
|
||||
"--use-cache/--no-cache",
|
||||
default=True,
|
||||
help="Use cached results",
|
||||
)
|
||||
@click.option(
|
||||
"--preset",
|
||||
type=str,
|
||||
help="Use a preset pattern from the library",
|
||||
)
|
||||
@click.pass_context
|
||||
def search(
|
||||
ctx: click.Context,
|
||||
pattern: Optional[str],
|
||||
language: Optional[str],
|
||||
stars_min: int,
|
||||
stars_max: Optional[int],
|
||||
repos: int,
|
||||
output: Optional[Path],
|
||||
use_cache: bool,
|
||||
preset: Optional[str],
|
||||
) -> None:
|
||||
"""Search for a pattern in GitHub repositories."""
|
||||
config: Config = ctx.obj["config"]
|
||||
verbose: bool = ctx.obj.get("verbose", False)
|
||||
|
||||
if preset:
|
||||
pattern = PatternLibrary.get_pattern(preset)
|
||||
if not pattern:
|
||||
console.print(f"[red]Error: Unknown preset '{preset}'[/red]")
|
||||
console.print(f"Available presets: {', '.join(PatternLibrary.list_presets())}")
|
||||
sys.exit(1)
|
||||
elif pattern is None:
|
||||
console.print("[red]Error: Pattern or --preset is required[/red]")
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
pattern_matcher = PatternMatcher(pattern)
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error: Invalid regex pattern: {e}[/red]")
|
||||
sys.exit(1)
|
||||
|
||||
cache_manager = CacheManager(
|
||||
cache_dir=config.cache_dir,
|
||||
ttl=config.cache_ttl,
|
||||
)
|
||||
|
||||
github_client = GitHubClient(
|
||||
token=config.token,
|
||||
verbose=verbose,
|
||||
)
|
||||
|
||||
console.print(f"Searching for pattern: [cyan]{pattern}[/cyan]")
|
||||
if language:
|
||||
console.print(f"Language filter: [cyan]{language}[/cyan]")
|
||||
console.print(f"Star range: [cyan]{stars_min} - {stars_max or 'unlimited'}[/cyan]")
|
||||
console.print()
|
||||
|
||||
try:
|
||||
results = search_repositories(
|
||||
github_client=github_client,
|
||||
pattern_matcher=pattern_matcher,
|
||||
language=language,
|
||||
stars_min=stars_min,
|
||||
stars_max=stars_max,
|
||||
max_repos=repos,
|
||||
cache_manager=cache_manager,
|
||||
use_cache=use_cache,
|
||||
verbose=verbose,
|
||||
console=console,
|
||||
)
|
||||
|
||||
if not results:
|
||||
console.print("[yellow]No matches found[/yellow]")
|
||||
return
|
||||
|
||||
sorted_results = sorted(
|
||||
results,
|
||||
key=lambda r: (r.score, r.total_matches),
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
display_results(sorted_results, console)
|
||||
|
||||
if output:
|
||||
Exporter.export_results(sorted_results, output)
|
||||
console.print(f"\n[green]Results exported to {output}[/green]")
|
||||
|
||||
except RateLimitExceededError as e:
|
||||
console.print("[red]Error: GitHub API rate limit exceeded[/red]")
|
||||
console.print(f"Reset time: {e.reset_time}")
|
||||
console.print("Tip: Set GITHUB_TOKEN environment variable for higher limits")
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error: {e}[/red]")
|
||||
if verbose:
|
||||
raise
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def search_repositories(
|
||||
github_client: GitHubClient,
|
||||
pattern_matcher: PatternMatcher,
|
||||
language: Optional[str],
|
||||
stars_min: int,
|
||||
stars_max: Optional[int],
|
||||
max_repos: int,
|
||||
cache_manager: CacheManager,
|
||||
use_cache: bool,
|
||||
verbose: bool,
|
||||
console: Console,
|
||||
) -> list[SearchResult]:
|
||||
"""Search repositories for the pattern."""
|
||||
|
||||
search_key = f"search:{pattern_matcher.pattern}:{language}:{stars_min}:{stars_max}:{max_repos}"
|
||||
|
||||
if use_cache:
|
||||
cached = cache_manager.get(search_key)
|
||||
if cached:
|
||||
console.print("[dim]Using cached search results[/dim]")
|
||||
return cached
|
||||
|
||||
if verbose:
|
||||
console.print("Fetching repositories...")
|
||||
|
||||
repos = github_client.search_repositories(
|
||||
language=language,
|
||||
stars_min=stars_min,
|
||||
stars_max=stars_max,
|
||||
per_page=max_repos,
|
||||
)
|
||||
|
||||
if not repos:
|
||||
return []
|
||||
|
||||
results: list[SearchResult] = []
|
||||
|
||||
def process_repo(repo) -> Optional[SearchResult]:
|
||||
repo_name = repo.full_name
|
||||
|
||||
if verbose:
|
||||
console.print(f" Processing {repo_name}...")
|
||||
|
||||
cache_key = f"repo:{repo_name}:{pattern_matcher.pattern}"
|
||||
|
||||
if use_cache:
|
||||
cached_result = cache_manager.get(cache_key)
|
||||
if cached_result:
|
||||
return cached_result
|
||||
|
||||
try:
|
||||
file_tree = github_client.get_file_tree(repo.owner.login, repo.name)
|
||||
|
||||
if not file_tree:
|
||||
return None
|
||||
|
||||
matches: list[MatchLocation] = []
|
||||
|
||||
for item in file_tree:
|
||||
item_dict = cast(dict[str, Any], item)
|
||||
file_path_item = item_dict["path"]
|
||||
if not pattern_matcher.matches_extension(file_path_item):
|
||||
continue
|
||||
|
||||
content = github_client.get_file_content(
|
||||
repo.owner.login,
|
||||
repo.name,
|
||||
file_path_item,
|
||||
)
|
||||
|
||||
if content:
|
||||
file_matches = pattern_matcher.find_matches(content, file_path_item)
|
||||
matches.extend(file_matches)
|
||||
|
||||
if matches:
|
||||
result = SearchResult(
|
||||
repo_name=repo_name,
|
||||
repo_url=repo.html_url,
|
||||
stars=repo.stargazers_count,
|
||||
description=repo.description,
|
||||
language=repo.language,
|
||||
matches=matches,
|
||||
total_matches=len(matches),
|
||||
score=calculate_score(repo.stargazers_count, len(matches)),
|
||||
)
|
||||
|
||||
cache_manager.set(cache_key, result)
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
if verbose:
|
||||
console.print(f" [yellow]Warning: Error processing {repo_name}: {e}[/yellow]")
|
||||
|
||||
return None
|
||||
|
||||
with console.status("[bold]Searching repositories...") as status:
|
||||
for repo in repos:
|
||||
status.update(f"Processing {repo.full_name}...")
|
||||
result = process_repo(repo)
|
||||
if result:
|
||||
results.append(result)
|
||||
|
||||
cache_manager.set(search_key, results)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def calculate_score(stars: int, match_count: int) -> float:
|
||||
"""Calculate relevance score based on stars and match count."""
|
||||
import math
|
||||
|
||||
star_score = math.log10(max(stars, 1)) * 10
|
||||
match_score = match_count * 5
|
||||
|
||||
return round(star_score + match_score, 2)
|
||||
|
||||
|
||||
def display_results(results: list[SearchResult], console: Console) -> None:
|
||||
"""Display search results in a formatted table."""
|
||||
table = Table(show_header=True, header_style="bold magenta")
|
||||
table.add_column("Repository", width=40)
|
||||
table.add_column("Stars")
|
||||
table.add_column("Matches")
|
||||
table.add_column("Score")
|
||||
|
||||
for result in results:
|
||||
repo_link = f"[link={result.repo_url}]{result.repo_name}[/link]"
|
||||
table.add_row(
|
||||
repo_link,
|
||||
str(result.stars),
|
||||
str(result.total_matches),
|
||||
str(result.score),
|
||||
)
|
||||
|
||||
console.print(Panel(table, title="Search Results", expand=False))
|
||||
|
||||
if len(results) > 0:
|
||||
console.print(f"\n[dim]Found {len(results)} repositories with matches[/dim]")
|
||||
|
||||
|
||||
@main.command()
|
||||
def presets() -> None:
|
||||
"""List available pattern presets."""
|
||||
presets_list = PatternLibrary.list_presets()
|
||||
|
||||
table = Table(show_header=True, header_style="bold magenta")
|
||||
table.add_column("Preset Name")
|
||||
table.add_column("Pattern")
|
||||
|
||||
for name in presets_list:
|
||||
pattern = PatternLibrary.get_pattern(name)
|
||||
table.add_row(name, pattern)
|
||||
|
||||
console.print(Panel(table, title="Available Presets", expand=False))
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.option(
|
||||
"--all",
|
||||
is_flag=True,
|
||||
help="Show all cached entries",
|
||||
)
|
||||
@click.option(
|
||||
"--clear",
|
||||
is_flag=True,
|
||||
help="Clear all cached data",
|
||||
)
|
||||
def cache(all: bool, clear: bool) -> None:
|
||||
"""Manage cache."""
|
||||
cache_manager = CacheManager()
|
||||
|
||||
if clear:
|
||||
cache_manager.clear()
|
||||
console.print("[green]Cache cleared successfully[/green]")
|
||||
return
|
||||
|
||||
stats = cache_manager.get_stats()
|
||||
|
||||
if all:
|
||||
entries = cache_manager.get_all()
|
||||
if entries:
|
||||
table = Table(show_header=True, header_style="bold magenta")
|
||||
table.add_column("Key", width=60)
|
||||
table.add_column("Type")
|
||||
table.add_column("Size")
|
||||
|
||||
for key, value in entries.items():
|
||||
value_type = type(value).__name__
|
||||
size = len(str(value))
|
||||
table.add_row(key[:60], value_type, str(size))
|
||||
|
||||
console.print(Panel(table, title="Cached Entries", expand=False))
|
||||
else:
|
||||
console.print("[yellow]Cache is empty[/yellow]")
|
||||
|
||||
console.print("\nCache statistics:")
|
||||
console.print(f" Total entries: {stats['size']}")
|
||||
console.print(f" Cache size: {stats['cache_size_mb']:.2f} MB")
|
||||
|
||||
|
||||
@main.command()
|
||||
def version() -> None:
|
||||
"""Show version information."""
|
||||
from . import __version__
|
||||
|
||||
console.print(f"[bold]Code Pattern Search CLI[/bold] v{__version__}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user