sankalp
f7001d4243
Add token‑limit error handling and switch git diff to null‑delimited output for reliable file detection.
223 lines
7.3 KiB
Python
223 lines
7.3 KiB
Python
"""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 get_git_root, 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 or get_git_root()
|
|
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 Path(f).name 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
|
|
files = groups.pop(deepest)
|
|
groups.setdefault(new_key, []).extend(files)
|
|
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(cwd: Optional[Path] = None) -> str:
|
|
cwd = cwd or get_git_root() or Path(".")
|
|
lines = [
|
|
s
|
|
for s in (
|
|
ManifestInspector._pyproject(cwd),
|
|
ManifestInspector._package_json(cwd),
|
|
ManifestInspector._requirements(cwd),
|
|
)
|
|
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(cwd: Path) -> Optional[str]:
|
|
path = cwd / "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(cwd: Path) -> Optional[str]:
|
|
path = cwd / "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(cwd: Path) -> Optional[str]:
|
|
path = cwd / "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
|