Initial upload: GitPulse project files
This commit is contained in:
266
gitpulse/cli.py
Normal file
266
gitpulse/cli.py
Normal file
@@ -0,0 +1,266 @@
|
||||
"""CLI entry point for GitPulse."""
|
||||
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import click
|
||||
from rich.console import Console
|
||||
|
||||
from gitpulse import __version__
|
||||
from gitpulse.analyzers import (
|
||||
AuthorAnalyzer,
|
||||
ChurnAnalyzer,
|
||||
CommitAnalyzer,
|
||||
ProductivityAnalyzer,
|
||||
)
|
||||
from gitpulse.config import Config
|
||||
from gitpulse.exporters import JSONExporter, MarkdownExporter
|
||||
from gitpulse.ui.render import render_analyze, render_authors, render_churn, render_heatmap, render_productivity
|
||||
from gitpulse.utils.git_utils import GitRepo
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
def get_config() -> Config:
|
||||
"""Load configuration from environment and config file."""
|
||||
return Config(
|
||||
default_path=os.environ.get("GITPULSE_DEFAULT_PATH", "."),
|
||||
cache_enabled=os.environ.get("GITPULSE_CACHE_ENABLED", "true").lower() == "true",
|
||||
)
|
||||
|
||||
|
||||
def parse_date(date_str: Optional[str]) -> Optional[datetime]:
|
||||
"""Parse date string to datetime."""
|
||||
if date_str is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
if date_str.endswith(" ago"):
|
||||
num = int(date_str.split()[0])
|
||||
unit = date_str.split()[1]
|
||||
if "week" in unit:
|
||||
return datetime.now() - timedelta(weeks=num)
|
||||
elif "day" in unit:
|
||||
return datetime.now() - timedelta(days=num)
|
||||
elif "month" in unit:
|
||||
return datetime.now() - timedelta(days=num * 30)
|
||||
elif "hour" in unit:
|
||||
return datetime.now() - timedelta(hours=num)
|
||||
else:
|
||||
return datetime.fromisoformat(date_str)
|
||||
except (ValueError, IndexError):
|
||||
return None
|
||||
|
||||
|
||||
def validate_repo(path: str) -> Path:
|
||||
"""Validate and return repository path."""
|
||||
repo_path = Path(path).expanduser().resolve()
|
||||
if not repo_path.exists():
|
||||
raise click.ClickException(f"Path does not exist: {repo_path}")
|
||||
if not (repo_path / ".git").exists():
|
||||
raise click.ClickException(f"Not a git repository: {repo_path}")
|
||||
return repo_path
|
||||
|
||||
|
||||
@click.group()
|
||||
@click.version_option(version=__version__)
|
||||
@click.option(
|
||||
"--path",
|
||||
default=".",
|
||||
help="Path to git repository",
|
||||
envvar="GITPULSE_DEFAULT_PATH",
|
||||
)
|
||||
@click.option(
|
||||
"--since",
|
||||
default=None,
|
||||
help="Start date for analysis (e.g., '2024-01-01' or '1 week ago')",
|
||||
)
|
||||
@click.option(
|
||||
"--until",
|
||||
default=None,
|
||||
help="End date for analysis",
|
||||
)
|
||||
@click.option(
|
||||
"--verbose",
|
||||
is_flag=True,
|
||||
default=False,
|
||||
help="Enable verbose output",
|
||||
)
|
||||
@click.pass_context
|
||||
def main(
|
||||
ctx: click.Context,
|
||||
path: str,
|
||||
since: Optional[str],
|
||||
until: Optional[str],
|
||||
verbose: bool,
|
||||
) -> None:
|
||||
"""GitPulse - Local Git Analytics CLI."""
|
||||
ctx.ensure_object(dict)
|
||||
ctx.obj["path"] = validate_repo(path)
|
||||
ctx.obj["since"] = parse_date(since)
|
||||
ctx.obj["until"] = parse_date(until)
|
||||
ctx.obj["verbose"] = verbose
|
||||
ctx.obj["config"] = get_config()
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.pass_context
|
||||
def analyze(ctx: click.Context) -> None:
|
||||
"""Show overview dashboard with all metrics."""
|
||||
repo_path = ctx.obj["path"]
|
||||
since = ctx.obj["since"]
|
||||
until = ctx.obj["until"]
|
||||
verbose = ctx.obj["verbose"]
|
||||
|
||||
try:
|
||||
repo = GitRepo(repo_path)
|
||||
commit_analyzer = CommitAnalyzer(repo)
|
||||
churn_analyzer = ChurnAnalyzer(repo)
|
||||
author_analyzer = AuthorAnalyzer(repo)
|
||||
productivity_analyzer = ProductivityAnalyzer(repo)
|
||||
|
||||
commit_data = commit_analyzer.analyze(since=since, until=until)
|
||||
churn_data = churn_analyzer.analyze(since=since, until=until)
|
||||
author_data = author_analyzer.analyze(since=since, until=until)
|
||||
productivity_data = productivity_analyzer.analyze(since=since, until=until)
|
||||
|
||||
render_analyze(
|
||||
console,
|
||||
commit_data,
|
||||
churn_data,
|
||||
author_data,
|
||||
productivity_data,
|
||||
verbose=verbose,
|
||||
)
|
||||
except Exception as e:
|
||||
raise click.ClickException(str(e))
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.pass_context
|
||||
def heatmap(ctx: click.Context) -> None:
|
||||
"""Display commit activity as a calendar-style heatmap."""
|
||||
repo_path = ctx.obj["path"]
|
||||
since = ctx.obj["since"]
|
||||
until = ctx.obj["until"]
|
||||
verbose = ctx.obj["verbose"]
|
||||
|
||||
try:
|
||||
repo = GitRepo(repo_path)
|
||||
analyzer = CommitAnalyzer(repo)
|
||||
data = analyzer.analyze(since=since, until=until)
|
||||
render_heatmap(console, data, verbose=verbose)
|
||||
except Exception as e:
|
||||
raise click.ClickException(str(e))
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.pass_context
|
||||
def churn(ctx: click.Context) -> None:
|
||||
"""Show code churn trends (lines added/deleted)."""
|
||||
repo_path = ctx.obj["path"]
|
||||
since = ctx.obj["since"]
|
||||
until = ctx.obj["until"]
|
||||
verbose = ctx.obj["verbose"]
|
||||
|
||||
try:
|
||||
repo = GitRepo(repo_path)
|
||||
analyzer = ChurnAnalyzer(repo)
|
||||
data = analyzer.analyze(since=since, until=until)
|
||||
render_churn(console, data, verbose=verbose)
|
||||
except Exception as e:
|
||||
raise click.ClickException(str(e))
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.pass_context
|
||||
def authors(ctx: click.Context) -> None:
|
||||
"""Display author contribution breakdown."""
|
||||
repo_path = ctx.obj["path"]
|
||||
since = ctx.obj["since"]
|
||||
until = ctx.obj["until"]
|
||||
verbose = ctx.obj["verbose"]
|
||||
|
||||
try:
|
||||
repo = GitRepo(repo_path)
|
||||
analyzer = AuthorAnalyzer(repo)
|
||||
data = analyzer.analyze(since=since, until=until)
|
||||
render_authors(console, data, verbose=verbose)
|
||||
except Exception as e:
|
||||
raise click.ClickException(str(e))
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.pass_context
|
||||
def productivity(ctx: click.Context) -> None:
|
||||
"""Show productivity scorecards."""
|
||||
repo_path = ctx.obj["path"]
|
||||
since = ctx.obj["since"]
|
||||
until = ctx.obj["until"]
|
||||
verbose = ctx.obj["verbose"]
|
||||
|
||||
try:
|
||||
repo = GitRepo(repo_path)
|
||||
analyzer = ProductivityAnalyzer(repo)
|
||||
data = analyzer.analyze(since=since, until=until)
|
||||
render_productivity(console, data, verbose=verbose)
|
||||
except Exception as e:
|
||||
raise click.ClickException(str(e))
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.option(
|
||||
"--format",
|
||||
type=click.Choice(["markdown", "json"]),
|
||||
default="markdown",
|
||||
help="Export format",
|
||||
)
|
||||
@click.option(
|
||||
"--output",
|
||||
"-o",
|
||||
type=click.Path(),
|
||||
default=None,
|
||||
help="Output file path",
|
||||
)
|
||||
@click.pass_context
|
||||
def export(ctx: click.Context, format: str, output: Optional[str]) -> None:
|
||||
"""Export analysis results to Markdown or JSON."""
|
||||
repo_path = ctx.obj["path"]
|
||||
since = ctx.obj["since"]
|
||||
until = ctx.obj["until"]
|
||||
|
||||
try:
|
||||
repo = GitRepo(repo_path)
|
||||
commit_analyzer = CommitAnalyzer(repo)
|
||||
churn_analyzer = ChurnAnalyzer(repo)
|
||||
author_analyzer = AuthorAnalyzer(repo)
|
||||
productivity_analyzer = ProductivityAnalyzer(repo)
|
||||
|
||||
commit_data = commit_analyzer.analyze(since=since, until=until)
|
||||
churn_data = churn_analyzer.analyze(since=since, until=until)
|
||||
author_data = author_analyzer.analyze(since=since, until=until)
|
||||
productivity_data = productivity_analyzer.analyze(since=since, until=until)
|
||||
|
||||
data = {
|
||||
"commit_analysis": commit_data,
|
||||
"churn_analysis": churn_data,
|
||||
"author_analysis": author_data,
|
||||
"productivity_analysis": productivity_data,
|
||||
}
|
||||
|
||||
if format == "markdown":
|
||||
exporter = MarkdownExporter()
|
||||
result = exporter.export(data)
|
||||
else:
|
||||
exporter = JSONExporter()
|
||||
result = exporter.export(data)
|
||||
|
||||
if output:
|
||||
Path(output).write_text(result)
|
||||
click.echo(f"Exported to {output}")
|
||||
else:
|
||||
click.echo(result)
|
||||
except Exception as e:
|
||||
raise click.ClickException(str(e))
|
||||
Reference in New Issue
Block a user