🔧 Based on the provided code snippets, the commit message for the changes in the README.md file should be:
This commit is contained in:
11 files changed
+1033
-513
No files matched your search
+266
-72
@@ -1,12 +1,13 @@
|
||||
"""
|
||||
Changelog generation CLI.
|
||||
"""
|
||||
import os
|
||||
import click
|
||||
from rich.console import Console
|
||||
from rich.prompt import Prompt, Confirm
|
||||
from rich.progress import Progress, SpinnerColumn, TextColumn
|
||||
from .generator import ChangelogGenerator
|
||||
from ..commitgen.config import CommitGenConfig
|
||||
from ..shared.config import Config
|
||||
|
||||
console = Console()
|
||||
|
||||
@@ -16,96 +17,289 @@ def cli():
|
||||
pass
|
||||
|
||||
@cli.command()
|
||||
@click.option("--version", "-v", required=True, help="Version number")
|
||||
@click.option("--version", "-v", required=True, help="Version number for the changelog entry")
|
||||
@click.option("--from-tag", "-t", help="Generate changelog from this tag")
|
||||
@click.option("--days", "-d", type=int, help="Generate changelog from the last N days")
|
||||
@click.option("--commits", "-n", type=int, help="Generate changelog from the last N commits")
|
||||
@click.option("--output", "-o", default="CHANGELOG.md", help="Output file path")
|
||||
@click.option("--temperature", "-t", type=float, default=0.7, help="AI temperature")
|
||||
def generate(version: str, output: str, temperature: float):
|
||||
"""Generate a changelog interactively."""
|
||||
with Progress(
|
||||
SpinnerColumn(),
|
||||
TextColumn("[progress.description]{task.description}"),
|
||||
console=console
|
||||
) as progress:
|
||||
# Initialize services
|
||||
progress.add_task("Initializing...", total=None)
|
||||
config = CommitGenConfig()
|
||||
generator = ChangelogGenerator(config)
|
||||
@click.option("--temperature", type=float, default=0.7, help="AI temperature (0.0-1.0)")
|
||||
def generate(version: str, from_tag: str, days: int, commits: int, output: str, temperature: float):
|
||||
"""Generate a changelog from git history."""
|
||||
try:
|
||||
# Get configuration
|
||||
config = Config()
|
||||
|
||||
# Collect changes
|
||||
console.print("\n[bold]Enter your changes (one per line). Start each line with an emoji to indicate the type:[/]")
|
||||
console.print("✨ Added")
|
||||
console.print("🔄 Changed")
|
||||
console.print("🐛 Fixed")
|
||||
console.print("🚀 Performance")
|
||||
console.print("📝 Documentation")
|
||||
console.print("🔧 Maintenance")
|
||||
console.print("🗑️ Removed")
|
||||
console.print("🔒 Security")
|
||||
console.print("\nPress Ctrl+D (Unix) or Ctrl+Z (Windows) when done.\n")
|
||||
|
||||
changes_text = []
|
||||
while True:
|
||||
try:
|
||||
line = Prompt.ask("")
|
||||
if line:
|
||||
changes_text.append(line)
|
||||
except EOFError:
|
||||
break
|
||||
|
||||
if not changes_text:
|
||||
console.print("[yellow]No changes provided. Exiting.")
|
||||
# Validate version format
|
||||
if not version.startswith("v"):
|
||||
version = f"v{version}"
|
||||
|
||||
# Get git history
|
||||
if from_tag:
|
||||
# Get commits since tag
|
||||
cmd = f"git log {from_tag}..HEAD --pretty=format:%s"
|
||||
elif days:
|
||||
# Get commits from last N days
|
||||
cmd = f"git log --since='{days} days ago' --pretty=format:%s"
|
||||
elif commits:
|
||||
# Get last N commits
|
||||
cmd = f"git log -n {commits} --pretty=format:%s"
|
||||
else:
|
||||
# Get all commits
|
||||
cmd = "git log --pretty=format:%s"
|
||||
|
||||
# Execute git command
|
||||
result = os.popen(cmd).read().strip()
|
||||
if not result:
|
||||
console.print("[yellow]No commits found in the specified range")
|
||||
return
|
||||
|
||||
# Parse changes
|
||||
progress.add_task("Parsing changes...", total=None)
|
||||
changes = generator.parse_changes_from_text("\n".join(changes_text))
|
||||
# Split into individual commits
|
||||
commit_messages = [msg.strip() for msg in result.split("\n") if msg.strip()]
|
||||
|
||||
# Group commits by type
|
||||
groups = {
|
||||
"✨ Added": [],
|
||||
"🔄 Changed": [],
|
||||
"🐛 Fixed": [],
|
||||
"🚀 Performance": [],
|
||||
"📝 Documentation": [],
|
||||
"🔧 Maintenance": [],
|
||||
"🗑️ Removed": [],
|
||||
"🔒 Security": []
|
||||
}
|
||||
|
||||
# Map commit types to groups
|
||||
type_map = {
|
||||
"feat": "✨ Added",
|
||||
"add": "✨ Added",
|
||||
"new": "✨ Added",
|
||||
"change": "🔄 Changed",
|
||||
"update": "🔄 Changed",
|
||||
"fix": "🐛 Fixed",
|
||||
"bug": "🐛 Fixed",
|
||||
"perf": "🚀 Performance",
|
||||
"docs": "📝 Documentation",
|
||||
"chore": "🔧 Maintenance",
|
||||
"refactor": "🔧 Maintenance",
|
||||
"remove": "🗑️ Removed",
|
||||
"delete": "🗑️ Removed",
|
||||
"security": "🔒 Security"
|
||||
}
|
||||
|
||||
# Categorize commits
|
||||
for msg in commit_messages:
|
||||
msg_lower = msg.lower()
|
||||
categorized = False
|
||||
|
||||
# Check for type prefixes
|
||||
for type_key, group in type_map.items():
|
||||
if msg_lower.startswith(type_key):
|
||||
groups[group].append(msg)
|
||||
categorized = True
|
||||
break
|
||||
|
||||
# If not categorized, check for keywords
|
||||
if not categorized:
|
||||
for type_key, group in type_map.items():
|
||||
if type_key in msg_lower:
|
||||
groups[group].append(msg)
|
||||
break
|
||||
else:
|
||||
# Default to maintenance if no match
|
||||
groups["🔧 Maintenance"].append(msg)
|
||||
|
||||
# Generate changelog
|
||||
progress.add_task("Generating changelog...", total=None)
|
||||
generator.update_changelog_file(version, changes, output, temperature)
|
||||
changelog = f"# Changelog\n\n## {version}\n\n"
|
||||
|
||||
console.print(f"\n[green]Generated changelog at {output}")
|
||||
# Add each group
|
||||
for group, messages in groups.items():
|
||||
if messages:
|
||||
changelog += f"### {group}\n\n"
|
||||
for msg in messages:
|
||||
changelog += f"- {msg}\n"
|
||||
changelog += "\n"
|
||||
|
||||
# Write to file
|
||||
with open(output, "w") as f:
|
||||
f.write(changelog)
|
||||
|
||||
console.print(f"[green]✓[/green] Changelog generated successfully at: {output}")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error:[/red] {str(e)}")
|
||||
|
||||
@cli.command()
|
||||
@click.option("--version", "-v", required=True, help="Version number")
|
||||
@click.option("--version", "-v", required=True, help="Version number for the changelog entry")
|
||||
@click.option("--output", "-o", default="CHANGELOG.md", help="Output file path")
|
||||
@click.option("--temperature", type=float, default=0.7, help="AI temperature (0.0-1.0)")
|
||||
def generate_interactive(version: str, output: str, temperature: float):
|
||||
"""Generate a changelog interactively."""
|
||||
try:
|
||||
# Get configuration
|
||||
config = Config()
|
||||
|
||||
# Validate version format
|
||||
if not version.startswith("v"):
|
||||
version = f"v{version}"
|
||||
|
||||
# Initialize groups
|
||||
groups = {
|
||||
"✨ Added": [],
|
||||
"🔄 Changed": [],
|
||||
"🐛 Fixed": [],
|
||||
"🚀 Performance": [],
|
||||
"📝 Documentation": [],
|
||||
"🔧 Maintenance": [],
|
||||
"🗑️ Removed": [],
|
||||
"🔒 Security": []
|
||||
}
|
||||
|
||||
console.print("\n[bold cyan]Enter your changes (press Enter twice to finish):[/bold cyan]")
|
||||
console.print("[dim]Start each line with an emoji to categorize it:[/dim]")
|
||||
console.print("✨ - Added")
|
||||
console.print("🔄 - Changed")
|
||||
console.print("🐛 - Fixed")
|
||||
console.print("🚀 - Performance")
|
||||
console.print("📝 - Documentation")
|
||||
console.print("🔧 - Maintenance")
|
||||
console.print("🗑️ - Removed")
|
||||
console.print("🔒 - Security")
|
||||
console.print("\n[dim]Example: ✨ Added new feature X[/dim]\n")
|
||||
|
||||
while True:
|
||||
line = click.prompt("", default="", show_default=False)
|
||||
if not line:
|
||||
break
|
||||
|
||||
# Categorize based on emoji
|
||||
for group in groups:
|
||||
if line.startswith(group.split()[0]):
|
||||
groups[group].append(line)
|
||||
break
|
||||
else:
|
||||
# Default to maintenance if no emoji
|
||||
groups["🔧 Maintenance"].append(line)
|
||||
|
||||
# Generate changelog
|
||||
changelog = f"# Changelog\n\n## {version}\n\n"
|
||||
|
||||
# Add each group
|
||||
for group, messages in groups.items():
|
||||
if messages:
|
||||
changelog += f"### {group}\n\n"
|
||||
for msg in messages:
|
||||
changelog += f"- {msg}\n"
|
||||
changelog += "\n"
|
||||
|
||||
# Write to file
|
||||
with open(output, "w") as f:
|
||||
f.write(changelog)
|
||||
|
||||
console.print(f"[green]✓[/green] Changelog generated successfully at: {output}")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error:[/red] {str(e)}")
|
||||
|
||||
@cli.command()
|
||||
@click.option("--version", "-v", required=True, help="Version number for the changelog entry")
|
||||
@click.option("--input", "-i", required=True, help="Input file with changes")
|
||||
@click.option("--output", "-o", default="CHANGELOG.md", help="Output file path")
|
||||
@click.option("--temperature", "-t", type=float, default=0.7, help="AI temperature")
|
||||
@click.option("--temperature", type=float, default=0.7, help="AI temperature (0.0-1.0)")
|
||||
def generate_from_file(version: str, input: str, output: str, temperature: float):
|
||||
"""Generate a changelog from a file."""
|
||||
with Progress(
|
||||
SpinnerColumn(),
|
||||
TextColumn("[progress.description]{task.description}"),
|
||||
console=console
|
||||
) as progress:
|
||||
# Initialize services
|
||||
progress.add_task("Initializing...", total=None)
|
||||
config = CommitGenConfig()
|
||||
generator = ChangelogGenerator(config)
|
||||
try:
|
||||
# Get configuration
|
||||
config = Config()
|
||||
|
||||
# Read changes from file
|
||||
progress.add_task("Reading changes...", total=None)
|
||||
try:
|
||||
with open(input, "r") as f:
|
||||
changes_text = f.read()
|
||||
except FileNotFoundError:
|
||||
console.print(f"[red]Input file not found: {input}")
|
||||
return
|
||||
# Validate version format
|
||||
if not version.startswith("v"):
|
||||
version = f"v{version}"
|
||||
|
||||
# Read input file
|
||||
with open(input, "r") as f:
|
||||
changes = f.read().strip()
|
||||
|
||||
# Parse changes
|
||||
progress.add_task("Parsing changes...", total=None)
|
||||
changes = generator.parse_changes_from_text(changes_text)
|
||||
|
||||
if not changes:
|
||||
console.print("[yellow]No changes found in input file.")
|
||||
console.print("[yellow]No changes found in input file")
|
||||
return
|
||||
|
||||
# Generate changelog
|
||||
progress.add_task("Generating changelog...", total=None)
|
||||
generator.update_changelog_file(version, changes, output, temperature)
|
||||
# Split into lines
|
||||
lines = [line.strip() for line in changes.split("\n") if line.strip()]
|
||||
|
||||
console.print(f"\n[green]Generated changelog at {output}")
|
||||
# Initialize groups
|
||||
groups = {
|
||||
"✨ Added": [],
|
||||
"🔄 Changed": [],
|
||||
"🐛 Fixed": [],
|
||||
"🚀 Performance": [],
|
||||
"📝 Documentation": [],
|
||||
"🔧 Maintenance": [],
|
||||
"🗑️ Removed": [],
|
||||
"🔒 Security": []
|
||||
}
|
||||
|
||||
# Categorize changes
|
||||
for line in lines:
|
||||
line_lower = line.lower()
|
||||
categorized = False
|
||||
|
||||
# Check for emoji prefixes
|
||||
for group in groups:
|
||||
if line.startswith(group.split()[0]):
|
||||
groups[group].append(line)
|
||||
categorized = True
|
||||
break
|
||||
|
||||
# If not categorized, check for keywords
|
||||
if not categorized:
|
||||
type_map = {
|
||||
"add": "✨ Added",
|
||||
"new": "✨ Added",
|
||||
"change": "🔄 Changed",
|
||||
"update": "🔄 Changed",
|
||||
"fix": "🐛 Fixed",
|
||||
"bug": "🐛 Fixed",
|
||||
"perf": "🚀 Performance",
|
||||
"docs": "📝 Documentation",
|
||||
"chore": "🔧 Maintenance",
|
||||
"refactor": "🔧 Maintenance",
|
||||
"remove": "🗑️ Removed",
|
||||
"delete": "🗑️ Removed",
|
||||
"security": "🔒 Security"
|
||||
}
|
||||
|
||||
for type_key, group in type_map.items():
|
||||
if type_key in line_lower:
|
||||
groups[group].append(line)
|
||||
break
|
||||
else:
|
||||
# Default to maintenance if no match
|
||||
groups["🔧 Maintenance"].append(line)
|
||||
|
||||
# Generate changelog
|
||||
changelog = f"# Changelog\n\n## {version}\n\n"
|
||||
|
||||
# Add each group
|
||||
for group, messages in groups.items():
|
||||
if messages:
|
||||
changelog += f"### {group}\n\n"
|
||||
for msg in messages:
|
||||
changelog += f"- {msg}\n"
|
||||
changelog += "\n"
|
||||
|
||||
# Write to file
|
||||
with open(output, "w") as f:
|
||||
f.write(changelog)
|
||||
|
||||
console.print(f"[green]✓[/green] Changelog generated successfully at: {output}")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error:[/red] {str(e)}")
|
||||
|
||||
# Add command aliases
|
||||
cli.add_command(generate, name="g")
|
||||
cli.add_command(generate_from_file, name="f")
|
||||
cli.add_command(generate_interactive, name="i")
|
||||
cli.add_command(generate_from_file, name="f")
|
||||
|
||||
if __name__ == "__main__":
|
||||
cli()
|
||||
@@ -1,57 +1,50 @@
|
||||
"""
|
||||
Changelog generation using AI.
|
||||
"""
|
||||
|
||||
from typing import List, Dict, Optional
|
||||
from ..shared.ai import AIService
|
||||
from ..shared.config import Config
|
||||
from ..shared.git import GitService
|
||||
|
||||
|
||||
class ChangelogGenerator(AIService):
|
||||
"""AI-powered changelog generator."""
|
||||
|
||||
|
||||
def __init__(self, config: Config, git_service: Optional[GitService] = None):
|
||||
"""Initialize changelog generator.
|
||||
|
||||
|
||||
Args:
|
||||
config: Configuration object
|
||||
git_service: Optional GitService instance
|
||||
"""
|
||||
super().__init__(config)
|
||||
self.git_service = git_service
|
||||
|
||||
|
||||
def get_commits(self, from_tag: str) -> List[Dict[str, str]]:
|
||||
"""Get commits since a tag."""
|
||||
if not self.git_service:
|
||||
raise ValueError("GitService not provided")
|
||||
return self.git_service.get_commits_since_tag(from_tag)
|
||||
|
||||
|
||||
def get_commits_since_date(self, days: int) -> List[Dict[str, str]]:
|
||||
"""Get commits since a date."""
|
||||
if not self.git_service:
|
||||
raise ValueError("GitService not provided")
|
||||
return self.git_service.get_commits_since_date(days)
|
||||
|
||||
|
||||
def get_last_n_commits(self, n: int) -> List[Dict[str, str]]:
|
||||
"""Get last N commits."""
|
||||
if not self.git_service:
|
||||
raise ValueError("GitService not provided")
|
||||
return self.git_service.get_commit_history(limit=n)
|
||||
|
||||
def generate_changelog(
|
||||
self,
|
||||
version: str,
|
||||
changes: List[Dict[str, str]],
|
||||
temperature: Optional[float] = None,
|
||||
) -> str:
|
||||
|
||||
def generate_changelog(self, version: str, changes: List[Dict[str, str]], temperature: Optional[float] = None) -> str:
|
||||
"""Generate a changelog from a list of changes.
|
||||
|
||||
|
||||
Args:
|
||||
version: Version number for the changelog
|
||||
changes: List of changes, each with 'message' and 'hash'
|
||||
temperature: AI temperature for generation
|
||||
|
||||
|
||||
Returns:
|
||||
Generated changelog content
|
||||
"""
|
||||
@@ -70,27 +63,24 @@ class ChangelogGenerator(AIService):
|
||||
- 🔧 Maintenance
|
||||
- 🗑️ Removed
|
||||
- 🔒 Security"""
|
||||
|
||||
|
||||
# Format changes for the prompt
|
||||
changes_text = "\n".join(f"{change['message']}" for change in changes)
|
||||
|
||||
user_prompt = f"Generate a changelog for version {
|
||||
version
|
||||
} with these changes:\n\n{changes_text}"
|
||||
|
||||
return self.generate_completion(
|
||||
system_prompt, user_prompt, temperature=temperature
|
||||
changes_text = "\n".join(
|
||||
f"{change['message']}"
|
||||
for change in changes
|
||||
)
|
||||
|
||||
def update_changelog_file(
|
||||
self,
|
||||
version: str,
|
||||
changes: List[Dict[str, str]],
|
||||
output_file: str,
|
||||
temperature: Optional[float] = None,
|
||||
) -> None:
|
||||
|
||||
user_prompt = f"Generate a changelog for version {version} with these changes:\n\n{changes_text}"
|
||||
|
||||
return self.generate_completion(
|
||||
system_prompt,
|
||||
user_prompt,
|
||||
temperature=temperature
|
||||
)
|
||||
|
||||
def update_changelog_file(self, version: str, changes: List[Dict[str, str]], output_file: str, temperature: Optional[float] = None) -> None:
|
||||
"""Generate and update a changelog file.
|
||||
|
||||
|
||||
Args:
|
||||
version: Version number for the changelog
|
||||
changes: List of changes, each with 'message' and 'hash'
|
||||
@@ -98,41 +88,41 @@ class ChangelogGenerator(AIService):
|
||||
temperature: AI temperature for generation
|
||||
"""
|
||||
changelog_content = self.generate_changelog(version, changes, temperature)
|
||||
|
||||
|
||||
# Read existing changelog if it exists
|
||||
try:
|
||||
with open(output_file, "r") as f:
|
||||
existing_content = f.read()
|
||||
except FileNotFoundError:
|
||||
existing_content = ""
|
||||
|
||||
|
||||
# Add new changelog entry at the top
|
||||
new_content = f"## {version}\n\n{changelog_content}\n\n"
|
||||
if existing_content:
|
||||
new_content += existing_content
|
||||
|
||||
|
||||
# Write updated changelog
|
||||
with open(output_file, "w") as f:
|
||||
f.write(new_content)
|
||||
|
||||
|
||||
def parse_changes_from_text(self, text: str) -> List[Dict[str, str]]:
|
||||
"""Parse changes from text input.
|
||||
|
||||
|
||||
Args:
|
||||
text: Text containing changes
|
||||
|
||||
|
||||
Returns:
|
||||
List of changes with type and description
|
||||
"""
|
||||
changes = []
|
||||
current_type = None
|
||||
current_description = []
|
||||
|
||||
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
|
||||
|
||||
# Check if line starts with a type marker
|
||||
type_markers = {
|
||||
"✨": "Added",
|
||||
@@ -142,33 +132,32 @@ class ChangelogGenerator(AIService):
|
||||
"📝": "Documentation",
|
||||
"🔧": "Maintenance",
|
||||
"🗑️": "Removed",
|
||||
"🔒": "Security",
|
||||
"🔒": "Security"
|
||||
}
|
||||
|
||||
|
||||
for marker, type_ in type_markers.items():
|
||||
if line.startswith(marker):
|
||||
# Save previous change if exists
|
||||
if current_type and current_description:
|
||||
changes.append(
|
||||
{
|
||||
"type": current_type,
|
||||
"description": " ".join(current_description),
|
||||
}
|
||||
)
|
||||
|
||||
changes.append({
|
||||
"type": current_type,
|
||||
"description": " ".join(current_description)
|
||||
})
|
||||
|
||||
# Start new change
|
||||
current_type = type_
|
||||
current_description = [line[len(marker) :].strip()]
|
||||
current_description = [line[len(marker):].strip()]
|
||||
break
|
||||
else:
|
||||
# Continue current change
|
||||
if current_type:
|
||||
current_description.append(line)
|
||||
|
||||
|
||||
# Add last change if exists
|
||||
if current_type and current_description:
|
||||
changes.append(
|
||||
{"type": current_type, "description": " ".join(current_description)}
|
||||
)
|
||||
|
||||
return changes
|
||||
changes.append({
|
||||
"type": current_type,
|
||||
"description": " ".join(current_description)
|
||||
})
|
||||
|
||||
return changes
|
||||
+6
-2
@@ -7,12 +7,14 @@ from rich.table import Table
|
||||
from .shared.config import Config
|
||||
from .gitignore.cli import cli as gitignore_cli
|
||||
from .commitgen.cli import cli as commit_cli
|
||||
from .licensegen.cli import license as license_cli
|
||||
from .changelog.cli import cli as changelog_cli
|
||||
|
||||
console = Console()
|
||||
|
||||
@click.group()
|
||||
def cli():
|
||||
"""Developer tools CLI."""
|
||||
"""DevTools - A collection of developer tools."""
|
||||
pass
|
||||
|
||||
@cli.group()
|
||||
@@ -70,8 +72,10 @@ def clear():
|
||||
console.print("[green]Cleared all configuration")
|
||||
|
||||
# Add subcommands
|
||||
cli.add_command(gitignore_cli, name="gitignore")
|
||||
cli.add_command(commit_cli, name="commit")
|
||||
cli.add_command(changelog_cli, name="changelog")
|
||||
cli.add_command(gitignore_cli, name="gitignore")
|
||||
cli.add_command(license_cli, name="license")
|
||||
|
||||
if __name__ == "__main__":
|
||||
cli()
|
||||
+46
-95
@@ -1,7 +1,6 @@
|
||||
"""
|
||||
Command-line interface for commitgen tool.
|
||||
"""
|
||||
|
||||
import click
|
||||
from rich.console import Console
|
||||
from rich.prompt import Prompt
|
||||
@@ -18,47 +17,27 @@ import os
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
@click.group()
|
||||
def cli():
|
||||
"""Generate commit messages and changelogs using AI."""
|
||||
pass
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.option(
|
||||
"--files",
|
||||
"-f",
|
||||
multiple=True,
|
||||
help="Specific files to generate commit messages for",
|
||||
)
|
||||
@click.option("--repo", "-r", help="Repository path")
|
||||
@click.option("--commit", "-c", is_flag=True, help="Automatically commit changes")
|
||||
@click.option("--push", "-p", is_flag=True, help="Push changes after commit")
|
||||
@click.option(
|
||||
"--conventional/--no-conventional",
|
||||
default=True,
|
||||
help="Use conventional commit format",
|
||||
)
|
||||
@click.option("--no-stage", is_flag=True, help="Skip automatic staging of changes")
|
||||
@click.option("--sign", is_flag=True, help="Sign commits with GPG")
|
||||
@click.option("--temperature", type=float, help="AI temperature (0.0-1.0)")
|
||||
def generate(
|
||||
files: tuple,
|
||||
repo: str,
|
||||
commit: bool,
|
||||
push: bool,
|
||||
conventional: bool,
|
||||
no_stage: bool,
|
||||
sign: bool,
|
||||
temperature: float,
|
||||
):
|
||||
@click.option('--files', '-f', multiple=True, help='Specific files to generate commit messages for')
|
||||
@click.option('--repo', '-r', help='Repository path')
|
||||
@click.option('--commit', '-c', is_flag=True, help='Automatically commit changes')
|
||||
@click.option('--push', '-p', is_flag=True, help='Push changes after commit')
|
||||
@click.option('--conventional/--no-conventional', default=True, help='Use conventional commit format')
|
||||
@click.option('--no-stage', is_flag=True, help='Skip automatic staging of changes')
|
||||
@click.option('--sign', is_flag=True, help='Sign commits with GPG')
|
||||
@click.option('--temperature', type=float, help='AI temperature (0.0-1.0)')
|
||||
def generate(files: tuple, repo: str, commit: bool, push: bool, conventional: bool, no_stage: bool, sign: bool, temperature: float):
|
||||
"""Generate commit messages for staged changes"""
|
||||
try:
|
||||
with Progress(
|
||||
SpinnerColumn(),
|
||||
TextColumn("[progress.description]{task.description}"),
|
||||
console=console,
|
||||
console=console
|
||||
) as progress:
|
||||
# Initialize services
|
||||
config = Config()
|
||||
@@ -82,9 +61,7 @@ def generate(
|
||||
|
||||
# Generate commit message
|
||||
task = progress.add_task("Generating commit message...", total=None)
|
||||
commit_message = ai_service.generate_commit_message(
|
||||
staged_changes, temperature
|
||||
)
|
||||
commit_message = ai_service.generate_commit_message(staged_changes, temperature)
|
||||
progress.update(task, completed=True)
|
||||
|
||||
# Show preview and confirm
|
||||
@@ -103,46 +80,31 @@ def generate(
|
||||
|
||||
console.print("[green]Changes committed successfully![/green]")
|
||||
else:
|
||||
console.print(
|
||||
"[yellow]Use --commit to apply the commit message.[/yellow]"
|
||||
)
|
||||
console.print("[yellow]Use --commit to apply the commit message.[/yellow]")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error: {str(e)}[/red]")
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
@cli.group()
|
||||
def changelog():
|
||||
"""Generate and manage changelogs"""
|
||||
pass
|
||||
|
||||
|
||||
@changelog.command(name="generate")
|
||||
@click.option(
|
||||
"--version", "-v", required=True, help="Version number for the changelog entry"
|
||||
)
|
||||
@click.option("--from-tag", "-t", help="Generate changelog from this tag")
|
||||
@click.option("--days", "-d", type=int, help="Generate changelog from the last N days")
|
||||
@click.option(
|
||||
"--commits", "-n", type=int, help="Generate changelog from the last N commits"
|
||||
)
|
||||
@click.option("--output", "-o", help="Output file path (default: CHANGELOG.md)")
|
||||
@click.option("--temperature", type=float, help="AI temperature (0.0-1.0)")
|
||||
def generate_log(
|
||||
version: str,
|
||||
from_tag: str,
|
||||
days: int,
|
||||
commits: int,
|
||||
output: str,
|
||||
temperature: float,
|
||||
):
|
||||
@changelog.command(name='generate')
|
||||
@click.option('--version', '-v', required=True, help='Version number for the changelog entry')
|
||||
@click.option('--from-tag', '-t', help='Generate changelog from this tag')
|
||||
@click.option('--days', '-d', type=int, help='Generate changelog from the last N days')
|
||||
@click.option('--commits', '-n', type=int, help='Generate changelog from the last N commits')
|
||||
@click.option('--output', '-o', help='Output file path (default: CHANGELOG.md)')
|
||||
@click.option('--temperature', type=float, help='AI temperature (0.0-1.0)')
|
||||
def generate_log(version: str, from_tag: str, days: int, commits: int, output: str, temperature: float):
|
||||
"""Generate a changelog from git history"""
|
||||
try:
|
||||
with Progress(
|
||||
SpinnerColumn(),
|
||||
TextColumn("[progress.description]{task.description}"),
|
||||
console=console,
|
||||
console=console
|
||||
) as progress:
|
||||
# Initialize services
|
||||
config = Config()
|
||||
@@ -150,7 +112,7 @@ def generate_log(
|
||||
changelog_gen = ChangelogGenerator(config, git_service)
|
||||
|
||||
task = progress.add_task("Analyzing commit history...", total=None)
|
||||
|
||||
|
||||
# Determine which method to use for getting commits
|
||||
if from_tag:
|
||||
entries = changelog_gen.get_commits(from_tag)
|
||||
@@ -161,13 +123,11 @@ def generate_log(
|
||||
else:
|
||||
# Default: get commits from the last 7 days
|
||||
entries = changelog_gen.get_commits_since_date(7)
|
||||
|
||||
|
||||
progress.update(task, completed=True)
|
||||
|
||||
if not entries:
|
||||
console.print(
|
||||
"[yellow]No conventional commits found in the specified range.[/yellow]"
|
||||
)
|
||||
console.print("[yellow]No conventional commits found in the specified range.[/yellow]")
|
||||
return
|
||||
|
||||
task = progress.add_task("Generating changelog...", total=None)
|
||||
@@ -175,24 +135,20 @@ def generate_log(
|
||||
changelog_gen.update_changelog_file(version, entries, output_path)
|
||||
progress.update(task, completed=True)
|
||||
|
||||
console.print(
|
||||
f"[green]Changelog generated successfully at {output_path}![/green]"
|
||||
)
|
||||
console.print(f"[green]Changelog generated successfully at {output_path}![/green]")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error: {str(e)}[/red]")
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
@cli.group()
|
||||
def config():
|
||||
"""Manage configuration"""
|
||||
pass
|
||||
|
||||
|
||||
@config.command()
|
||||
@click.option("--key", "-k", required=True, help="Configuration key")
|
||||
@click.option("--value", "-v", required=True, help="Configuration value")
|
||||
@click.option('--key', '-k', required=True, help='Configuration key')
|
||||
@click.option('--value', '-v', required=True, help='Configuration value')
|
||||
def set(key: str, value: str):
|
||||
"""Set configuration value"""
|
||||
try:
|
||||
@@ -203,66 +159,62 @@ def set(key: str, value: str):
|
||||
console.print(f"[red]Error: {str(e)}[/red]")
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
@config.command()
|
||||
def show():
|
||||
"""Show current configuration"""
|
||||
try:
|
||||
config = Config()
|
||||
console.print(
|
||||
Panel(
|
||||
"\n".join(f"{k}: {v}" for k, v in config.get_all().items()),
|
||||
title="[blue]Current Configuration[/blue]",
|
||||
border_style="blue",
|
||||
)
|
||||
)
|
||||
console.print(Panel(
|
||||
"\n".join(f"{k}: {v}" for k, v in config.get_all().items()),
|
||||
title="[blue]Current Configuration[/blue]",
|
||||
border_style="blue"
|
||||
))
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error: {str(e)}[/red]")
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
@cli.command()
|
||||
def status():
|
||||
"""Show the current state of the repository."""
|
||||
try:
|
||||
# Initialize services
|
||||
git = GitService(os.getcwd())
|
||||
|
||||
|
||||
# Get repository info
|
||||
repo_name = git.get_repo_name()
|
||||
current_branch = git.get_current_branch()
|
||||
|
||||
|
||||
# Get changes
|
||||
staged_files = git.get_staged_files()
|
||||
unstaged_files = git.get_unstaged_files()
|
||||
|
||||
|
||||
# Create status table
|
||||
table = Table(title=f"Repository Status: {repo_name} ({current_branch})")
|
||||
table.add_column("Status", style="cyan")
|
||||
table.add_column("Files", style="green")
|
||||
|
||||
|
||||
# Add staged changes
|
||||
if staged_files:
|
||||
staged_content = "\n".join(f" • {file}" for file in staged_files)
|
||||
table.add_row("Staged", staged_content)
|
||||
else:
|
||||
table.add_row("Staged", "No staged changes")
|
||||
|
||||
|
||||
# Add unstaged changes
|
||||
if unstaged_files:
|
||||
unstaged_content = "\n".join(f" • {file}" for file in unstaged_files)
|
||||
table.add_row("Unstaged", unstaged_content)
|
||||
else:
|
||||
table.add_row("Unstaged", "No unstaged changes")
|
||||
|
||||
|
||||
# Show status
|
||||
console.print("\n[bold]Repository Status:[/bold]")
|
||||
console.print(table)
|
||||
|
||||
|
||||
# Show diffs if there are changes
|
||||
if staged_files or unstaged_files:
|
||||
console.print("\n[bold]Changes:[/bold]")
|
||||
|
||||
|
||||
# Show staged diffs
|
||||
if staged_files:
|
||||
console.print("\n[cyan]Staged Changes:[/cyan]")
|
||||
@@ -270,7 +222,7 @@ def status():
|
||||
diff = git.get_diff(file, staged=True)
|
||||
if diff:
|
||||
console.print(Syntax(diff, "diff", theme="monokai"))
|
||||
|
||||
|
||||
# Show unstaged diffs
|
||||
if unstaged_files:
|
||||
console.print("\n[yellow]Unstaged Changes:[/yellow]")
|
||||
@@ -278,16 +230,15 @@ def status():
|
||||
diff = git.get_diff(file, staged=False)
|
||||
if diff:
|
||||
console.print(Syntax(diff, "diff", theme="monokai"))
|
||||
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error: {str(e)}[/red]")
|
||||
raise click.Abort()
|
||||
|
||||
|
||||
# Add command aliases
|
||||
cli.add_command(generate, name="g")
|
||||
cli.add_command(config, name="c")
|
||||
cli.add_command(changelog, name="log")
|
||||
cli.add_command(generate, name='g')
|
||||
cli.add_command(config, name='c')
|
||||
cli.add_command(changelog, name='log')
|
||||
|
||||
if __name__ == "__main__":
|
||||
cli()
|
||||
cli()
|
||||
@@ -34,6 +34,11 @@ class CommitGenGitService(GitService):
|
||||
"""Commit changes with the given message."""
|
||||
if not self.commit(message, sign=sign):
|
||||
raise Exception("Failed to commit changes")
|
||||
|
||||
def push_changes(self) -> None:
|
||||
"""Push changes to the current branch."""
|
||||
if not self.push():
|
||||
raise Exception("Failed to push changes")
|
||||
|
||||
def get_recent_commits(self, limit: int = 10) -> List[str]:
|
||||
"""Get recent commit messages."""
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
"""License generator module for DevTools."""
|
||||
|
||||
from .generator import generate_license
|
||||
|
||||
__all__ = ["generate_license"]
|
||||
@@ -0,0 +1,249 @@
|
||||
"""License generator CLI."""
|
||||
|
||||
import os
|
||||
import click
|
||||
from rich.console import Console
|
||||
from rich.table import Table
|
||||
from rich.prompt import Confirm, Prompt
|
||||
from rich.panel import Panel
|
||||
from ..shared.config import Config
|
||||
from .generator import license_generator, generate_license
|
||||
import re
|
||||
|
||||
console = Console()
|
||||
|
||||
@click.group()
|
||||
def license():
|
||||
"""License generation commands."""
|
||||
pass
|
||||
|
||||
@license.command()
|
||||
@click.option("--type", "-t", help="Type of license to generate")
|
||||
@click.option("--author", "-a", help="Name of the author/copyright holder")
|
||||
@click.option("--year", "-y", type=int, help="Year for the copyright notice")
|
||||
@click.option("--output", "-o", default="LICENSE", help="Output file path")
|
||||
@click.option("--multi", "-m", is_flag=True, help="Generate multiple licenses")
|
||||
def generate(type: str, author: str, year: int, output: str, multi: bool):
|
||||
"""Generate a license file."""
|
||||
try:
|
||||
# If no author specified, prompt for it
|
||||
if not author:
|
||||
author = click.prompt("Enter author name")
|
||||
|
||||
# Fetch available licenses
|
||||
licenses = license_generator.fetch_available_licenses()
|
||||
if not licenses:
|
||||
console.print("[yellow]No licenses available")
|
||||
return
|
||||
|
||||
# Create selection table
|
||||
table = Table(show_header=True, header_style="bold magenta")
|
||||
table.add_column("No.", style="dim", width=4)
|
||||
table.add_column("License Key", style="dim", width=15)
|
||||
table.add_column("Name", width=30)
|
||||
table.add_column("Permissions", width=40)
|
||||
|
||||
# Add rows with numbers
|
||||
for i, license in enumerate(licenses, 1):
|
||||
table.add_row(
|
||||
str(i),
|
||||
license["key"],
|
||||
license["name"],
|
||||
format_permissions(license.get("permissions", []))
|
||||
)
|
||||
|
||||
# Display table
|
||||
console.print("\nSelect license(s) to generate:")
|
||||
console.print(table)
|
||||
|
||||
if multi:
|
||||
# Get user selection for multiple licenses
|
||||
while True:
|
||||
try:
|
||||
selection = click.prompt("Enter license numbers (e.g., 1,3,5)").split(",")
|
||||
selected_indices = [int(s.strip()) for s in selection if s.strip()]
|
||||
|
||||
# Validate selection
|
||||
if not selected_indices:
|
||||
console.print("[yellow]No licenses selected")
|
||||
continue
|
||||
|
||||
if any(i < 1 or i > len(licenses) for i in selected_indices):
|
||||
console.print("[red]Invalid selection. Please enter valid numbers.")
|
||||
continue
|
||||
|
||||
# Get selected licenses
|
||||
selected_licenses = [licenses[i-1] for i in selected_indices]
|
||||
|
||||
# Show confirmation
|
||||
console.print("\nSelected licenses:")
|
||||
for license in selected_licenses:
|
||||
console.print(f"- {license['name']} ({license['key']})")
|
||||
|
||||
if Confirm.ask("\nGenerate these licenses?"):
|
||||
break
|
||||
else:
|
||||
console.print("\nPlease select licenses again:")
|
||||
continue
|
||||
|
||||
except ValueError:
|
||||
console.print("[red]Invalid input. Please enter numbers separated by commas.")
|
||||
continue
|
||||
else:
|
||||
# Get user selection for single license
|
||||
while True:
|
||||
try:
|
||||
selection = click.prompt("Enter license number")
|
||||
selected_index = int(selection.strip())
|
||||
|
||||
# Validate selection
|
||||
if selected_index < 1 or selected_index > len(licenses):
|
||||
console.print("[red]Invalid selection. Please enter a valid number.")
|
||||
continue
|
||||
|
||||
# Get selected license
|
||||
selected_license = licenses[selected_index - 1]
|
||||
|
||||
# Show confirmation
|
||||
console.print(f"\nSelected license: {selected_license['name']} ({selected_license['key']})")
|
||||
|
||||
if Confirm.ask("\nGenerate this license?"):
|
||||
selected_licenses = [selected_license]
|
||||
break
|
||||
else:
|
||||
console.print("\nPlease select a license again:")
|
||||
continue
|
||||
|
||||
except ValueError:
|
||||
console.print("[red]Invalid input. Please enter a number.")
|
||||
continue
|
||||
|
||||
# Generate each license
|
||||
for license in selected_licenses:
|
||||
license_type = license["key"]
|
||||
# Generate output filename
|
||||
if len(selected_licenses) == 1:
|
||||
output_file = output
|
||||
else:
|
||||
base, ext = os.path.splitext(output)
|
||||
output_file = f"{base}_{license_type}{ext}"
|
||||
|
||||
try:
|
||||
output_path = generate_license(
|
||||
license_type=license_type,
|
||||
author=author,
|
||||
year=year,
|
||||
output_file=output_file
|
||||
)
|
||||
console.print(f"[green]✓[/green] Generated {license_type} license at: {output_path}")
|
||||
except ValueError as e:
|
||||
console.print(f"[red]Error generating {license_type} license:[/red] {str(e)}")
|
||||
continue
|
||||
|
||||
# Create a combined README section
|
||||
if len(selected_licenses) > 1:
|
||||
readme_path = "LICENSE.md"
|
||||
with open(readme_path, "w") as f:
|
||||
f.write("# Licenses\n\n")
|
||||
f.write("This project is licensed under multiple licenses:\n\n")
|
||||
for license in selected_licenses:
|
||||
f.write(f"- [{license['name']}](LICENSE_{license['key']})\n")
|
||||
console.print(f"\n[green]✓[/green] Created license summary at: {readme_path}")
|
||||
|
||||
except ValueError as e:
|
||||
console.print(f"[red]Error:[/red] {str(e)}")
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error:[/red] {str(e)}")
|
||||
|
||||
@license.command()
|
||||
def list():
|
||||
"""List available licenses."""
|
||||
show_licenses()
|
||||
|
||||
def format_permissions(permissions: list) -> str:
|
||||
"""Format permissions list into a readable string."""
|
||||
if not permissions:
|
||||
return "N/A"
|
||||
|
||||
# Map permission keys to readable names
|
||||
permission_map = {
|
||||
"commercial-use": "Commercial Use",
|
||||
"modifications": "Modifications",
|
||||
"distribution": "Distribution",
|
||||
"private-use": "Private Use",
|
||||
"patent-use": "Patent Use"
|
||||
}
|
||||
|
||||
# Convert each permission to its readable name
|
||||
readable_permissions = [
|
||||
permission_map.get(perm, perm.replace("-", " ").title())
|
||||
for perm in permissions
|
||||
]
|
||||
|
||||
return ", ".join(readable_permissions)
|
||||
|
||||
def truncate_description(description: str, max_length: int = 60) -> str:
|
||||
"""Truncate description to a maximum length and add ellipsis if needed."""
|
||||
if not description:
|
||||
return "No description available"
|
||||
|
||||
# Remove HTML tags
|
||||
description = re.sub(r'<[^>]+>', '', description)
|
||||
|
||||
# Remove URLs
|
||||
description = re.sub(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', '', description)
|
||||
|
||||
# Clean up whitespace
|
||||
description = ' '.join(description.split())
|
||||
|
||||
if len(description) <= max_length:
|
||||
return description
|
||||
|
||||
# Find the last space before max_length
|
||||
last_space = description[:max_length].rfind(" ")
|
||||
if last_space == -1:
|
||||
last_space = max_length
|
||||
|
||||
return description[:last_space] + "..."
|
||||
|
||||
def show_licenses():
|
||||
"""Display available licenses in a table."""
|
||||
try:
|
||||
# Load token from config
|
||||
config = Config()
|
||||
github_token = config.get("github_token")
|
||||
if github_token:
|
||||
os.environ["GITHUB_TOKEN"] = github_token
|
||||
|
||||
# Fetch available licenses
|
||||
licenses = license_generator.fetch_available_licenses()
|
||||
|
||||
if not licenses:
|
||||
console.print("[yellow]No licenses available")
|
||||
return
|
||||
|
||||
# Create table
|
||||
table = Table(show_header=True, header_style="bold magenta")
|
||||
table.add_column("License Key", style="dim", width=15)
|
||||
table.add_column("Name", width=30)
|
||||
table.add_column("Permissions", width=40)
|
||||
|
||||
# Add rows
|
||||
for license in licenses:
|
||||
table.add_row(
|
||||
license["key"],
|
||||
license["name"],
|
||||
format_permissions(license.get("permissions", []))
|
||||
)
|
||||
|
||||
# Display table
|
||||
console.print("\nAvailable Licenses:")
|
||||
console.print(table)
|
||||
console.print("\nTo generate a license, use: [bold]devtools license generate[/bold]")
|
||||
console.print("To generate multiple licenses, use: [bold]devtools license generate --multi[/bold]")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error:[/red] {str(e)}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
license()
|
||||
@@ -0,0 +1,130 @@
|
||||
"""License generator module."""
|
||||
|
||||
import os
|
||||
import requests
|
||||
from typing import Dict, List, Optional
|
||||
from ..shared.config import Config
|
||||
|
||||
class LicenseGenerator:
|
||||
"""License generator using GitHub API."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the license generator."""
|
||||
self.api_base_url = "https://api.github.com"
|
||||
self.config = Config()
|
||||
|
||||
def _get_headers(self) -> Dict[str, str]:
|
||||
"""Get headers for API requests."""
|
||||
# Try to get token from config first
|
||||
token = self.config.get("github_token")
|
||||
|
||||
# Then try environment variable
|
||||
if not token:
|
||||
token = os.environ.get("GITHUB_TOKEN")
|
||||
|
||||
headers = {
|
||||
"Accept": "application/vnd.github.v3+json",
|
||||
"User-Agent": "DevTools/1.0"
|
||||
}
|
||||
|
||||
if token:
|
||||
headers["Authorization"] = f"Bearer {token}"
|
||||
|
||||
return headers
|
||||
|
||||
def fetch_available_licenses(self) -> List[Dict]:
|
||||
"""Fetch available licenses from GitHub API."""
|
||||
try:
|
||||
# Get headers with token
|
||||
headers = self._get_headers()
|
||||
|
||||
# First check rate limit
|
||||
rate_limit_response = requests.get(
|
||||
f"{self.api_base_url}/rate_limit",
|
||||
headers=headers
|
||||
)
|
||||
if rate_limit_response.status_code == 200:
|
||||
rate_limit_data = rate_limit_response.json()
|
||||
|
||||
# Check if we have remaining requests
|
||||
remaining = rate_limit_data.get("resources", {}).get("core", {}).get("remaining", 0)
|
||||
if remaining == 0:
|
||||
raise ValueError("GitHub API rate limit exceeded. Please set a GitHub token using:\ndevtools config set GITHUB_TOKEN YOUR_GITHUB_TOKEN")
|
||||
|
||||
# Fetch list of licenses
|
||||
response = requests.get(
|
||||
f"{self.api_base_url}/licenses",
|
||||
headers=headers
|
||||
)
|
||||
|
||||
if response.status_code == 403:
|
||||
raise ValueError("GitHub API rate limit exceeded. Please set a GitHub token using:\ndevtools config set GITHUB_TOKEN YOUR_GITHUB_TOKEN")
|
||||
|
||||
response.raise_for_status()
|
||||
licenses = response.json()
|
||||
|
||||
# Fetch detailed info for each license
|
||||
detailed_licenses = []
|
||||
for license in licenses:
|
||||
try:
|
||||
license_response = requests.get(
|
||||
f"{self.api_base_url}/licenses/{license['key']}",
|
||||
headers=headers
|
||||
)
|
||||
if license_response.status_code == 200:
|
||||
license_data = license_response.json()
|
||||
detailed_licenses.append({
|
||||
"key": license_data["key"],
|
||||
"name": license_data["name"],
|
||||
"description": license_data.get("description", "No description available"),
|
||||
"permissions": license_data.get("permissions", []),
|
||||
"conditions": license_data.get("conditions", []),
|
||||
"limitations": license_data.get("limitations", [])
|
||||
})
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
return detailed_licenses
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise ValueError(f"Error fetching licenses: {str(e)}")
|
||||
|
||||
def get_license_template(self, license_type: str) -> str:
|
||||
"""Get license template from GitHub API."""
|
||||
try:
|
||||
response = requests.get(
|
||||
f"{self.api_base_url}/licenses/{license_type}",
|
||||
headers=self._get_headers()
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response.json()["body"]
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise ValueError(f"Error fetching license template: {str(e)}")
|
||||
|
||||
def generate_license(self, license_type: str, author: str, year: Optional[int] = None) -> str:
|
||||
"""Generate license content with author and year."""
|
||||
template = self.get_license_template(license_type)
|
||||
|
||||
# Replace placeholders
|
||||
content = template.replace("[year]", str(year) if year else "")
|
||||
content = content.replace("[fullname]", author)
|
||||
|
||||
return content
|
||||
|
||||
# Create a singleton instance
|
||||
license_generator = LicenseGenerator()
|
||||
|
||||
def generate_license(license_type: str, author: str, year: Optional[int] = None, output_file: str = "LICENSE") -> str:
|
||||
"""Generate a license file."""
|
||||
try:
|
||||
# Generate license content
|
||||
content = license_generator.generate_license(license_type, author, year)
|
||||
|
||||
# Write to file
|
||||
with open(output_file, "w") as f:
|
||||
f.write(content)
|
||||
|
||||
return output_file
|
||||
|
||||
except Exception as e:
|
||||
raise ValueError(f"Error generating license: {str(e)}")
|
||||
+174
-237
@@ -1,100 +1,84 @@
|
||||
"""
|
||||
Shared AI service for devtools.
|
||||
"""
|
||||
|
||||
import requests
|
||||
import os
|
||||
from typing import List, Dict, Optional, Any
|
||||
from abc import ABC, abstractmethod
|
||||
from .config import BaseConfig
|
||||
|
||||
|
||||
class AIProvider(ABC):
|
||||
"""Base class for AI providers."""
|
||||
|
||||
|
||||
@abstractmethod
|
||||
def setup(self, config: BaseConfig) -> None:
|
||||
"""Set up the provider with configuration."""
|
||||
pass
|
||||
|
||||
|
||||
@abstractmethod
|
||||
def generate_completion(
|
||||
self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95,
|
||||
) -> str:
|
||||
def generate_completion(self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95) -> str:
|
||||
"""Generate a completion from the AI model."""
|
||||
pass
|
||||
|
||||
|
||||
class OpenRouterProvider(AIProvider):
|
||||
"""OpenRouter AI provider implementation."""
|
||||
|
||||
|
||||
def setup(self, config: BaseConfig) -> None:
|
||||
self.config = config
|
||||
self.api_key = config.get_env_or_config("OPENROUTER_API_KEY")
|
||||
if not self.api_key:
|
||||
raise ValueError(
|
||||
"OpenRouter API key not found. Please set it in config or environment."
|
||||
)
|
||||
|
||||
raise ValueError("OpenRouter API key not found. Please set it in config or environment.")
|
||||
|
||||
self.model = config.get("model", "mistralai/mixtral-8x7b-instruct")
|
||||
self.api_url = "https://openrouter.ai/api/v1/chat/completions"
|
||||
self.headers = {
|
||||
"Authorization": f"Bearer {self.api_key}",
|
||||
"HTTP-Referer": "https://github.com/S4NKALP/DevTools",
|
||||
"X-Title": "DevTools",
|
||||
"Content-Type": "application/json",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
def _create_prompt(
|
||||
self, system_prompt: str, user_prompt: str
|
||||
) -> List[Dict[str, str]]:
|
||||
def _create_prompt(self, system_prompt: str, user_prompt: str) -> List[Dict[str, str]]:
|
||||
return [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_prompt},
|
||||
{"role": "user", "content": user_prompt}
|
||||
]
|
||||
|
||||
def generate_completion(
|
||||
self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95,
|
||||
) -> str:
|
||||
def generate_completion(self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95) -> str:
|
||||
messages = self._create_prompt(system_prompt, user_prompt)
|
||||
|
||||
|
||||
try:
|
||||
temp = (
|
||||
temperature
|
||||
if temperature is not None
|
||||
else float(self.config.get("temperature", 0.7))
|
||||
)
|
||||
tokens = (
|
||||
max_tokens
|
||||
if max_tokens is not None
|
||||
else int(self.config.get("max_tokens", 150))
|
||||
)
|
||||
|
||||
temp = temperature if temperature is not None else float(self.config.get("temperature", 0.7))
|
||||
tokens = max_tokens if max_tokens is not None else int(self.config.get("max_tokens", 150))
|
||||
|
||||
payload = {
|
||||
"model": self.model,
|
||||
"messages": messages,
|
||||
"temperature": temp,
|
||||
"max_tokens": tokens,
|
||||
"top_p": float(top_p),
|
||||
"stream": False,
|
||||
"stream": False
|
||||
}
|
||||
|
||||
|
||||
payload = {k: v for k, v in payload.items() if v is not None}
|
||||
|
||||
|
||||
response = requests.post(
|
||||
self.api_url, headers=self.headers, json=payload, timeout=30
|
||||
self.api_url,
|
||||
headers=self.headers,
|
||||
json=payload,
|
||||
timeout=30
|
||||
)
|
||||
|
||||
|
||||
if response.status_code != 200:
|
||||
error_msg = response.text
|
||||
try:
|
||||
@@ -104,80 +88,68 @@ class OpenRouterProvider(AIProvider):
|
||||
except:
|
||||
pass
|
||||
raise Exception(f"OpenRouter API error: {error_msg}")
|
||||
|
||||
|
||||
result = response.json()
|
||||
|
||||
|
||||
if "choices" in result and len(result["choices"]) > 0:
|
||||
return result["choices"][0]["message"]["content"].strip()
|
||||
|
||||
|
||||
return ""
|
||||
|
||||
|
||||
except Exception as e:
|
||||
raise Exception(f"AI generation failed: {str(e)}")
|
||||
|
||||
|
||||
class OpenAIProvider(AIProvider):
|
||||
"""OpenAI provider implementation."""
|
||||
|
||||
|
||||
def setup(self, config: BaseConfig) -> None:
|
||||
self.config = config
|
||||
self.api_key = config.get_env_or_config("OPENAI_API_KEY")
|
||||
if not self.api_key:
|
||||
raise ValueError(
|
||||
"OpenAI API key not found. Please set it in config or environment."
|
||||
)
|
||||
|
||||
raise ValueError("OpenAI API key not found. Please set it in config or environment.")
|
||||
|
||||
self.model = config.get("model", "gpt-4-turbo-preview")
|
||||
self.api_url = "https://api.openai.com/v1/chat/completions"
|
||||
self.headers = {
|
||||
"Authorization": f"Bearer {self.api_key}",
|
||||
"Content-Type": "application/json",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
def _create_prompt(
|
||||
self, system_prompt: str, user_prompt: str
|
||||
) -> List[Dict[str, str]]:
|
||||
def _create_prompt(self, system_prompt: str, user_prompt: str) -> List[Dict[str, str]]:
|
||||
return [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_prompt},
|
||||
{"role": "user", "content": user_prompt}
|
||||
]
|
||||
|
||||
def generate_completion(
|
||||
self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95,
|
||||
) -> str:
|
||||
def generate_completion(self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95) -> str:
|
||||
messages = self._create_prompt(system_prompt, user_prompt)
|
||||
|
||||
|
||||
try:
|
||||
temp = (
|
||||
temperature
|
||||
if temperature is not None
|
||||
else float(self.config.get("temperature", 0.7))
|
||||
)
|
||||
tokens = (
|
||||
max_tokens
|
||||
if max_tokens is not None
|
||||
else int(self.config.get("max_tokens", 150))
|
||||
)
|
||||
|
||||
temp = temperature if temperature is not None else float(self.config.get("temperature", 0.7))
|
||||
tokens = max_tokens if max_tokens is not None else int(self.config.get("max_tokens", 150))
|
||||
|
||||
payload = {
|
||||
"model": self.model,
|
||||
"messages": messages,
|
||||
"temperature": temp,
|
||||
"max_tokens": tokens,
|
||||
"top_p": float(top_p),
|
||||
"top_p": float(top_p)
|
||||
}
|
||||
|
||||
|
||||
payload = {k: v for k, v in payload.items() if v is not None}
|
||||
|
||||
|
||||
response = requests.post(
|
||||
self.api_url, headers=self.headers, json=payload, timeout=30
|
||||
self.api_url,
|
||||
headers=self.headers,
|
||||
json=payload,
|
||||
timeout=30
|
||||
)
|
||||
|
||||
|
||||
if response.status_code != 200:
|
||||
error_msg = response.text
|
||||
try:
|
||||
@@ -187,74 +159,63 @@ class OpenAIProvider(AIProvider):
|
||||
except:
|
||||
pass
|
||||
raise Exception(f"OpenAI API error: {error_msg}")
|
||||
|
||||
|
||||
result = response.json()
|
||||
|
||||
|
||||
if "choices" in result and len(result["choices"]) > 0:
|
||||
return result["choices"][0]["message"]["content"].strip()
|
||||
|
||||
|
||||
return ""
|
||||
|
||||
|
||||
except Exception as e:
|
||||
raise Exception(f"AI generation failed: {str(e)}")
|
||||
|
||||
|
||||
class GeminiProvider(AIProvider):
|
||||
"""Google Gemini provider implementation."""
|
||||
|
||||
|
||||
def setup(self, config: BaseConfig) -> None:
|
||||
self.config = config
|
||||
self.api_key = config.get_env_or_config("GOOGLE_API_KEY")
|
||||
if not self.api_key:
|
||||
raise ValueError(
|
||||
"Google API key not found. Please set it in config or environment."
|
||||
)
|
||||
|
||||
raise ValueError("Google API key not found. Please set it in config or environment.")
|
||||
|
||||
self.model = config.get("model", "gemini-pro")
|
||||
self.api_url = f"https://generativelanguage.googleapis.com/v1beta/models/{
|
||||
self.model
|
||||
}:generateContent"
|
||||
self.headers = {"Content-Type": "application/json"}
|
||||
self.api_url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:generateContent"
|
||||
self.headers = {
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
def generate_completion(
|
||||
self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95,
|
||||
) -> str:
|
||||
def generate_completion(self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95) -> str:
|
||||
try:
|
||||
temp = (
|
||||
temperature
|
||||
if temperature is not None
|
||||
else float(self.config.get("temperature", 0.7))
|
||||
)
|
||||
tokens = (
|
||||
max_tokens
|
||||
if max_tokens is not None
|
||||
else int(self.config.get("max_tokens", 150))
|
||||
)
|
||||
|
||||
temp = temperature if temperature is not None else float(self.config.get("temperature", 0.7))
|
||||
tokens = max_tokens if max_tokens is not None else int(self.config.get("max_tokens", 150))
|
||||
|
||||
# Combine system and user prompts for Gemini
|
||||
full_prompt = f"{system_prompt}\n\n{user_prompt}"
|
||||
|
||||
|
||||
payload = {
|
||||
"contents": [{"parts": [{"text": full_prompt}]}],
|
||||
"contents": [{
|
||||
"parts": [{"text": full_prompt}]
|
||||
}],
|
||||
"generationConfig": {
|
||||
"temperature": temp,
|
||||
"maxOutputTokens": tokens,
|
||||
"topP": float(top_p),
|
||||
},
|
||||
"topP": float(top_p)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
response = requests.post(
|
||||
f"{self.api_url}?key={self.api_key}",
|
||||
headers=self.headers,
|
||||
json=payload,
|
||||
timeout=30,
|
||||
timeout=30
|
||||
)
|
||||
|
||||
|
||||
if response.status_code != 200:
|
||||
error_msg = response.text
|
||||
try:
|
||||
@@ -264,81 +225,69 @@ class GeminiProvider(AIProvider):
|
||||
except:
|
||||
pass
|
||||
raise Exception(f"Gemini API error: {error_msg}")
|
||||
|
||||
|
||||
result = response.json()
|
||||
|
||||
|
||||
if "candidates" in result and len(result["candidates"]) > 0:
|
||||
return result["candidates"][0]["content"]["parts"][0]["text"].strip()
|
||||
|
||||
|
||||
return ""
|
||||
|
||||
|
||||
except Exception as e:
|
||||
raise Exception(f"AI generation failed: {str(e)}")
|
||||
|
||||
|
||||
class ClaudeProvider(AIProvider):
|
||||
"""Anthropic Claude provider implementation."""
|
||||
|
||||
|
||||
def setup(self, config: BaseConfig) -> None:
|
||||
self.config = config
|
||||
self.api_key = config.get_env_or_config("ANTHROPIC_API_KEY")
|
||||
if not self.api_key:
|
||||
raise ValueError(
|
||||
"Anthropic API key not found. Please set it in config or environment."
|
||||
)
|
||||
|
||||
raise ValueError("Anthropic API key not found. Please set it in config or environment.")
|
||||
|
||||
self.model = config.get("model", "claude-3-opus-20240229")
|
||||
self.api_url = "https://api.anthropic.com/v1/messages"
|
||||
self.headers = {
|
||||
"x-api-key": self.api_key,
|
||||
"anthropic-version": "2023-06-01",
|
||||
"content-type": "application/json",
|
||||
"content-type": "application/json"
|
||||
}
|
||||
|
||||
def _create_prompt(
|
||||
self, system_prompt: str, user_prompt: str
|
||||
) -> List[Dict[str, str]]:
|
||||
def _create_prompt(self, system_prompt: str, user_prompt: str) -> List[Dict[str, str]]:
|
||||
return [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_prompt},
|
||||
{"role": "user", "content": user_prompt}
|
||||
]
|
||||
|
||||
def generate_completion(
|
||||
self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95,
|
||||
) -> str:
|
||||
def generate_completion(self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95) -> str:
|
||||
messages = self._create_prompt(system_prompt, user_prompt)
|
||||
|
||||
|
||||
try:
|
||||
temp = (
|
||||
temperature
|
||||
if temperature is not None
|
||||
else float(self.config.get("temperature", 0.7))
|
||||
)
|
||||
tokens = (
|
||||
max_tokens
|
||||
if max_tokens is not None
|
||||
else int(self.config.get("max_tokens", 150))
|
||||
)
|
||||
|
||||
temp = temperature if temperature is not None else float(self.config.get("temperature", 0.7))
|
||||
tokens = max_tokens if max_tokens is not None else int(self.config.get("max_tokens", 150))
|
||||
|
||||
payload = {
|
||||
"model": self.model,
|
||||
"messages": messages,
|
||||
"temperature": temp,
|
||||
"max_tokens": tokens,
|
||||
"top_p": float(top_p),
|
||||
"top_p": float(top_p)
|
||||
}
|
||||
|
||||
|
||||
payload = {k: v for k, v in payload.items() if v is not None}
|
||||
|
||||
|
||||
response = requests.post(
|
||||
self.api_url, headers=self.headers, json=payload, timeout=30
|
||||
self.api_url,
|
||||
headers=self.headers,
|
||||
json=payload,
|
||||
timeout=30
|
||||
)
|
||||
|
||||
|
||||
if response.status_code != 200:
|
||||
error_msg = response.text
|
||||
try:
|
||||
@@ -348,76 +297,66 @@ class ClaudeProvider(AIProvider):
|
||||
except:
|
||||
pass
|
||||
raise Exception(f"Claude API error: {error_msg}")
|
||||
|
||||
|
||||
result = response.json()
|
||||
|
||||
|
||||
if "content" in result and len(result["content"]) > 0:
|
||||
return result["content"][0]["text"].strip()
|
||||
|
||||
|
||||
return ""
|
||||
|
||||
|
||||
except Exception as e:
|
||||
raise Exception(f"AI generation failed: {str(e)}")
|
||||
|
||||
|
||||
class HuggingFaceProvider(AIProvider):
|
||||
"""Hugging Face provider implementation."""
|
||||
|
||||
|
||||
def setup(self, config: BaseConfig) -> None:
|
||||
self.config = config
|
||||
self.api_key = config.get_env_or_config("HUGGINGFACE_API_KEY")
|
||||
if not self.api_key:
|
||||
raise ValueError(
|
||||
"Hugging Face API key not found. Please set it in config or environment."
|
||||
)
|
||||
|
||||
raise ValueError("Hugging Face API key not found. Please set it in config or environment.")
|
||||
|
||||
self.model = config.get("model", "mistralai/Mixtral-8x7B-Instruct-v0.1")
|
||||
self.api_url = f"https://api-inference.huggingface.co/models/{self.model}"
|
||||
self.headers = {
|
||||
"Authorization": f"Bearer {self.api_key}",
|
||||
"Content-Type": "application/json",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
def _create_prompt(self, system_prompt: str, user_prompt: str) -> str:
|
||||
# Format prompt according to model's requirements
|
||||
return f"<s>[INST] {system_prompt}\n\n{user_prompt} [/INST]"
|
||||
|
||||
def generate_completion(
|
||||
self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95,
|
||||
) -> str:
|
||||
def generate_completion(self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95) -> str:
|
||||
try:
|
||||
temp = (
|
||||
temperature
|
||||
if temperature is not None
|
||||
else float(self.config.get("temperature", 0.7))
|
||||
)
|
||||
tokens = (
|
||||
max_tokens
|
||||
if max_tokens is not None
|
||||
else int(self.config.get("max_tokens", 150))
|
||||
)
|
||||
|
||||
temp = temperature if temperature is not None else float(self.config.get("temperature", 0.7))
|
||||
tokens = max_tokens if max_tokens is not None else int(self.config.get("max_tokens", 150))
|
||||
|
||||
prompt = self._create_prompt(system_prompt, user_prompt)
|
||||
|
||||
|
||||
payload = {
|
||||
"inputs": prompt,
|
||||
"parameters": {
|
||||
"temperature": temp,
|
||||
"max_new_tokens": tokens,
|
||||
"top_p": float(top_p),
|
||||
"return_full_text": False,
|
||||
},
|
||||
"return_full_text": False
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
response = requests.post(
|
||||
self.api_url, headers=self.headers, json=payload, timeout=30
|
||||
self.api_url,
|
||||
headers=self.headers,
|
||||
json=payload,
|
||||
timeout=30
|
||||
)
|
||||
|
||||
|
||||
if response.status_code != 200:
|
||||
error_msg = response.text
|
||||
try:
|
||||
@@ -427,21 +366,20 @@ class HuggingFaceProvider(AIProvider):
|
||||
except:
|
||||
pass
|
||||
raise Exception(f"Hugging Face API error: {error_msg}")
|
||||
|
||||
|
||||
result = response.json()
|
||||
|
||||
|
||||
if isinstance(result, list) and len(result) > 0:
|
||||
return result[0]["generated_text"].strip()
|
||||
|
||||
|
||||
return ""
|
||||
|
||||
|
||||
except Exception as e:
|
||||
raise Exception(f"AI generation failed: {str(e)}")
|
||||
|
||||
|
||||
class AIService:
|
||||
"""Base AI service that can be extended by specific tools."""
|
||||
|
||||
|
||||
def __init__(self, config: BaseConfig):
|
||||
"""Initialize AI service with configuration."""
|
||||
self.config = config
|
||||
@@ -451,49 +389,48 @@ class AIService:
|
||||
def _get_provider(self) -> AIProvider:
|
||||
"""Get the appropriate AI provider based on configuration."""
|
||||
provider = self.config.get("provider", "openrouter").lower()
|
||||
|
||||
|
||||
providers = {
|
||||
"openrouter": OpenRouterProvider,
|
||||
"openai": OpenAIProvider,
|
||||
"gemini": GeminiProvider,
|
||||
"claude": ClaudeProvider,
|
||||
"huggingface": HuggingFaceProvider,
|
||||
"huggingface": HuggingFaceProvider
|
||||
}
|
||||
|
||||
|
||||
if provider not in providers:
|
||||
raise ValueError(
|
||||
f"Unsupported AI provider: {provider}. Supported providers: {
|
||||
', '.join(providers.keys())
|
||||
}"
|
||||
)
|
||||
|
||||
raise ValueError(f"Unsupported AI provider: {provider}. Supported providers: {', '.join(providers.keys())}")
|
||||
|
||||
return providers[provider]()
|
||||
|
||||
def generate_completion(
|
||||
self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95,
|
||||
) -> str:
|
||||
def generate_completion(self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
top_p: float = 0.95) -> str:
|
||||
"""Generate a completion from the AI model."""
|
||||
return self.provider.generate_completion(
|
||||
system_prompt, user_prompt, temperature, max_tokens, top_p
|
||||
system_prompt,
|
||||
user_prompt,
|
||||
temperature,
|
||||
max_tokens,
|
||||
top_p
|
||||
)
|
||||
|
||||
def generate_batch_completions(
|
||||
self,
|
||||
system_prompt: str,
|
||||
prompts: List[str],
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None,
|
||||
) -> List[str]:
|
||||
def generate_batch_completions(self,
|
||||
system_prompt: str,
|
||||
prompts: List[str],
|
||||
temperature: Optional[float] = None,
|
||||
max_tokens: Optional[int] = None) -> List[str]:
|
||||
"""Generate completions for multiple prompts."""
|
||||
completions = []
|
||||
for prompt in prompts:
|
||||
completion = self.generate_completion(
|
||||
system_prompt, prompt, temperature, max_tokens
|
||||
system_prompt,
|
||||
prompt,
|
||||
temperature,
|
||||
max_tokens
|
||||
)
|
||||
completions.append(completion)
|
||||
return completions
|
||||
return completions
|
||||
+44
-48
@@ -1,21 +1,19 @@
|
||||
"""
|
||||
Shared Git service for devtools.
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Optional, Tuple
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
|
||||
class GitService:
|
||||
"""Base Git service that can be extended by specific tools."""
|
||||
|
||||
|
||||
def __init__(self, repo_path: str):
|
||||
"""Initialize GitService with repository path."""
|
||||
self.repo_path = repo_path
|
||||
if not os.path.exists(os.path.join(repo_path, ".git")):
|
||||
if not os.path.exists(os.path.join(repo_path, '.git')):
|
||||
raise ValueError(f"Not a git repository: {repo_path}")
|
||||
|
||||
def _is_git_repo(self) -> bool:
|
||||
@@ -25,7 +23,7 @@ class GitService:
|
||||
["git", "rev-parse", "--is-inside-work-tree"],
|
||||
cwd=self.repo_path,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
capture_output=True
|
||||
)
|
||||
return True
|
||||
except subprocess.CalledProcessError:
|
||||
@@ -39,7 +37,7 @@ class GitService:
|
||||
cwd=self.repo_path,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
text=True
|
||||
)
|
||||
return result.stdout.strip(), result.stderr.strip(), result.returncode
|
||||
except subprocess.CalledProcessError as e:
|
||||
@@ -52,9 +50,7 @@ class GitService:
|
||||
|
||||
def get_unstaged_files(self) -> List[str]:
|
||||
"""Get list of unstaged files."""
|
||||
stdout, _, _ = self._run_git_command(
|
||||
["ls-files", "--modified", "--others", "--exclude-standard"]
|
||||
)
|
||||
stdout, _, _ = self._run_git_command(["ls-files", "--modified", "--others", "--exclude-standard"])
|
||||
return stdout.splitlines() if stdout else []
|
||||
|
||||
def get_diff(self, file_path: str, staged: bool = True) -> str:
|
||||
@@ -104,36 +100,36 @@ class GitService:
|
||||
if files:
|
||||
for file in files:
|
||||
result = subprocess.run(
|
||||
["git", "diff", "--staged", "--", file],
|
||||
['git', 'diff', '--staged', '--', file],
|
||||
cwd=self.repo_path,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
text=True
|
||||
)
|
||||
if result.stdout:
|
||||
diffs[file] = result.stdout
|
||||
else:
|
||||
result = subprocess.run(
|
||||
["git", "diff", "--staged"],
|
||||
['git', 'diff', '--staged'],
|
||||
cwd=self.repo_path,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
text=True
|
||||
)
|
||||
if result.stdout:
|
||||
# Parse the diff to get individual file changes
|
||||
current_file = None
|
||||
current_diff = []
|
||||
|
||||
for line in result.stdout.split("\n"):
|
||||
if line.startswith("diff --git"):
|
||||
for line in result.stdout.split('\n'):
|
||||
if line.startswith('diff --git'):
|
||||
if current_file and current_diff:
|
||||
diffs[current_file] = "\n".join(current_diff)
|
||||
current_file = line.split("b/")[-1].strip()
|
||||
diffs[current_file] = '\n'.join(current_diff)
|
||||
current_file = line.split('b/')[-1].strip()
|
||||
current_diff = [line]
|
||||
elif current_file:
|
||||
current_diff.append(line)
|
||||
|
||||
if current_file and current_diff:
|
||||
diffs[current_file] = "\n".join(current_diff)
|
||||
diffs[current_file] = '\n'.join(current_diff)
|
||||
|
||||
return diffs
|
||||
except subprocess.CalledProcessError as e:
|
||||
@@ -142,15 +138,17 @@ class GitService:
|
||||
def stage_all_changes(self) -> None:
|
||||
"""Stage all changes in the repository."""
|
||||
try:
|
||||
subprocess.run(["git", "add", "."], cwd=self.repo_path, check=True)
|
||||
subprocess.run(
|
||||
['git', 'add', '.'],
|
||||
cwd=self.repo_path,
|
||||
check=True
|
||||
)
|
||||
except subprocess.CalledProcessError as e:
|
||||
raise Exception(f"Failed to stage changes: {str(e)}")
|
||||
|
||||
def get_commit_history(
|
||||
self, since: Optional[str] = None, limit: Optional[int] = None
|
||||
) -> List[Dict[str, str]]:
|
||||
def get_commit_history(self, since: Optional[str] = None, limit: Optional[int] = None) -> List[Dict[str, str]]:
|
||||
"""Get commit history.
|
||||
|
||||
|
||||
Args:
|
||||
since: Get commits since this reference (tag, commit, etc.)
|
||||
limit: Maximum number of commits to return
|
||||
@@ -158,78 +156,76 @@ class GitService:
|
||||
args = [
|
||||
"log",
|
||||
"--pretty=format:%H%n%an%n%ad%n%B%n==END==",
|
||||
"--date=format:%Y-%m-%d %H:%M:%S %z",
|
||||
"--date=format:%Y-%m-%d %H:%M:%S %z"
|
||||
]
|
||||
|
||||
|
||||
if since:
|
||||
args.append(f"{since}..HEAD")
|
||||
if limit:
|
||||
args.append(f"-n {limit}")
|
||||
|
||||
|
||||
stdout, _, _ = self._run_git_command(args)
|
||||
|
||||
|
||||
commits = []
|
||||
current_commit = []
|
||||
|
||||
for line in stdout.split("\n"):
|
||||
|
||||
for line in stdout.split('\n'):
|
||||
if line == "==END==":
|
||||
if current_commit:
|
||||
commit_hash, author, date, *message_lines = current_commit
|
||||
commits.append(
|
||||
{
|
||||
"hash": commit_hash,
|
||||
"author": author,
|
||||
"date": date,
|
||||
"message": "\n".join(message_lines),
|
||||
}
|
||||
)
|
||||
commits.append({
|
||||
'hash': commit_hash,
|
||||
'author': author,
|
||||
'date': date,
|
||||
'message': '\n'.join(message_lines)
|
||||
})
|
||||
current_commit = []
|
||||
else:
|
||||
current_commit.append(line)
|
||||
|
||||
|
||||
return commits
|
||||
|
||||
def get_commits_since_tag(self, tag: str) -> List[Dict[str, str]]:
|
||||
"""Get commits since a tag.
|
||||
|
||||
|
||||
Args:
|
||||
tag: Git tag to use as reference point
|
||||
|
||||
|
||||
Returns:
|
||||
List of commit dictionaries with hash and message
|
||||
"""
|
||||
cmd = ["git", "log", f"{tag}..HEAD", "--pretty=format:%H|%s"]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=self.repo_path)
|
||||
result.check_returncode()
|
||||
|
||||
|
||||
commits = []
|
||||
for line in result.stdout.splitlines():
|
||||
hash_, message = line.split("|", 1)
|
||||
commits.append({"hash": hash_, "message": message})
|
||||
|
||||
|
||||
return commits
|
||||
|
||||
def get_commits_since_date(self, days: int) -> List[Dict[str, str]]:
|
||||
"""Get commits since a date.
|
||||
|
||||
|
||||
Args:
|
||||
days: Number of days to look back
|
||||
|
||||
|
||||
Returns:
|
||||
List of commit dictionaries with hash and message
|
||||
"""
|
||||
# Calculate date
|
||||
date = datetime.now() - timedelta(days=days)
|
||||
date_str = date.strftime("%Y-%m-%d")
|
||||
|
||||
|
||||
# Get commits
|
||||
cmd = ["git", "log", f"--since={date_str}", "--pretty=format:%H|%s"]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, cwd=self.repo_path)
|
||||
result.check_returncode()
|
||||
|
||||
|
||||
commits = []
|
||||
for line in result.stdout.splitlines():
|
||||
hash_, message = line.split("|", 1)
|
||||
commits.append({"hash": hash_, "message": message})
|
||||
|
||||
return commits
|
||||
|
||||
return commits
|
||||
Reference in new issue
Block a user