diff --git a/devgen/modules/changelog_generator.py b/devgen/modules/changelog_generator.py index f618f81..fb44d9a 100644 --- a/devgen/modules/changelog_generator.py +++ b/devgen/modules/changelog_generator.py @@ -5,189 +5,151 @@ from datetime import datetime from pathlib import Path from typing import Dict, List, Optional +from devgen.modules.changelog_sections import ( + DEFAULT_STYLES, + Section, + section_for_type, +) from devgen.utils import configure_logger, run_git_command class ChangelogGenerator: """Generates a changelog from git history using Semantic Release style.""" + #: Conventional commit header: ``type(scope)!: subject`` + _CC_PATTERN = re.compile(r"^(\w+)(?:\(([^)]+)\))?(!?):\s+(.*)") + def __init__(self, logger=None): self.logger = logger or configure_logger("devgen.changelog") + # ------------------------------------------------------------------ commits + 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 = run_git_command(["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, subprocess.CalledProcessError): - self.logger.info("No tags found, generating for all commits.") - cmd = ["git", "log", f"--format={fmt}", "--date=short"] - + """Fetch commit lines in the requested range.""" + if from_ref: + range_spec = f"{from_ref}..{to_ref}" + cmd = self._log_cmd(range_spec) + else: + cmd = self._resolve_range(to_ref) try: return run_git_command(cmd).split("\n") except subprocess.CalledProcessError as e: self.logger.error(f"Git command failed: {e}") - raise RuntimeError(f"Git command failed: {e}") + raise RuntimeError(f"Git command failed: {e}") from e - 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 == "perf": - groups["Performance"].append(entry) - elif c_type == "refactor": - groups["Refactor"].append(entry) - elif c_type == "test": - groups["Tests"].append(entry) - elif c_type == "style": - groups["Style"].append(entry) - elif c_type in ["build", "ci", "chore"]: - groups["Chore"].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, Refactor, Perf, Tests, Style, Chore, Others - order = [ - "BREAKING CHANGES", - "Features", - "Bug Fixes", - "Performance", - "Documentation", - "Refactor", - "Tests", - "Style", - "Chore", - "Other Changes", + def _log_cmd(self, range_spec: str) -> List[str]: + # Format: hash|author|date|subject|body + return [ + "git", + "log", + "--format=%H|%an|%ad|%s|%b", + "--date=short", + range_spec, ] - emoji_map = { - "BREAKING CHANGES": "๐Ÿ’ฅ BREAKING CHANGES", - "Features": "โœจ Features", - "Bug Fixes": "๐Ÿ› Bug Fixes", - "Performance": "โšก Performance", - "Documentation": "๐Ÿ“š Documentation", - "Refactor": "โ™ป๏ธ Refactor", - "Tests": "โœ… Tests", - "Style": "๐Ÿ’„ Style", - "Chore": "๐Ÿ”ง Chore", - "Other Changes": "๐Ÿงน Other Changes", + def _resolve_range(self, to_ref: str) -> List[str]: + try: + last_tag = run_git_command(["git", "describe", "--tags", "--abbrev=0"]) + self.logger.info(f"Generating changelog from last tag: {last_tag}") + return self._log_cmd(f"{last_tag}..HEAD") + except (RuntimeError, subprocess.CalledProcessError): + self.logger.info("No tags found, generating for all commits.") + return ["git", "log", "--format=%H|%an|%ad|%s|%b", "--date=short"] + + # ----------------------------------------------------------------- parsing + + def parse_commits(self, raw_commits: List[str]) -> Dict[Section, List[Dict]]: + """Group raw commit lines into :class:`Section` buckets.""" + groups: Dict[Section, List[Dict]] = defaultdict(list) + for line in raw_commits: + entry = self._parse_line(line) + if entry is None: + continue + section = section_for_type(entry["type"]) + entry["section"] = section + if entry["breaking"]: + groups[Section.BREAKING].append(entry) + groups[section].append(entry) + return dict(groups) + + def _parse_line(self, line: str) -> Dict | None: + if not line.strip(): + return None + parts = line.split("|", 4) + if len(parts) < 4: + return None + commit_hash, author, date, subject = parts[:4] + body = parts[4] if len(parts) > 4 else "" + + match = self._CC_PATTERN.match(subject) + if match: + c_type, c_scope, breaking, c_subject = match.groups() + return { + "type": c_type, + "hash": commit_hash, + "author": author, + "date": date, + "scope": c_scope, + "subject": c_subject, + "body": body, + "breaking": bool(breaking), + } + return { + "type": "", + "hash": commit_hash, + "author": author, + "date": date, + "scope": None, + "subject": subject, + "body": body, + "breaking": False, } - for section in order: - commits = groups.get(section) - if commits: - header = emoji_map.get(section, section) - md.append(f"## {header}\n") - for c in commits: - scope = f"**{c['scope']}**: " if c["scope"] else "" - md.append(f"- {scope}{c['subject']} ({c['hash'][:7]})") - md.append("") + # ------------------------------------------------------------------ output - return "\n".join(md) + def generate_markdown( + self, + groups: Dict[Section, List[Dict]], + version: str = "Unreleased", + ) -> str: + """Render groups as a markdown changelog block.""" + date_str = datetime.now().strftime("%Y-%m-%d") + out = [f"## {version} ({date_str})\n"] + for section in Section.ordered(): + commits = groups.get(section) + if not commits: + continue + style = DEFAULT_STYLES[section] + out.append(f"## {style.emoji} {style.heading}\n") + for c in commits: + scope = f"**{c['scope']}**: " if c.get("scope") else "" + out.append(f"- {scope}{c['subject']} ({c['hash'][:7]})") + out.append("") + return "\n".join(out) def run(self, output_file: Optional[str] = "CHANGELOG.md", from_ref: str = ""): - """Main execution method.""" + """Fetch โ†’ parse โ†’ write (or print) the changelog.""" raw_commits = self.get_commits(from_ref) if not raw_commits or not raw_commits[0]: self.logger.warning("No commits found.") return + groups = self.parse_commits(raw_commits) + md_content = self.generate_markdown(groups) - parsed = self.parse_commits(raw_commits) - md_content = self.generate_markdown(parsed) - - if output_file: - path = Path(output_file) - if path.exists(): - old_content = path.read_text(encoding="utf-8") - # Prepend the new content, assume # CHANGELOG is at the top or needs to be - if old_content.strip().startswith("# CHANGELOG"): - lines = old_content.split("\n", 1) - header = lines[0] - rest = lines[1] if len(lines) > 1 else "" - new_content = f"{header}\n\n{md_content}\n{rest.lstrip()}" - else: - new_content = f"# CHANGELOG\n\n{md_content}\n\n{old_content}" - path.write_text(new_content, encoding="utf-8") - else: - path.write_text(f"# CHANGELOG\n\n{md_content}", encoding="utf-8") - - self.logger.info(f"Changelog written to {output_file}") - print(f" Changelog updated: {output_file}") - else: + if not output_file: print(md_content) + return + + path = Path(output_file) + if path.exists(): + old = path.read_text(encoding="utf-8") + if old.strip().startswith("# CHANGELOG"): + header, _, rest = old.partition("\n") + new_content = f"{header}\n\n{md_content}\n{rest.lstrip()}" + else: + new_content = f"# CHANGELOG\n\n{md_content}\n\n{old}" + else: + new_content = f"# CHANGELOG\n\n{md_content}" + path.write_text(new_content, encoding="utf-8") + self.logger.info(f"Changelog written to {output_file}") + print(f" Changelog updated: {output_file}") diff --git a/devgen/modules/changelog_sections.py b/devgen/modules/changelog_sections.py new file mode 100644 index 0000000..d4f3c7e --- /dev/null +++ b/devgen/modules/changelog_sections.py @@ -0,0 +1,79 @@ +"""Shared section definitions for changelog and release-note generators. + +Both generators render the same set of Conventional Commit groups; this +module is the single source of truth for the section name, the emoji, +and the display order. +""" + +from dataclasses import dataclass +from enum import Enum +from typing import Mapping + + +class Section(str, Enum): + """Sections in the order they should be rendered.""" + + BREAKING = "BREAKING CHANGES" + FEATURES = "Features" + BUG_FIXES = "Bug Fixes" + PERFORMANCE = "Performance" + DOCUMENTATION = "Documentation" + REFACTOR = "Refactor" + TESTS = "Tests" + STYLE = "Style" + CHORE = "Chore" + OTHER = "Other Changes" + + @classmethod + def ordered(cls) -> tuple["Section", ...]: + return tuple(cls) + + +@dataclass(frozen=True) +class SectionStyle: + """Visual style for a section (emoji + name for the heading).""" + + emoji: str + heading: str + + +# Default style for each section. Generators can override per output +# format (changelog vs. release notes). +DEFAULT_STYLES: Mapping[Section, SectionStyle] = { + Section.BREAKING: SectionStyle("๐Ÿ’ฅ", "BREAKING CHANGES"), + Section.FEATURES: SectionStyle("โœจ", "Features"), + Section.BUG_FIXES: SectionStyle("๐Ÿ›", "Bug Fixes"), + Section.PERFORMANCE: SectionStyle("โšก", "Performance"), + Section.DOCUMENTATION: SectionStyle("๐Ÿ“š", "Documentation"), + Section.REFACTOR: SectionStyle("โ™ป๏ธ", "Refactor"), + Section.TESTS: SectionStyle("โœ…", "Tests"), + Section.STYLE: SectionStyle("๐Ÿ’„", "Style"), + Section.CHORE: SectionStyle("๐Ÿ”ง", "Chore"), + Section.OTHER: SectionStyle("๐Ÿงน", "Other Changes"), +} + + +# Conventional commit type โ†’ Section mapping. Long-form aliases ("feature", +# "bug") are also accepted. +TYPE_TO_SECTION: Mapping[str, Section] = { + "feat": Section.FEATURES, + "feature": Section.FEATURES, + "fix": Section.BUG_FIXES, + "bug": Section.BUG_FIXES, + "perf": Section.PERFORMANCE, + "docs": Section.DOCUMENTATION, + "refactor": Section.REFACTOR, + "test": Section.TESTS, + "style": Section.STYLE, + "build": Section.CHORE, + "ci": Section.CHORE, + "chore": Section.CHORE, +} + + +def section_for_type(commit_type: str) -> Section: + """Return the Section for a Conventional Commit ``type``. + + Unknown types fall back to :attr:`Section.OTHER`. + """ + return TYPE_TO_SECTION.get(commit_type.lower(), Section.OTHER) diff --git a/devgen/modules/commit_generator.py b/devgen/modules/commit_generator.py index ad63221..0a20127 100644 --- a/devgen/modules/commit_generator.py +++ b/devgen/modules/commit_generator.py @@ -1,9 +1,18 @@ -import subprocess +"""Orchestrates the AI-powered commit flow. + +The :class:`CommitEngine` is a thin coordinator. The real work lives in: + +* :class:`devgen.modules.git_ops.GitOperator` โ€” all git side-effects +* :class:`devgen.modules.diff_builder.FileGrouper` โ€” files โ†’ commit groups +* :class:`devgen.modules.diff_builder.DiffBuilder` โ€” staged diff + truncation +* :class:`devgen.modules.diff_builder.ManifestInspector` โ€” project context +* :class:`devgen.ai.generate_with_ai` โ€” provider-agnostic AI call +""" + import sys -from collections import defaultdict from datetime import datetime from pathlib import Path -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional import questionary from rich.console import Console @@ -12,6 +21,13 @@ from rich.panel import Panel from rich.theme import Theme from devgen.ai import generate_with_ai +from devgen.modules.diff_builder import ( + DEFAULT_MAX_DIFF_SIZE, + DiffBuilder, + FileGrouper, + ManifestInspector, +) +from devgen.modules.git_ops import GitOperator from devgen.utils import ( configure_logger, extract_commit_messages, @@ -19,34 +35,112 @@ from devgen.utils import ( is_file_recent, load_template_env, render_custom_template, - run_git_command, sanitize_ai_commit_message, ) class CommitEngineError(Exception): - """Exception raised for errors in the commit engine.""" - - pass + """Raised when the commit flow cannot proceed.""" -# Token Optimization Constants -MAX_DIFF_SIZE = 8000 # Maximum characters for a single group diff -IGNORE_PATTERNS = [ - "uv.lock", - "package-lock.json", - "pnpm-lock.yaml", - "yarn.lock", - "composer.lock", - "Gemfile.lock", - "poetry.lock", -] +class CacheManager: + """Manages the dry-run cache file at ``~/.cache/devgen/commit_dry_run.md``.""" + + def __init__(self, path: Path, max_age_minutes: int = 120) -> None: + self.path = path + + def init_dry_run(self) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + ts = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S (%Z)") + with self.path.open("w", encoding="utf-8") as f: + f.write(f"# Dry Run: Commit Messages\n_Generated: {ts}_\n\n") + + def load(self, *, dry_run: bool, force_rebuild: bool) -> Dict[str, str]: + if dry_run: + self.init_dry_run() + return {} + if not force_rebuild and is_file_recent(self.path): + return extract_commit_messages(self.path) + return {} + + def append(self, group: str, message: str) -> None: + with self.path.open("a", encoding="utf-8") as f: + f.write(f"## Group: `{group}`\n\n```md\n{message}\n```\n\n---\n\n") + + +class CommitMessageBuilder: + """Renders the prompt and calls the AI provider.""" + + def __init__( + self, + template_env, + config: Dict[str, Any], + provider: str, + model: str, + debug: bool = False, + ) -> None: + self.template_env = template_env + self.config = config + self.provider = provider + self.model = model + self.debug = debug + + def build_message( + self, + group: str, + diff: str, + manifest_context: str, + extra_kwargs: Dict[str, Any], + ) -> str: + custom_template = self.config.get("custom_template") + use_emoji = self.config.get("emoji", True) + ollama_host = self.config.get("ollama_host") + + if custom_template: + prompt = render_custom_template( + custom_template, + group_name=group, + diff_text=diff, + use_emoji=use_emoji, + context=manifest_context, + ) + else: + prompt = ( + self.template_env.get_template("commit_message.tpl") + .render( + group_name=group, + diff_text=diff, + use_emoji=use_emoji, + context=manifest_context, + ) + .strip() + ) + + provider = ( + extra_kwargs.get("provider") or self.config.get("provider") or self.provider + ) + model = extra_kwargs.get("model") or self.config.get("model") or self.model + api_key = extra_kwargs.get("api_key") or self.config.get("api_key") + + call_kwargs = dict(extra_kwargs) + if ollama_host and provider == "ollama": + call_kwargs.setdefault("ollama_host", ollama_host) + return generate_with_ai( + prompt, + provider=provider, + model=model, + api_key=api_key, + debug=self.debug, + **call_kwargs, + ) class CommitEngine: - """ - Engine for generating AI-powered commit messages. - Handles detection of changes, grouping, AI generation, and git operations. + """Coordinates the AI-powered commit flow. + + The constructor signature is stable (used by ``cli/commit.py``) but + most logic now lives in composed objects. The public ``execute()`` + method remains the single entry point. """ def __init__( @@ -61,8 +155,8 @@ class CommitEngine: logger: Any | None = None, max_groups: int | None = None, max_diff_size: int | None = None, - **kwargs, - ): + **kwargs: Any, + ) -> None: self.dry_run = dry_run self.push = push self.debug = debug @@ -76,532 +170,218 @@ class CommitEngine: console=debug, ) self.kwargs = kwargs - self.dry_run_path = get_commit_dry_run_path() - self.template_env = load_template_env("commit") - # CLI flags take precedence over config. Config can still be consulted - # in group_files / generate_diff when these are None. - self._max_groups_override = max_groups - self._max_diff_size_override = max_diff_size + + # CLI flags win over config; config wins over default. + from devgen.utils import load_config + + self.config = load_config() self.console = Console( theme=Theme( {"info": "dim cyan", "warning": "magenta", "danger": "bold red"} ) ) - - # Load config from ~/.devgen.yaml - from devgen.utils import load_config - - self.config = load_config() - - def detect_changes(self) -> List[str]: - """Detects changed, deleted, or untracked files.""" - try: - # Get modified, deleted and untracked files - out = run_git_command( - [ - "git", - "ls-files", - "--deleted", - "--modified", - "--others", - "--exclude-standard", - ] - ) - files = [f.strip() for f in out.split("\n") if f.strip()] - - # Also get staged files (in case of a previous failed run) - staged_out = run_git_command(["git", "diff", "--name-only", "--cached"]) - staged_files = [f.strip() for f in staged_out.split("\n") if f.strip()] - - # Combine and deduplicate - all_files = list(set(files + staged_files)) - return all_files - 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 group_files(self, files: List[str]) -> Dict[str, List[str]]: - """Groups files by their parent directory with smart merging if limit is exceeded.""" - max_groups = ( - self._max_groups_override - if self._max_groups_override is not None - else self.config.get("max_groups", 5) - ) - - # 1. Initial grouping by immediate parent - groups = defaultdict(list) - for f in files: - parent = str(Path(f).parent) - key = "root" if parent == "." else parent - groups[key].append(f) - - if len(groups) <= max_groups: - return dict(groups) - - self.logger.info( - f"Too many groups ({len(groups)}). Merging based on max_groups={max_groups}" - ) - - # 2. Iteratively merge the deepest group into its parent until we hit the limit - while len(groups) > max_groups: - # Find the deepest path among the current group keys - # Skip 'root' as it's the top level - potential_merges = [k for k in groups.keys() if k != "root"] - if not potential_merges: - # This could happen if only 'root' is left or if max_groups is very small - break - - # Deepest path is the one with the most segments - deepest = max(potential_merges, key=lambda p: len(Path(p).parts)) - - # Find the parent of this deepest path - parent_path = str(Path(deepest).parent) - new_key = "root" if parent_path == "." else parent_path - - # Merge files into the new key - self.logger.debug(f"Merging group '{deepest}' into '{new_key}'") - groups[new_key].extend(groups.pop(deepest)) - - return dict(groups) - - def generate_diff(self, files: List[str]) -> str: - """Generates diff for specific files, with truncation for token optimization.""" - try: - # Filter out very large metadata files that don't need full diffs - summary_info = [] - files_to_diff = [] - for f in files: - if any(p in f for p in IGNORE_PATTERNS): - summary_info.append(f"[METADATA UPDATED] {f}") - else: - files_to_diff.append(f) - - diff = "" - if files_to_diff: - diff = run_git_command( - ["git", "--no-pager", "diff", "--staged", "--", *files_to_diff] - ) - - full_content = "\n".join(summary_info + [diff]).strip() - - # Truncate if too large - max_size = ( - self._max_diff_size_override - if self._max_diff_size_override is not None - else self.config.get("max_diff_size", MAX_DIFF_SIZE) - ) - if len(full_content) > max_size: - self.logger.info( - f"Truncating diff from {len(full_content)} to {max_size} chars " - f"(config: max_diff_size). If you still hit token-limit errors, " - f"lower this further or split the commit into more groups." - ) - half = max_size // 2 - return ( - full_content[:half] - + "\n\n... [DIFF TRUNCATED FOR TOKEN OPTIMIZATION] ...\n\n" - + full_content[-half:] - ) - - return full_content - 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 _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 and prints to console.""" - self.console.print( - Panel( - Markdown(msg), - title=f"Dry Run: {group}", - border_style="yellow", - expand=False, + self.git = GitOperator(logger=self.logger) + self.grouper = FileGrouper( + max_groups=( + max_groups + if max_groups is not None + else self.config.get("max_groups", 5) ) ) - 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.console.print(f"[info]Staging {len(files)} files...[/info]") - try: - run_git_command(["git", "add", *files]) - 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 commit_staged(self, msg: str): - """Commits staged changes.""" - self.logger.info(f"Committing:\n{msg}") - self.console.print( - Panel(Markdown(msg), title="Commit Message", border_style="green") + self.diff_builder = DiffBuilder( + max_size=( + max_diff_size + if max_diff_size is not None + else self.config.get("max_diff_size", DEFAULT_MAX_DIFF_SIZE) + ), + logger=self.logger, ) - try: - run_git_command(["git", "commit", "-m", msg]) - 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 push_commits(self): - """Pushes commits to remote.""" - self.logger.info("Pushing to remote...") - with self.console.status("[bold green]Pushing to remote...[/bold green]"): - try: - # Fetch first so a non-fast-forward push fails fast with a - # clear message instead of getting rejected by the remote. - try: - upstream = run_git_command( - [ - "git", - "rev-parse", - "--abbrev-ref", - "--symbolic-full-name", - "@{u}", - ], - check=False, - ) - if upstream: - run_git_command( - ["git", "fetch", "origin", upstream], check=False - ) - except (subprocess.CalledProcessError, CommitEngineError): - pass - - run_git_command(["git", "push"]) - except subprocess.CalledProcessError as e: - msg = ( - f"Git command failed: {' '.join(e.cmd)}\nError: {e.stderr.strip()}" - ) - self.logger.error(msg) - # Check for "no upstream branch" specifically to give a hint? - if "no upstream branch" in msg.lower(): - self.console.print( - "[warning]No upstream branch. Skipping push.[/warning]" - ) - return - raise CommitEngineError(msg) from e - self.console.print("[bold green]Push successful.[/bold green]") - - def _get_manifest_context(self) -> str: - """Return a compact one-line-per-file summary of project manifests. - - Dumping the first 100 lines of pyproject.toml/requirements.txt is - extremely wasteful โ€” most of it is build/tool config the model does - not need. We extract just the project identity and the package list - so the model can write context-aware commit messages. - """ - import json - import re - - import toml - - def _pkg_name(spec: str) -> str: - return re.split(r"[><=!~\[]", spec, 1)[0].strip() - - def _pyproject() -> str | None: - path = Path("pyproject.toml") - if not path.exists(): - return None - try: - data = toml.load(path) - except Exception: - return None - proj = data.get("project", {}) if isinstance(data, dict) else {} - parts = [ - f"{k}={proj[k]}" - for k in ("name", "version", "description") - if proj.get(k) - ] - deps = proj.get("dependencies") or [] - if deps: - names = [n for n in (_pkg_name(d) for d in deps) if n] - if names: - parts.append(f"deps={','.join(names)}") - return f"[pyproject.toml] {' '.join(parts)}" if parts else None - - def _package_json() -> str | None: - path = Path("package.json") - if not path.exists(): - return None - try: - data = json.loads(path.read_text(encoding="utf-8")) - except Exception: - return None - parts = [ - f"{k}={data[k]}" - for k in ("name", "version", "description") - if data.get(k) - ] - deps = data.get("dependencies") or {} - if isinstance(deps, dict) and deps: - parts.append(f"deps={','.join(deps.keys())}") - return f"[package.json] {' '.join(parts)}" if parts else None - - def _requirements() -> str | None: - path = Path("requirements.txt") - if not path.exists(): - return None - try: - names = [ - n - for n in ( - _pkg_name(line) - for line in path.read_text(encoding="utf-8").splitlines() - ) - if n and not n.startswith("#") - ] - except Exception: - return None - return f"[requirements.txt] deps={','.join(names)}" if names else None - - lines = [s for s in (_pyproject(), _package_json(), _requirements()) if s] - if not lines: - return "" - return "\nProject: " + " | ".join(lines) - - 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 + self.template_env = load_template_env("commit") + self.cache = CacheManager(get_commit_dry_run_path()) + self.message_builder = CommitMessageBuilder( + self.template_env, + self.config, + provider=self.provider, + model=self.model, + debug=self.debug, ) - 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) - custom_template = self.config.get("custom_template") - ollama_host = self.config.get("ollama_host") - manifest_context = self._get_manifest_context() - if manifest_context: - self.logger.info("Including manifest context in prompt") + # ------------------------------------------------------------------ public - if custom_template: - self.logger.info("Using custom template from config") - prompt = render_custom_template( - custom_template, - group_name=group, - diff_text=diff, - use_emoji=use_emoji, - context=manifest_context, - ) - else: - template = self.template_env.get_template("commit_message.tpl") - prompt = template.render( - group_name=group, - diff_text=diff, - use_emoji=use_emoji, - context=manifest_context, - ).strip() - - with self.console.status("[bold blue]Generating commit message...[/bold blue]"): - call_kwargs = dict(self.kwargs) - if ollama_host and provider == "ollama": - call_kwargs.setdefault("ollama_host", ollama_host) - raw = generate_with_ai( - prompt, - provider=provider, - model=model, - api_key=api_key, - debug=self.debug, - **call_kwargs, - ) - return sanitize_ai_commit_message(raw) - - def is_ahead_of_remote(self, fetch: bool = False) -> bool: - """Checks if local branch has unpushed commits. - - Args: - fetch: When True, refresh remote refs first (call only when you - intend to push). Avoids an unconditional network call on - every run and a TOCTOU window between fetch and rev-list. - """ - try: - upstream = run_git_command( - ["git", "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"], - check=False, - ) - except (subprocess.CalledProcessError, CommitEngineError): - upstream = "" - - if not upstream: - return False - - if fetch: - try: - run_git_command(["git", "fetch", "origin", upstream], check=False) - except (subprocess.CalledProcessError, CommitEngineError): - pass - - try: - count = run_git_command( - ["git", "rev-list", "--count", "@{u}..HEAD"], check=False - ) - return bool(count) and int(count) > 0 - except (subprocess.CalledProcessError, CommitEngineError, ValueError): - 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}") - try: - run_git_command(["git", "reset", "HEAD", "--", *files]) - except subprocess.CalledProcessError: - pass # Ignore reset errors - return True - - msg = self.generate_message(group, diff, cache) - try: - if not msg: - self.logger.error(f"Empty message for {group}") - self._reset_group(files) - return False - - if self.dry_run: - self._log_dry_run(group, msg) - self._reset_group(files) - else: - if self.check: - self.console.print( - Panel( - Markdown(msg), - title=f"Proposed Commit Message [group: {group}]", - border_style="cyan", - ) - ) - choice = questionary.select( - "How would you like to proceed?", - choices=[ - "Confirm", - "Edit", - "Abort", - ], - default="Confirm", - ).ask() - - if not choice or choice == "Abort": - self.logger.info(f"Commit aborted by user at group {group}") - self._reset_group(files) - raise KeyboardInterrupt("User aborted") - - if choice == "Edit": - msg = questionary.text( - "Edit commit message:", - multiline=True, - default=msg, - ).ask() - if not msg: - self.logger.info( - f"Empty edit, commit cancelled for {group}" - ) - self._reset_group(files) - return True - - self.commit_staged(msg) - - return True - except Exception as e: - self.logger.error(f"Failed to process group {group!r}: {e}", exc_info=True) - self._reset_group(files) - return False - - def _reset_group(self, files: List[str]): - """Unstages files for a group.""" - try: - run_git_command(["git", "reset", "HEAD", "--", *files]) - except subprocess.CalledProcessError: - pass # Ignore reset errors - - def execute(self): - """Main execution method.""" - files = self.detect_changes() - # Only consult the remote state if the user actually wants to push. - # This avoids a network round-trip on every run and a TOCTOU window. - ahead = self.is_ahead_of_remote(fetch=self.push) if self.push else False + def execute(self) -> None: + files = self.git.detect_changed() + # Only consult the remote state if the user wants to push โ€” + # avoids a network round-trip on every run. + ahead = self.git.is_ahead_of_remote(fetch=self.push) if self.push else False if not files and not ahead: self.logger.info("Nothing to commit or push.") return - failed = [] + failed: list[str] = [] if files: - groups = self.group_files(files) - cache = self.load_cache() - - for i, (group, group_files) in enumerate(groups.items()): - try: - if not self.process_group(group, group_files, cache): - failed.append(group) - # Add a small delay between groups to respect RPM limits if not at the last group - if len(groups) > 1 and i < len(groups) - 1: - import time - - time.sleep( - 15 - ) # Conservative 15s delay to stay within 4 RPM (Free Tier) - except KeyboardInterrupt: - self.logger.warning("\nOperation interrupted by user.") - raise - else: - self.logger.info("No changes to commit, checking push...") + failed = self._process_groups(files) if self.push and not self.dry_run: if not failed: - self.push_commits() + self.git.push() + self.console.print("[bold green]Push successful.[/bold green]") else: self.logger.error("Push aborted due to failed commits.") if self.dry_run: self.console.print( - f"[bold green]Dry run done.[/bold green] See {self.dry_run_path}" + f"[bold green]Dry run done.[/bold green] See {self.cache.path}" ) else: self.console.print("[bold green]Done.[/bold green]") if failed: self.console.print(f"[bold red]Failed groups: {failed}[/bold red]") + # ------------------------------------------------------------------ groups -def run_commit_engine(**kwargs): - """Entry point for the commit engine.""" + def _process_groups(self, files: List[str]) -> list[str]: + groups = self.grouper.group(files) + cache_map = self.cache.load( + dry_run=self.dry_run, force_rebuild=self.force_rebuild + ) + failed: list[str] = [] + for i, (group, group_files) in enumerate(groups.items()): + if not self._process_group(group, group_files, cache_map): + failed.append(group) + if len(groups) > 1 and i < len(groups) - 1: + import time + + time.sleep(15) # Conservative 4 RPM guard for free-tier + return failed + + def _process_group( + self, + group: str, + files: List[str], + cache: Dict[str, str], + ) -> bool: + self.git.stage(files) + diff = self.diff_builder.build(files) + + if not diff.strip(): + self.logger.info(f"Skipping empty diff for {group}") + self.git.reset(files) + return True + + if not self.force_rebuild and group in cache: + self.logger.info(f"Using cached message for {group}") + return self._apply_message(group, cache[group], files) + + try: + manifest_context = ManifestInspector.summary() + if manifest_context: + self.logger.info("Including manifest context in prompt") + with self.console.status( + "[bold blue]Generating commit message...[/bold blue]" + ): + raw = self.message_builder.build_message( + group=group, + diff=diff, + manifest_context=manifest_context, + extra_kwargs=self.kwargs, + ) + message = sanitize_ai_commit_message(raw) + except Exception as e: + self.logger.error( + f"Failed to generate message for group {group!r}: {e}", + exc_info=True, + ) + self.git.reset(files) + return False + + if not message: + self.logger.error(f"Empty message for {group}") + self.git.reset(files) + return False + + return self._apply_message(group, message, files) + + def _apply_message( + self, + group: str, + message: str, + files: List[str], + ) -> bool: + if self.dry_run: + self.console.print( + Panel( + Markdown(message), + title=f"Dry Run: {group}", + border_style="yellow", + expand=False, + ) + ) + self.cache.append(group, message) + self.git.reset(files) + return True + + if self.check: + message = self._review_with_user(group, message, files) + if message is None: + return False + + try: + self.console.print( + Panel(Markdown(message), title="Commit Message", border_style="green") + ) + self.git.commit(message) + except Exception as e: + self.logger.error(f"Failed to process group {group}: {e}", exc_info=True) + self.git.reset(files) + return False + return True + + def _review_with_user( + self, + group: str, + message: str, + files: List[str], + ) -> Optional[str]: + """Present the message; return the final message, or None to abort.""" + self.console.print( + Panel( + Markdown(message), + title=f"Proposed Commit Message [group: {group}]", + border_style="cyan", + ) + ) + choice = questionary.select( + "How would you like to proceed?", + choices=["Confirm", "Edit", "Abort"], + default="Confirm", + ).ask() + + if not choice or choice == "Abort": + self.logger.info(f"Commit aborted by user at group {group}") + self.git.reset(files) + return None + if choice == "Edit": + edited = questionary.text( + "Edit commit message:", + multiline=True, + default=message, + ).ask() + if not edited: + self.logger.info(f"Empty edit, commit cancelled for {group}") + self.git.reset(files) + return None + return edited + return message + + +def run_commit_engine(**kwargs: Any) -> None: + """Entry point used by the CLI.""" debug = kwargs.get("debug", False) - logger = configure_logger("devgen.commit", console=debug) + logger = configure_logger("devgen.cli.commit", console=debug) try: - engine = CommitEngine(**kwargs) - engine.execute() + CommitEngine(**kwargs).execute() except KeyboardInterrupt: logger.warning("Interrupted by user. Exiting...") sys.exit(0) @@ -609,4 +389,4 @@ def run_commit_engine(**kwargs): logger.error(f"Commit engine failed: {e}", exc_info=True) -__all__ = ["run_commit_engine"] +__all__ = ["CommitEngine", "CommitEngineError", "run_commit_engine"] diff --git a/devgen/modules/diff_builder.py b/devgen/modules/diff_builder.py new file mode 100644 index 0000000..90cf7dc --- /dev/null +++ b/devgen/modules/diff_builder.py @@ -0,0 +1,220 @@ +"""Builds the prompt inputs for the commit engine. + +Three small composable classes: + +* :class:`FileGrouper` โ€” groups a flat file list into per-commit folders +* :class:`DiffBuilder` โ€” produces a (possibly truncated) staged diff +* :class:`ManifestInspector` โ€” extracts a one-line-per-file project summary +""" + +import json +import re +import subprocess +from collections import defaultdict +from pathlib import Path +from typing import Dict, List, Optional + +from devgen.utils import run_git_command + + +# Lock files and other metadata blobs whose full diff would just bloat the +# prompt. We mark them as updated and skip the actual diff. +IGNORED_DIFF_PATTERNS = ( + "uv.lock", + "package-lock.json", + "pnpm-lock.yaml", + "yarn.lock", + "composer.lock", + "Gemfile.lock", + "poetry.lock", +) + +# Default cap for a single group diff. Real-world prompts above this size +# are likely to hit the model's context window; the cap is overridable +# via ``max_diff_size`` in the config. +DEFAULT_MAX_DIFF_SIZE = 8000 + +# Manifests we read for project context, in priority order. Lock files +# are excluded โ€” they're noisy and rarely change in a way that informs +# a commit message. +_MANIFEST_FILES = ( + "pyproject.toml", + "package.json", + "requirements.txt", +) + + +class DiffBuilder: + """Produces a staged diff for a list of paths with truncation.""" + + def __init__( + self, + max_size: int = DEFAULT_MAX_DIFF_SIZE, + cwd: Optional[Path] = None, + logger=None, + ) -> None: + self.max_size = max_size + self.cwd = cwd + self.logger = logger + + def build(self, files: List[str]) -> str: + summary, to_diff = self._partition(files) + diff = self._run_staged_diff(to_diff) + full = "\n".join(summary + [diff]).strip() + if len(full) > self.max_size: + return self._truncate(full) + return full + + # --- internals -------------------------------------------------------- + + def _partition(self, files: List[str]) -> tuple[list[str], list[str]]: + summary, to_diff = [], [] + for f in files: + if any(p in f for p in IGNORED_DIFF_PATTERNS): + summary.append(f"[METADATA UPDATED] {f}") + else: + to_diff.append(f) + return summary, to_diff + + def _run_staged_diff(self, files: List[str]) -> str: + if not files: + return "" + try: + return run_git_command( + ["git", "--no-pager", "diff", "--staged", "--", *files], + cwd=self.cwd, + ) + except subprocess.CalledProcessError as e: + raise RuntimeError( + f"Git command failed: {' '.join(e.cmd)}\nError: {e.stderr.strip()}" + ) from e + + def _truncate(self, full: str) -> str: + if self.logger is not None: + self.logger.info( + f"Truncating diff from {len(full)} to {self.max_size} chars " + f"(config: max_diff_size). If you still hit token-limit errors, " + f"lower this further or split the commit into more groups." + ) + half = self.max_size // 2 + return ( + full[:half] + + "\n\n... [DIFF TRUNCATED FOR TOKEN OPTIMIZATION] ...\n\n" + + full[-half:] + ) + + +class FileGrouper: + """Groups files by parent directory, merging down to ``max_groups``.""" + + def __init__(self, max_groups: int = 5) -> None: + self.max_groups = max(1, max_groups) + + def group(self, files: List[str]) -> Dict[str, List[str]]: + groups: Dict[str, List[str]] = defaultdict(list) + for f in files: + parent = str(Path(f).parent) + key = "root" if parent == "." else parent + groups[key].append(f) + if len(groups) <= self.max_groups: + return dict(groups) + return self._merge_deepest(dict(groups)) + + def _merge_deepest(self, groups: Dict[str, List[str]]) -> Dict[str, List[str]]: + while len(groups) > self.max_groups: + candidates = [k for k in groups if k != "root"] + if not candidates: + break + deepest = max(candidates, key=lambda p: len(Path(p).parts)) + parent_path = str(Path(deepest).parent) + new_key = "root" if parent_path == "." else parent_path + groups[new_key].extend(groups.pop(deepest)) + return groups + + +class ManifestInspector: + """Extracts a compact one-line-per-file project summary. + + Dumping the first 100 lines of ``pyproject.toml``/``requirements.txt`` + is extremely wasteful โ€” most of it is build/tool config the model does + not need. We extract just the project identity and the package list + so the model can write context-aware commit messages. + """ + + @staticmethod + def summary() -> str: + lines = [ + s + for s in ( + ManifestInspector._pyproject(), + ManifestInspector._package_json(), + ManifestInspector._requirements(), + ) + if s + ] + if not lines: + return "" + return "\nProject: " + " | ".join(lines) + + # --- extractors ------------------------------------------------------- + + @staticmethod + def _pkg_name(spec: str) -> str: + return re.split(r"[><=!~\[]", spec, 1)[0].strip() + + @staticmethod + def _pyproject() -> Optional[str]: + path = Path("pyproject.toml") + if not path.exists(): + return None + try: + import toml + + data = toml.load(path) + except Exception: + return None + proj = data.get("project", {}) if isinstance(data, dict) else {} + parts = [ + f"{k}={proj[k]}" for k in ("name", "version", "description") if proj.get(k) + ] + deps = proj.get("dependencies") or [] + if deps: + names = [n for n in (ManifestInspector._pkg_name(d) for d in deps) if n] + if names: + parts.append(f"deps={','.join(names)}") + return f"[pyproject.toml] {' '.join(parts)}" if parts else None + + @staticmethod + def _package_json() -> Optional[str]: + path = Path("package.json") + if not path.exists(): + return None + try: + data = json.loads(path.read_text(encoding="utf-8")) + except Exception: + return None + parts = [ + f"{k}={data[k]}" for k in ("name", "version", "description") if data.get(k) + ] + deps = data.get("dependencies") or {} + if isinstance(deps, dict) and deps: + parts.append(f"deps={','.join(deps.keys())}") + return f"[package.json] {' '.join(parts)}" if parts else None + + @staticmethod + def _requirements() -> Optional[str]: + path = Path("requirements.txt") + if not path.exists(): + return None + try: + names = [ + n + for n in ( + ManifestInspector._pkg_name(line) + for line in path.read_text(encoding="utf-8").splitlines() + ) + if n and not n.startswith("#") + ] + except Exception: + return None + return f"[requirements.txt] deps={','.join(names)}" if names else None diff --git a/devgen/modules/git_ops.py b/devgen/modules/git_ops.py new file mode 100644 index 0000000..455e866 --- /dev/null +++ b/devgen/modules/git_ops.py @@ -0,0 +1,144 @@ +"""Encapsulates every ``git`` invocation the commit engine makes. + +Wraps :func:`devgen.utils.run_git_command` with the project's preferred +error handling and turns raw CLI output into typed results. Higher-level +classes (:class:`CommitEngine`) depend on this instead of calling git +themselves, so the engine code stays focused on orchestration. +""" + +import subprocess +from pathlib import Path +from typing import Optional + +from devgen.utils import configure_logger, run_git_command + + +class GitError(RuntimeError): + """Raised when a git command exits non-zero.""" + + +class GitOperator: + """All git side-effects in one place.""" + + def __init__( + self, + cwd: Optional[Path] = None, + logger=None, + quiet_commands: bool = True, + ) -> None: + self.cwd = cwd + self.logger = logger or configure_logger("devgen.git") + self.quiet_commands = quiet_commands + + # ------------------------------------------------------------------ helpers + + def _run(self, args, *, check: bool = True) -> str: + try: + return run_git_command(args, check=check, cwd=self.cwd) + except subprocess.CalledProcessError as e: + cmd_str = " ".join(e.cmd) if e.cmd else "git" + stderr = (e.stderr or "").strip() + raise GitError(f"Git command failed: {cmd_str}\nError: {stderr}") from e + + def _format_error(self, e: subprocess.CalledProcessError) -> str: + cmd_str = " ".join(e.cmd) if e.cmd else "git" + return f"Git command failed: {cmd_str}\nError: {(e.stderr or '').strip()}" + + # ------------------------------------------------------------------ queries + + def has_upstream(self) -> bool: + """Return True if the current branch tracks an upstream.""" + return bool(self.upstream()) + + def upstream(self) -> str: + """Return the upstream branch name, or empty string if none.""" + return self._run( + [ + "git", + "rev-parse", + "--abbrev-ref", + "--symbolic-full-name", + "@{u}", + ], + check=False, + ) + + def is_ahead_of_remote(self, fetch: bool = False) -> bool: + """Return True if local HEAD has unpushed commits. + + When ``fetch`` is True, refresh the upstream ref first โ€” but only + do that when the caller is about to push. Skipping the fetch by + default avoids a network round-trip on every run and shrinks the + TOCTOU window between fetch and rev-list. + """ + upstream = self.upstream() + if not upstream: + return False + if fetch: + try: + self._run(["git", "fetch", "origin", upstream], check=False) + except GitError: + pass + try: + count = self._run(["git", "rev-list", "--count", "@{u}..HEAD"], check=False) + return bool(count) and int(count) > 0 + except (GitError, ValueError): + return False + + def detect_changed(self) -> list[str]: + """Return modified, deleted, and untracked files (deduplicated).""" + out = self._run( + [ + "git", + "ls-files", + "--deleted", + "--modified", + "--others", + "--exclude-standard", + ] + ) + files = {f.strip() for f in out.split("\n") if f.strip()} + staged_out = self._run(["git", "diff", "--name-only", "--cached"]) + files.update(f.strip() for f in staged_out.split("\n") if f.strip()) + return sorted(files) + + # ------------------------------------------------------------------ actions + + def stage(self, files: list[str]) -> None: + """``git add`` the given paths.""" + if not files: + return + try: + self._run(["git", "add", *files]) + except GitError as e: + raise GitError(f"Could not stage files: {e}") from e + + def commit(self, message: str) -> None: + """``git commit -m`` with the given message.""" + self._run(["git", "commit", "-m", message]) + + def reset(self, files: list[str]) -> None: + """Unstage the given paths. Errors are swallowed โ€” best-effort.""" + if not files: + return + try: + self._run(["git", "reset", "HEAD", "--", *files], check=False) + except GitError: + pass + + def push(self) -> None: + """Fetch upstream then push. Falls back to a no-upstream warning.""" + upstream = self.upstream() + if upstream: + try: + self._run(["git", "fetch", "origin", upstream], check=False) + except GitError: + pass + try: + self._run(["git", "push"]) + except GitError as e: + if "no upstream branch" in str(e).lower(): + if not self.quiet_commands: + print("[warning]No upstream branch. Skipping push.[/warning]") + return + raise diff --git a/devgen/modules/release_note_generator.py b/devgen/modules/release_note_generator.py index 80948d7..2487fe9 100644 --- a/devgen/modules/release_note_generator.py +++ b/devgen/modules/release_note_generator.py @@ -2,71 +2,55 @@ from datetime import datetime from pathlib import Path from devgen.modules.changelog_generator import ChangelogGenerator +from devgen.modules.changelog_sections import Section from devgen.utils import configure_logger +# Release-note-specific heading styles (use the same emoji, simpler heading). +RELEASE_NOTE_STYLES = { + Section.BREAKING: ("โš ๏ธ", "BREAKING CHANGES"), + Section.FEATURES: ("โœจ", "Features"), + Section.BUG_FIXES: ("๐Ÿ›", "Bug Fixes"), + Section.PERFORMANCE: ("โšก", "Performance"), + Section.DOCUMENTATION: ("๐Ÿ“š", "Documentation"), + Section.REFACTOR: ("โ™ป๏ธ", "Refactor"), + Section.TESTS: ("โœ…", "Tests"), + Section.STYLE: ("๐Ÿ’„", "Style"), + Section.CHORE: ("๐Ÿ”ง", "Chore"), + Section.OTHER: ("๐Ÿงน", "Other Changes"), +} + + class ReleaseNotesGenerator(ChangelogGenerator): """Generate short, clean release notes from git history.""" - SECTION_EMOJIS = { - "BREAKING CHANGES": "โš ๏ธ", - "Features": "โœจ", - "Bug Fixes": "๐Ÿ›", - "Performance": "โšก", - "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.""" + """Render a human-friendly release-notes block.""" date_str = datetime.now().strftime("%Y-%m-%d") - md = [f"## ๐Ÿš€ Release {version} โ€” {date_str}\n"] - - order = [ - "BREAKING CHANGES", - "Features", - "Bug Fixes", - "Performance", - "Documentation", - "Refactor", - "Tests", - "Style", - "Chore", - "Other Changes", - ] - - for section in order: + out = [f"## ๐Ÿš€ Release {version} โ€” {date_str}\n"] + for section in Section.ordered(): commits = groups.get(section) if not commits: continue - - emoji = self.SECTION_EMOJIS.get(section, "") - md.append(f"### {emoji} {section}") - + emoji, heading = RELEASE_NOTE_STYLES[section] + out.append(f"### {emoji} {heading}") for c in commits: - scope = f"**{c['scope']}**: " if c["scope"] else "" - md.append(f"- {scope}{c['subject']}") - - md.append("") - - return "\n".join(md) + scope = f"**{c['scope']}**: " if c.get("scope") else "" + out.append(f"- {scope}{c['subject']}") + out.append("") + return "\n".join(out) def run(self, output_file="RELEASE-NOTES.md", version="Unreleased", from_ref=""): - """Main logic: get commits โ†’ parse โ†’ generate โ†’ write.""" + """Fetch โ†’ parse โ†’ write release notes.""" 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) - + groups = self.parse_commits(raw_commits) + md = self.generate_release_markdown(groups, version=version) Path(output_file).write_text(md, encoding="utf-8") print(f" Release notes written to {output_file}")