sankalp
957dda9d49
- Replace jinja2 with custom render_template() helper
- Support {{ var }} substitution and {% if %}...{% endif %} conditionals
- Remove jinja2 from pyproject.toml and requirements.txt
- Simplifies dependency tree and reduces install size
406 lines
12 KiB
Python
406 lines
12 KiB
Python
import logging
|
|
import re
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
import yaml
|
|
|
|
|
|
def ensure_log_directory() -> Path:
|
|
log_dir = Path.home() / ".cache" / "devgen"
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
return log_dir
|
|
|
|
|
|
def get_main_log_path() -> Path:
|
|
return ensure_log_directory() / "devgen.log"
|
|
|
|
|
|
def get_commit_dry_run_path() -> Path:
|
|
return ensure_log_directory() / "commit_dry_run.md"
|
|
|
|
|
|
def is_file_recent(file_path: Path | str, max_age_minutes: int = 120) -> bool:
|
|
try:
|
|
return (time.time() - Path(file_path).stat().st_mtime) <= max_age_minutes * 60
|
|
except FileNotFoundError:
|
|
return False
|
|
|
|
|
|
def sanitize_ai_commit_message(raw_text: str) -> str:
|
|
"""
|
|
Cleans up the AI-generated commit message.
|
|
It looks for a conventional commit header and extracts everything from there.
|
|
It handles bolded headers and common AI prefixes.
|
|
"""
|
|
if not raw_text:
|
|
return ""
|
|
|
|
lines = raw_text.strip().split("\n")
|
|
|
|
# Regex for conventional commit header (including optional bolding)
|
|
# Examples:
|
|
# - feat(root): summary
|
|
# - **fix: bug fix**
|
|
# - chore(deps)!: breaking change
|
|
header_pattern = re.compile(
|
|
r"^(\*\*)?(feat|fix|chore|refactor|docs|style|test|build|ci)(\(.*\))?!?: .*",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
cleaned_lines = []
|
|
found_header = False
|
|
|
|
for line in lines:
|
|
stripped = line.strip()
|
|
if not found_header:
|
|
# First, strip bolding if it wraps the whole line or parts of it
|
|
# This is simpler than handling it in the regex
|
|
candidate = stripped.replace("**", "").strip()
|
|
|
|
# Clean the line by removing labels like "1. Title:", "### Message:", etc.
|
|
clean_line = re.sub(
|
|
r"^(#+\s+)?(\d+\.?|\*|-)?\s*(Title|Commit Message|Message):\s*",
|
|
"",
|
|
candidate,
|
|
flags=re.IGNORECASE,
|
|
).strip()
|
|
|
|
if header_pattern.match(clean_line):
|
|
found_header = True
|
|
cleaned_lines.append(clean_line)
|
|
else:
|
|
# If we hit another header or a known separator, we stop
|
|
if "**Sponsor**" in line:
|
|
break
|
|
cleaned_lines.append(line)
|
|
|
|
if cleaned_lines:
|
|
return "\n".join(cleaned_lines).strip()
|
|
|
|
# Fallback: if no conventional commit header found, just take the first non-empty line
|
|
# to avoid failing completely, but log a warning if possible.
|
|
for line in lines:
|
|
if line.strip():
|
|
return line.strip()
|
|
|
|
return ""
|
|
|
|
|
|
def parse_markdown_sections(
|
|
filepath: Path | str, marker_pattern: str
|
|
) -> dict[str, str]:
|
|
path = Path(filepath)
|
|
if not path.exists():
|
|
return {}
|
|
|
|
with path.open(encoding="utf-8") as f:
|
|
content = f.read()
|
|
|
|
results = {}
|
|
matches = re.findall(marker_pattern, content, re.DOTALL)
|
|
for key, value in matches:
|
|
results[key] = value.strip()
|
|
return results
|
|
|
|
|
|
def extract_commit_messages(filepath: Path | str) -> dict[str, str]:
|
|
pattern = r"## Group: `(.*?)`\s*```md\n(.*?)\n```"
|
|
return parse_markdown_sections(filepath, pattern)
|
|
|
|
|
|
def configure_logger(
|
|
name: str = "devgen", log_file: Optional[Path | str] = None, console: bool = True
|
|
) -> logging.Logger:
|
|
logger = logging.getLogger(name)
|
|
logger.setLevel(logging.INFO)
|
|
|
|
if logger.hasHandlers():
|
|
logger.handlers.clear()
|
|
|
|
formatter = logging.Formatter(
|
|
"%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
|
)
|
|
|
|
# Console handler
|
|
if console:
|
|
ch = logging.StreamHandler()
|
|
ch.setFormatter(formatter)
|
|
logger.addHandler(ch)
|
|
|
|
# File handler
|
|
if log_file:
|
|
path = Path(log_file)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fh = logging.FileHandler(path, mode="w", encoding="utf-8")
|
|
fh.setFormatter(formatter)
|
|
logger.addHandler(fh)
|
|
|
|
return logger
|
|
|
|
|
|
def get_git_root(cwd: Optional[Path] = None) -> Optional[Path]:
|
|
"""Return the root directory of the current git repository, or None."""
|
|
try:
|
|
output = run_git_command(["git", "rev-parse", "--show-toplevel"], cwd=cwd)
|
|
return Path(output) if output else None
|
|
except (subprocess.CalledProcessError, FileNotFoundError):
|
|
return None
|
|
|
|
|
|
def run_git_command(
|
|
args: list[str],
|
|
check: bool = True,
|
|
cwd: Optional[Path] = None,
|
|
encoding: str = "utf-8",
|
|
errors: str = "replace",
|
|
) -> str:
|
|
"""
|
|
Executes a git command and returns the output.
|
|
|
|
Args:
|
|
args: List of command arguments (e.g., ["git", "status"]).
|
|
check: Whether to raise an exception on non-zero exit code.
|
|
cwd: Current working directory for the command.
|
|
encoding: Output encoding.
|
|
errors: Error handling strategy for encoding.
|
|
|
|
Returns:
|
|
The standard output of the command, stripped of leading/trailing whitespace.
|
|
|
|
Raises:
|
|
subprocess.CalledProcessError: If the command fails and check is True.
|
|
"""
|
|
try:
|
|
res = subprocess.run(
|
|
args,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding=encoding,
|
|
errors=errors,
|
|
check=check,
|
|
cwd=cwd,
|
|
)
|
|
return res.stdout.strip()
|
|
except subprocess.CalledProcessError as e:
|
|
# Log the error if a logger is configured, but re-raise
|
|
# We don't have access to a specific logger here easily without passing it in,
|
|
# so we rely on the caller to handle logging if needed, or we could log to a default one.
|
|
# For now, just re-raise as the caller expects.
|
|
raise e
|
|
|
|
|
|
def get_git_staged_files() -> list[str]:
|
|
try:
|
|
output = run_git_command(["git", "diff", "--name-only", "--cached"])
|
|
return [f for f in output.splitlines() if f.strip()]
|
|
except subprocess.CalledProcessError:
|
|
return []
|
|
|
|
|
|
def read_file_content(filepath: Path | str) -> Optional[str]:
|
|
try:
|
|
return Path(filepath).read_text(encoding="utf-8")
|
|
except FileNotFoundError:
|
|
return None
|
|
|
|
|
|
def delete_file(filepath: Path | str) -> bool:
|
|
try:
|
|
Path(filepath).unlink()
|
|
return True
|
|
except FileNotFoundError:
|
|
return False
|
|
|
|
|
|
def load_template(sub_dir: str, name: str) -> str:
|
|
"""Load a template file from the templates directory."""
|
|
template_dir = Path(__file__).parent / "templates" / sub_dir
|
|
template_path = template_dir / name
|
|
if not template_path.is_file():
|
|
raise RuntimeError(
|
|
f"Template not found: {template_path}. "
|
|
"The installation may be corrupted."
|
|
)
|
|
return template_path.read_text(encoding="utf-8")
|
|
|
|
|
|
def render_template(template_str: str, **context) -> str:
|
|
"""Render a simple template with variable substitution and conditionals.
|
|
|
|
Supports:
|
|
- {{ variable }} - variable substitution
|
|
- {% if condition %}...{% endif %} - conditional blocks
|
|
- {% if condition %}...{% else %}...{% endif %} - conditional with else
|
|
"""
|
|
result = template_str
|
|
|
|
# Process conditional blocks: {% if var %}...{% endif %}
|
|
# and {% if var %}...{% else %}...{% endif %}
|
|
if_pattern = re.compile(
|
|
r"\{%\s*if\s+(\w+)\s*%\}(.*?)\{%\s*endif\s*%\}",
|
|
re.DOTALL,
|
|
)
|
|
|
|
def replace_if(match):
|
|
var_name = match.group(1)
|
|
content = match.group(2)
|
|
value = context.get(var_name)
|
|
if value:
|
|
# Check for else block
|
|
else_pattern = re.compile(r"\{%\s*else\s*%\}", re.DOTALL)
|
|
parts = else_pattern.split(content, 1)
|
|
return parts[0]
|
|
else:
|
|
else_pattern = re.compile(r"\{%\s*else\s*%\}", re.DOTALL)
|
|
parts = else_pattern.split(content, 1)
|
|
return parts[1] if len(parts) > 1 else ""
|
|
|
|
result = if_pattern.sub(replace_if, result)
|
|
|
|
# Process variable substitutions: {{ variable }}
|
|
var_pattern = re.compile(r"\{\{\s*(\w+)\s*\}\}")
|
|
|
|
def replace_var(match):
|
|
var_name = match.group(1)
|
|
return str(context.get(var_name, ""))
|
|
|
|
result = var_pattern.sub(replace_var, result)
|
|
|
|
return result
|
|
|
|
|
|
def load_config() -> Dict[str, Any]:
|
|
config_path = Path.home() / ".devgen.yaml"
|
|
|
|
if not config_path.exists():
|
|
default_config = {
|
|
"provider": "gemini",
|
|
"model": "gemini-2.5-flash",
|
|
"api_key": "",
|
|
"emoji": True,
|
|
"max_groups": 5,
|
|
"ollama_host": "http://localhost:11434",
|
|
}
|
|
try:
|
|
with config_path.open("w", encoding="utf-8") as f:
|
|
yaml.dump(default_config, f, default_flow_style=False)
|
|
# We don't print here to avoid noise during normal execution
|
|
except Exception as e:
|
|
print(f"Warning: Failed to create default config at {config_path}: {e}")
|
|
return {}
|
|
|
|
try:
|
|
with config_path.open("r", encoding="utf-8") as f:
|
|
data = yaml.safe_load(f)
|
|
if not isinstance(data, dict):
|
|
print(
|
|
f"Warning: Config at {config_path} is not a dict, using defaults."
|
|
)
|
|
return {}
|
|
return data
|
|
except Exception as e:
|
|
print(f"Warning: Failed to load config from {config_path}: {e}")
|
|
return {}
|
|
|
|
|
|
# Substrings used to identify token / context-length errors from AI providers.
|
|
# Matched case-insensitively against the full stringified exception, so we
|
|
# catch the wide variety of error formats each provider returns.
|
|
_TOKEN_LIMIT_PATTERNS = (
|
|
"context length",
|
|
"context_length",
|
|
"context_length_exceeded",
|
|
"context window",
|
|
"context window exceeded",
|
|
"context_size",
|
|
"maximum context",
|
|
"max context",
|
|
"max_tokens",
|
|
"max tokens",
|
|
"prompt is too long",
|
|
"too many tokens",
|
|
"token limit",
|
|
"tokens in the prompt",
|
|
"input is too long",
|
|
"request too large",
|
|
"request was too large",
|
|
"string too long",
|
|
"too long for the model",
|
|
"reduce the length of the messages",
|
|
"tokens_exceeded",
|
|
)
|
|
|
|
|
|
def is_token_limit_error(error: BaseException | str) -> bool:
|
|
"""Return True if the error looks like a model context/token overflow.
|
|
|
|
Different SDKs phrase the same condition differently. We look for the
|
|
common substrings so we can give a single, actionable message instead of
|
|
the raw provider text.
|
|
"""
|
|
text = str(error).lower()
|
|
return any(p in text for p in _TOKEN_LIMIT_PATTERNS)
|
|
|
|
|
|
def format_token_limit_error(
|
|
provider: str,
|
|
error: BaseException | str,
|
|
*,
|
|
group: str | None = None,
|
|
) -> str:
|
|
"""Build a friendly message for a context-window overflow.
|
|
|
|
Kept short so it doesn't drown the log line. The original provider
|
|
error is included so the user can still see the exact cause.
|
|
"""
|
|
scope = f" ({group})" if group else ""
|
|
return (
|
|
f"{provider}: diff exceeds context window{scope}. "
|
|
f"Cut token usage: --max-groups N, --max-diff-size N, "
|
|
f"or a larger-context model (gemini-2.5-flash, gpt-4o, claude-3-5-sonnet). "
|
|
f"Details: {error}"
|
|
)
|
|
|
|
|
|
__all__ = [
|
|
"ensure_log_directory",
|
|
"get_main_log_path",
|
|
"get_commit_dry_run_path",
|
|
"is_file_recent",
|
|
"sanitize_ai_commit_message",
|
|
"extract_commit_messages",
|
|
"configure_logger",
|
|
"run_git_command",
|
|
"get_git_staged_files",
|
|
"read_file_content",
|
|
"delete_file",
|
|
"load_template",
|
|
"render_template",
|
|
"load_config",
|
|
"get_questionary_style",
|
|
"is_token_limit_error",
|
|
"format_token_limit_error",
|
|
]
|
|
|
|
|
|
def get_questionary_style():
|
|
from questionary import Style
|
|
|
|
return Style(
|
|
[
|
|
("qmark", "fg:#673ab7 bold"), # Token.QuestionMark
|
|
("question", "bold"), # Token.Question
|
|
("answer", "fg:#f44336 bold"), # Token.Answer
|
|
("pointer", "fg:#673ab7 bold"), # Token.Pointer
|
|
("highlighted", "fg:#673ab7 bold"), # Token.Selected
|
|
("selected", "fg:#cc5454"), # Token.SelectedItem
|
|
("separator", "fg:#cc5454"), # Token.Separator
|
|
("instruction", ""), # Token.Instruction
|
|
("text", ""), # Token.Text
|
|
("disabled", "fg:#858585 italic"), # Token.Disabled
|
|
]
|
|
)
|