local-commit-message-generator/src/generator.py

"""Commit message generation logic."""

from pathlib import Path
from typing import Optional, Tuple

from .analyzer import ChangeAnalyzer, ChangeSet, ChangeType
from .config import get_type_rules, load_config
from .templates import format_message


class GenerationError(Exception):
    """Raised when commit message generation fails."""
    pass


def detect_commit_type(
    change_set: ChangeSet,
    type_rules: Optional[dict] = None
) -> str:
    """Detect the commit type based on staged changes.

    Args:
        change_set: Collection of staged changes.
        type_rules: Optional custom type rules dictionary.

    Returns:
        Detected commit type string.
    """
    if type_rules is None:
        type_rules = get_type_rules()

    scores = {type_: 0 for type_ in type_rules.keys()}
    scores["chore"] = 0

    for change in change_set.changes:
        path = change.path.lower()
        for type_, patterns in type_rules.items():
            for pattern in patterns:
                if pattern.startswith("."):
                    if path.endswith(pattern):
                        scores[type_] += 1
                elif "/" in pattern or pattern.startswith("."):
                    if pattern in path or path.startswith(pattern):
                        scores[type_] += 1
                else:
                    if pattern in path:
                        scores[type_] += 1

    scores["chore"] = sum(1 for c in change_set.changes if c.change_type in [
        ChangeType.DELETED, ChangeType.TYPE_CHANGE
    ])

    if not scores or all(v == 0 for v in scores.values()):
        return "chore"

    return max(scores, key=scores.get)


def detect_scope(
    change_set: ChangeSet,
    scope_overrides: Optional[dict] = None
) -> str:
    """Detect the commit scope based on staged changes.

    Args:
        change_set: Collection of staged changes.
        scope_overrides: Optional custom scope mapping.

    Returns:
        Detected scope string or empty string.
    """
    if not change_set.has_changes:
        return ""

    if scope_overrides is None:
        config = load_config()
        scope_overrides = config.get("scopes", {})

    paths = [c.path for c in change_set.changes]

    common_parts = []
    for parts in (p.split("/") for p in paths):
        if not common_parts:
            common_parts = parts
        else:
            common_parts = [
                common_parts[i] if (
                    i < len(common_parts) and
                    i < len(parts) and
                    common_parts[i] == parts[i]
                ) else None
                for i in range(min(len(common_parts), len(parts)))
            ]
            common_parts = [p for p in common_parts if p is not None]

    if len(common_parts) <= 1:
        candidates = set()
        for path in paths:
            parts = path.split("/")
            if len(parts) > 1:
                candidates.add(parts[0])

        for path in paths:
            for scope, match in scope_overrides.items():
                if match in path:
                    return scope

        if len(candidates) == 1:
            return candidates.pop()

        if len(candidates) > 1:
            return ",".join(sorted(candidates)[:2])

        return ""

    scope = common_parts[0] if common_parts else ""
    return scope


def generate_description(change_set: ChangeSet) -> str:
    """Generate a description based on changes.

    Args:
        change_set: Collection of staged changes.

    Returns:
        Generated description string.
    """
    if not change_set.has_changes:
        return "update"

    added = change_set.added
    deleted = change_set.deleted
    modified = change_set.modified

    if deleted:
        return f"remove {Path(deleted[0].path).name}"
    elif added and len(added) == 1:
        return f"add {Path(added[0].path).name}"
    elif modified and len(modified) == 1:
        return f"update {Path(modified[0].path).name}"
    elif added:
        return f"add {len(added)} files"
    elif modified:
        return f"update {len(modified)} files"
    else:
        return "update"


def generate_commit_message(
    repo_path: Optional[str] = None,
    config_path: Optional[str] = None
) -> str:
    """Generate a conventional commit message.

    Args:
        repo_path: Optional path to git repository.
        config_path: Optional path to configuration file.

    Returns:
        Generated commit message string.

    Raises:
        GenerationError: If generation fails.
    """
    try:
        analyzer = ChangeAnalyzer(repo_path)
        change_set = analyzer.get_staged_changes()

        if not change_set.has_changes:
            raise GenerationError("No staged changes found. Run 'git add' first.")

        config = load_config() if config_path is None else load_config(config_path)
        type_rules = config.get("type_rules", {})
        template = config.get("template")
        scope_overrides = config.get("scopes", {})

        commit_type = detect_commit_type(change_set, type_rules)
        scope = detect_scope(change_set, scope_overrides)
        description = generate_description(change_set)

        files = change_set.file_paths if config.get("include_file_list", True) else None
        max_files = config.get("max_files", 5)
        if files and len(files) > max_files:
            files = files[:max_files]

        message = format_message(
            type=commit_type,
            scope=scope,
            description=description,
            template=template,
            files=files
        )

        return message

    except ValueError as e:
        raise GenerationError(str(e))


def get_commit_message_preview(
    repo_path: Optional[str] = None
) -> Tuple[str, bool]:
    """Get a preview of the commit message.

    Args:
        repo_path: Optional path to git repository.

    Returns:
        Tuple of (message, has_changes).
    """
    try:
        analyzer = ChangeAnalyzer(repo_path)
        change_set = analyzer.get_staged_changes()

        if not change_set.has_changes:
            return ("", False)

        message = generate_commit_message(repo_path)
        return (message, True)

    except GenerationError:
        return ("", False)