This commit is contained in:
sankalp committed 2025-12-12 11:10:30 +05:45
1 parent b5362ade00
commit dec9c15b7a
73 files changed
+3290 -4522

No files matched your search

View File
Whitespace-only changes.
+181
View File
@@ -0,0 +1,181 @@
import re
import subprocess
from collections import defaultdict
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional
from devgen.utils import configure_logger
class ChangelogGenerator:
"""Generates a changelog from git history using Semantic Release style."""
def __init__(self, logger=None):
self.logger = logger or configure_logger("devgen.changelog")
def _exec_git(self, args: List[str]) -> str:
"""Executes a git command."""
try:
res = subprocess.run(
args,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
check=True,
)
return res.stdout.strip()
except subprocess.CalledProcessError as e:
self.logger.error(f"Git command failed: {e}")
raise RuntimeError(f"Git command failed: {e}")
def get_commits(self, from_ref: str = "", to_ref: str = "HEAD") -> List[str]:
"""Fetches commit messages in the specified range."""
range_spec = f"{from_ref}..{to_ref}" if from_ref else to_ref
# Format: hash|author|date|subject|body
fmt = "%H|%an|%ad|%s|%b"
cmd = ["git", "log", f"--format={fmt}", "--date=short", range_spec]
if not from_ref:
# If no start ref, try to find the last tag
try:
last_tag = self._exec_git(["git", "describe", "--tags", "--abbrev=0"])
cmd = [
"git",
"log",
f"--format={fmt}",
"--date=short",
f"{last_tag}..HEAD",
]
self.logger.info(f"Generating changelog from last tag: {last_tag}")
except RuntimeError:
self.logger.info("No tags found, generating for all commits.")
cmd = ["git", "log", f"--format={fmt}", "--date=short"]
return self._exec_git(cmd).split("\n")
def parse_commits(self, raw_commits: List[str]) -> Dict[str, List[Dict]]:
"""Parses raw commit strings into structured data."""
groups = defaultdict(list)
# Conventional Commit Regex: type(scope)!: subject
cc_pattern = re.compile(r"^(\w+)(?:\(([^)]+)\))?(!?):\s+(.*)")
for line in raw_commits:
if not line.strip():
continue
parts = line.split("|", 4)
if len(parts) < 4:
continue
commit_hash, author, date, subject, body = (
parts[0],
parts[1],
parts[2],
parts[3],
parts[4] if len(parts) > 4 else "",
)
match = cc_pattern.match(subject)
if match:
c_type, c_scope, breaking, c_subject = match.groups()
entry = {
"hash": commit_hash,
"author": author,
"date": date,
"scope": c_scope,
"subject": c_subject,
"body": body,
"breaking": bool(breaking),
}
if breaking:
groups["BREAKING CHANGES"].append(entry)
if c_type in ["feat", "feature"]:
groups["Features"].append(entry)
elif c_type in ["fix", "bug"]:
groups["Bug Fixes"].append(entry)
elif c_type in ["docs"]:
groups["Documentation"].append(entry)
elif c_type in [
"style",
"refactor",
"perf",
"test",
"build",
"ci",
"chore",
]:
groups["Other Changes"].append(entry)
else:
groups["Other Changes"].append(entry)
else:
# Non-conventional commits
groups["Other Changes"].append(
{
"hash": commit_hash,
"author": author,
"date": date,
"scope": None,
"subject": subject,
"body": body,
"breaking": False,
}
)
return groups
def generate_markdown(
self, groups: Dict[str, List[Dict]], version: str = "Unreleased"
) -> str:
"""Generates markdown changelog from parsed commits."""
date_str = datetime.now().strftime("%Y-%m-%d")
md = [f"# {version} ({date_str})\n"]
# Order: Breaking, Features, Fixes, Docs, Others
order = [
"BREAKING CHANGES",
"Features",
"Bug Fixes",
"Documentation",
"Other Changes",
]
for section in order:
commits = groups.get(section)
if commits:
md.append(f"## {section}\n")
for c in commits:
scope = f"**{c['scope']}**: " if c["scope"] else ""
md.append(f"- {scope}{c['subject']} ({c['hash'][:7]})")
md.append("")
return "\n".join(md)
def run(self, output_file: Optional[str] = "CHANGELOG.md", from_ref: str = ""):
"""Main execution method."""
raw_commits = self.get_commits(from_ref)
if not raw_commits or not raw_commits[0]:
self.logger.warning("No commits found.")
return
parsed = self.parse_commits(raw_commits)
md_content = self.generate_markdown(parsed)
if output_file:
path = Path(output_file)
# If file exists, prepend? For now, just overwrite or append logic could be complex.
# Let's implement prepend logic if file exists.
if path.exists():
old_content = path.read_text(encoding="utf-8")
new_content = md_content + "\n\n" + old_content
path.write_text(new_content, encoding="utf-8")
else:
path.write_text(md_content, encoding="utf-8")
self.logger.info(f"Changelog written to {output_file}")
print(f"Changelog updated: {output_file}")
else:
print(md_content)
+257
View File
@@ -0,0 +1,257 @@
import subprocess
from collections import defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List
from devgen.ai import generate_with_ai
from devgen.utils import (
configure_logger,
extract_commit_messages,
get_commit_dry_run_path,
is_file_recent,
load_template_env,
sanitize_ai_commit_message,
)
class CommitEngineError(Exception):
"""Exception raised for errors in the commit engine."""
pass
class CommitEngine:
"""
Engine for generating AI-powered commit messages.
Handles detection of changes, grouping, AI generation, and git operations.
"""
def __init__(
self,
dry_run: bool = False,
push: bool = False,
debug: bool = False,
force_rebuild: bool = False,
provider: str = "gemini",
model: str = "gemini-2.5-flash",
logger: Any | None = None,
**kwargs,
):
self.dry_run = dry_run
self.push = push
self.debug = debug
self.force_rebuild = force_rebuild
self.provider = provider
self.model = model
self.logger = logger or configure_logger(
"devgen.commit", Path.home() / ".cache" / "devgen" / "commit.log"
)
self.kwargs = kwargs
self.dry_run_path = get_commit_dry_run_path()
self.template_env = load_template_env("commit")
# Load config from ~/.devgen.yaml
from devgen.utils import load_config
self.config = load_config()
def _exec_git(self, args: List[str], allow_error: bool = False) -> str:
"""Executes a git command."""
try:
res = subprocess.run(
args,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
check=not allow_error,
)
return res.stdout.strip()
except subprocess.CalledProcessError as e:
msg = f"Git command failed: {' '.join(e.cmd)}\nError: {e.stderr.strip()}"
self.logger.error(msg)
raise CommitEngineError(msg) from e
def detect_changes(self) -> List[str]:
"""Detects changed, deleted, or untracked files."""
out = self._exec_git(
[
"git",
"ls-files",
"--deleted",
"--modified",
"--others",
"--exclude-standard",
]
)
return [f.strip() for f in out.split("\n") if f.strip()]
def group_files(self, files: List[str]) -> Dict[str, List[str]]:
"""Groups files by their parent directory."""
groups = defaultdict(list)
for f in files:
parent = str(Path(f).parent)
key = "root" if parent == "." else parent
groups[key].append(f)
return groups
def generate_diff(self, files: List[str]) -> str:
"""Generates diff for specific files."""
return self._exec_git(["git", "--no-pager", "diff", "--staged", "--", *files])
def _init_dry_run(self):
"""Initializes the dry-run file."""
self.dry_run_path.parent.mkdir(parents=True, exist_ok=True)
with self.dry_run_path.open("w", encoding="utf-8") as f:
ts = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S (%Z)")
f.write(f"# Dry Run: Commit Messages\n_Generated: {ts}_\n\n")
def _log_dry_run(self, group: str, msg: str):
"""Appends a dry-run entry."""
with self.dry_run_path.open("a", encoding="utf-8") as f:
f.write(f"## Group: `{group}`\n\n```md\n{msg}\n```\n\n---\n\n")
def stage_files(self, files: List[str]):
"""Stages files in git."""
if not files:
return
self.logger.info(f"Staging: {files}")
self._exec_git(["git", "add", *files])
def commit_staged(self, msg: str):
"""Commits staged changes."""
self.logger.info(f"Committing:\n{msg}")
self._exec_git(["git", "commit", "-m", msg])
def push_commits(self):
"""Pushes commits to remote."""
self.logger.info("Pushing to remote...")
self._exec_git(["git", "push"])
self.logger.info("Push successful.")
def generate_message(self, group: str, diff: str, cache: Dict[str, str]) -> str:
"""Generates a commit message using AI or cache."""
if not self.force_rebuild and group in cache:
self.logger.info(f"Using cached message for {group}")
return cache[group]
# Get settings from config or kwargs
provider = (
self.kwargs.get("provider") or self.config.get("provider") or self.provider
)
model = self.kwargs.get("model") or self.config.get("model") or self.model
api_key = self.kwargs.get("api_key") or self.config.get("api_key")
use_emoji = self.config.get("emoji", True)
template = self.template_env.get_template("commit_message.j2")
prompt = template.render(group_name=group, diff_text=diff, use_emoji=use_emoji)
raw = generate_with_ai(
prompt,
provider=provider,
model=model,
api_key=api_key,
debug=self.debug,
**self.kwargs,
)
return sanitize_ai_commit_message(raw)
def is_ahead_of_remote(self) -> bool:
"""Checks if local branch has unpushed commits."""
try:
self._exec_git(["git", "fetch", "origin"])
count = self._exec_git(
["git", "rev-list", "--count", "@{u}..HEAD"], allow_error=True
)
if count and int(count) > 0:
return True
except CommitEngineError:
# Maybe no upstream
return bool(self._exec_git(["git", "rev-parse", "HEAD"], allow_error=True))
return False
def load_cache(self) -> Dict[str, str]:
"""Loads dry-run cache."""
if self.dry_run:
self._init_dry_run()
return {}
if not self.force_rebuild and is_file_recent(self.dry_run_path):
self.logger.info(f"Loading cache from {self.dry_run_path}")
return extract_commit_messages(self.dry_run_path)
return {}
def process_group(
self, group: str, files: List[str], cache: Dict[str, str]
) -> bool:
"""Processes a single file group."""
self.stage_files(files)
diff = self.generate_diff(files)
if not diff.strip():
self.logger.info(f"Skipping empty diff for {group}")
self._exec_git(["git", "reset", "HEAD", "--", *files])
return True
msg = self.generate_message(group, diff, cache)
if not msg:
self.logger.error(f"Empty message for {group}")
return False
if self.dry_run:
self._log_dry_run(group, msg)
self._exec_git(["git", "reset", "HEAD", "--", *files])
else:
self.commit_staged(msg)
return True
def execute(self):
"""Main execution method."""
files = self.detect_changes()
ahead = self.is_ahead_of_remote()
if not files and not ahead:
self.logger.info("Nothing to commit or push.")
return
failed = []
if files:
groups = self.group_files(files)
cache = self.load_cache()
for group, group_files in groups.items():
try:
if not self.process_group(group, group_files, cache):
failed.append(group)
except KeyboardInterrupt:
self.logger.warning("\nOperation interrupted by user.")
raise
else:
self.logger.info("No changes to commit, checking push...")
if self.push and not self.dry_run:
if not failed:
self.push_commits()
else:
self.logger.error("Push aborted due to failed commits.")
if self.dry_run:
self.logger.info(f"Dry run done. See {self.dry_run_path}")
else:
self.logger.info("Done.")
if failed:
self.logger.warning(f"Failed groups: {failed}")
def run_commit_engine(**kwargs):
"""Entry point for the commit engine."""
logger = configure_logger("devgen.commit")
try:
engine = CommitEngine(**kwargs)
engine.execute()
except Exception as e:
logger.error(f"Commit engine failed: {e}", exc_info=True)
__all__ = ["run_commit_engine"]
+116
View File
@@ -0,0 +1,116 @@
from pathlib import Path
from typing import List
import requests
from devgen.utils import configure_logger
class GitignoreGenerator:
"""Fetches and manages .gitignore templates from GitHub."""
def __init__(self, logger=None):
self.logger = logger or configure_logger("devgen.gitignore")
self.cache_dir = Path.home() / ".cache" / "devgen" / "gitignore"
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.api_url = "https://api.github.com/repos/github/gitignore/contents/"
self.raw_url_base = "https://raw.githubusercontent.com/github/gitignore/main/"
def list_available_templates(self) -> List[str]:
"""Lists available templates from GitHub."""
try:
response = requests.get(self.api_url, timeout=10)
response.raise_for_status()
data = response.json()
return [
item["name"].replace(".gitignore", "")
for item in data
if item["name"].endswith(".gitignore")
]
except requests.RequestException as e:
self.logger.error(f"Failed to fetch templates: {e}")
raise RuntimeError(f"Failed to fetch templates: {e}")
def list_cached_templates(self) -> List[str]:
"""Lists cached templates."""
if not self.cache_dir.exists():
return []
return [f.stem for f in self.cache_dir.glob("*.gitignore")]
def get_template_content(
self, name: str, use_cache: bool = True, offline: bool = False
) -> str:
"""Fetches template content from cache or GitHub."""
cache_file = self.cache_dir / f"{name}.gitignore"
if offline:
if cache_file.exists():
self.logger.info(f"Using cached template for {name}")
return cache_file.read_text(encoding="utf-8")
else:
raise RuntimeError(
f"Template '{name}' not found in cache (offline mode)."
)
# Try to fetch from GitHub
try:
url = f"{self.raw_url_base}{name}.gitignore"
response = requests.get(url, timeout=10)
if response.status_code == 404:
# Try with capitalized first letter if lowercase fails, though the list should be accurate
# The GitHub repo is case-sensitive for raw URLs usually matching the filename.
# The list_available_templates returns exact filenames (minus extension).
# So if we use the name from the list, it should be correct.
# However, if user inputs manually, we might need to handle case.
raise RuntimeError(f"Template '{name}' not found on GitHub.")
response.raise_for_status()
content = response.text
if use_cache:
cache_file.write_text(content, encoding="utf-8")
self.logger.info(f"Cached template for {name}")
return content
except requests.RequestException as e:
if cache_file.exists():
self.logger.warning(f"Failed to fetch {name}, using cache: {e}")
return cache_file.read_text(encoding="utf-8")
raise RuntimeError(f"Failed to fetch {name} and no cache available: {e}")
def generate(
self,
templates: List[str],
output_file: str = ".gitignore",
append: bool = True,
offline: bool = False,
):
"""Generates the .gitignore file."""
content = ""
for name in templates:
try:
tpl_content = self.get_template_content(name, offline=offline)
content += f"\n### {name} ###\n{tpl_content}\n"
except Exception as e:
self.logger.error(str(e))
print(f" Error fetching {name}: {e}")
if not content:
return
mode = "a" if append else "w"
path = Path(output_file)
if path.exists() and append:
print(f"Appending to {output_file}...")
elif path.exists() and not append:
print(f"Overwriting {output_file}...")
else:
print(f"Creating {output_file}...")
with path.open(mode, encoding="utf-8") as f:
f.write(content)
print(f" Successfully wrote to {output_file}")
+80
View File
@@ -0,0 +1,80 @@
import json
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional
class LicenseGenerator:
"""Generates license files from templates."""
def __init__(self):
self.templates_dir = (
Path(__file__).parent.parent / "prompt_templates" / "licenses"
)
def list_licenses(self) -> List[Dict[str, str]]:
"""Lists available license templates."""
licenses = []
if not self.templates_dir.exists():
return []
for file_path in self.templates_dir.glob("*.json"):
try:
with file_path.open("r", encoding="utf-8") as f:
data = json.load(f)
licenses.append(
{
"key": data.get("key", file_path.stem),
"name": data.get("name", "Unknown License"),
"description": data.get("description", ""),
}
)
except Exception:
continue
# Sort by name
return sorted(licenses, key=lambda x: x["name"])
def get_license_template(self, key: str) -> Optional[Dict]:
"""Loads a specific license template by key."""
# We assume key matches filename for simplicity, or we search.
# Based on file listing, filenames match keys (e.g. mit.json -> mit)
file_path = self.templates_dir / f"{key}.json"
if not file_path.exists():
return None
try:
with file_path.open("r", encoding="utf-8") as f:
return json.load(f)
except Exception:
return None
def render_license(self, key: str, author: str, year: str = "") -> str:
"""Renders the license content with placeholders replaced."""
template_data = self.get_license_template(key)
if not template_data:
raise ValueError(f"License template '{key}' not found.")
content = template_data.get("template", "")
if not year:
year = str(datetime.now().year)
# Replace placeholders
# Support various formats found in templates
year_placeholders = ["[year]", "[yyyy]", "{{year}}", "<year>"]
author_placeholders = [
"[fullname]",
"[name of copyright owner]",
"{{author}}",
"<name of author>",
"[author]",
]
for p in year_placeholders:
content = content.replace(p, year)
for p in author_placeholders:
content = content.replace(p, author)
return content
+66
View File
@@ -0,0 +1,66 @@
from datetime import datetime
from pathlib import Path
from devgen.modules.changelog_generator import ChangelogGenerator
from devgen.utils import configure_logger
class ReleaseNotesGenerator(ChangelogGenerator):
"""Generate short, clean release notes from git history."""
SECTION_EMOJIS = {
"BREAKING CHANGES": "⚠️",
"Features": "✨",
"Bug Fixes": "🐛",
"Documentation": "📚",
"Refactor": "♻️",
"Tests": "✅",
"Chore": "🔧",
"Style": "✨",
"Other Changes": "🧹",
}
def __init__(self, logger=None):
super().__init__(logger or configure_logger("devgen.releasenotes"))
def generate_release_markdown(self, groups, version="Unreleased"):
"""Generate human-friendly release notes."""
date_str = datetime.now().strftime("%Y-%m-%d")
md = [f"## 🚀 Release {version} — {date_str}\n"]
order = [
"BREAKING CHANGES",
"Features",
"Bug Fixes",
"Documentation",
"Other Changes",
]
for section in order:
commits = groups.get(section)
if not commits:
continue
emoji = self.SECTION_EMOJIS.get(section, "")
md.append(f"### {emoji} {section}")
for c in commits:
scope = f"**{c['scope']}**: " if c["scope"] else ""
md.append(f"- {scope}{c['subject']}")
md.append("")
return "\n".join(md)
def run(self, output_file="RELEASE-NOTES.md", version="Unreleased", from_ref=""):
"""Main logic: get commits → parse → generate → write."""
raw_commits = self.get_commits(from_ref)
if not raw_commits or not raw_commits[0]:
print("❗ No commits found for release notes.")
return
parsed = self.parse_commits(raw_commits)
md = self.generate_release_markdown(parsed, version=version)
Path(output_file).write_text(md, encoding="utf-8")
print(f" Release notes written to {output_file}")