Coverage for src / rhiza / commands / summarise / _gather.py: 100%
113 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-07-16 12:51 +0000
1"""Data-gathering helpers for the ``summarise`` command.
3Collects staged git changes, categorises files by type, and reads template
4metadata from ``template.lock`` / ``template.yml``. This module performs no
5rendering — see :mod:`rhiza.commands.summarise._render` for output formatting.
6"""
8import subprocess # nosec B404
9from collections import defaultdict
10from pathlib import Path
11from typing import NamedTuple
13import yaml
14from loguru import logger
16from rhiza.models.lock import TemplateLock
17from rhiza.models.template import RhizaTemplate
20class _TemplateInfo(NamedTuple):
21 """Lightweight container for template metadata used during rendering."""
23 repo: str
24 branch: str
25 last_sync: str | None
28def run_git_command(args: list[str], cwd: Path | None = None) -> str:
29 """Run a git command and return the output.
31 Args:
32 args: Git command arguments (without 'git' prefix)
33 cwd: Working directory for the command
35 Returns:
36 Command output as string
37 """
38 try:
39 result = subprocess.run( # nosec B603 B607 # noqa: S603
40 ["git", *args], # noqa: S607
41 cwd=cwd,
42 capture_output=True,
43 text=True,
44 check=True,
45 )
46 return result.stdout.strip()
47 except subprocess.CalledProcessError as e:
48 logger.error(f"Error running git {' '.join(args)}: {e.stderr}")
49 return ""
52def get_staged_changes(repo_path: Path, compare_ref: str | None = None) -> dict[str, list[str]]:
53 """Get list of changes categorized by type.
55 Args:
56 repo_path: Path to the repository
57 compare_ref: Optional git ref to compare against. When provided the
58 working tree is diffed against this ref instead of the staged index.
60 Returns:
61 Dictionary with keys 'added', 'modified', 'deleted' containing file lists
62 """
63 changes: dict[str, list[str]] = {
64 "added": [],
65 "modified": [],
66 "deleted": [],
67 }
69 # Compare against a specific ref, or fall back to staged changes
70 diff_args = ["diff", compare_ref, "--name-status"] if compare_ref else ["diff", "--cached", "--name-status"]
72 output = run_git_command(diff_args, cwd=repo_path)
74 for line in output.split("\n"):
75 if not line:
76 continue
77 parts = line.split("\t", 1)
78 if len(parts) != 2:
79 continue
80 status, filepath = parts
82 if status == "A":
83 changes["added"].append(filepath)
84 elif status == "M":
85 changes["modified"].append(filepath)
86 elif status == "D":
87 changes["deleted"].append(filepath)
88 elif status.startswith("R"):
89 # Renamed file - treat as modified
90 changes["modified"].append(filepath)
92 return changes
95_CONFIG_FILES: frozenset[str] = frozenset(
96 {
97 "Makefile",
98 "ruff.toml",
99 "pytest.ini",
100 ".editorconfig",
101 ".gitignore",
102 ".pre-commit-config.yaml",
103 "renovate.json",
104 ".python-version",
105 }
106)
109_DIR_CATEGORIES: dict[str, str] = {
110 "tests": "Tests",
111 "src": "Source Code",
112}
113_DOC_DIRS: frozenset[str] = frozenset({"book", "docs"})
116def _categorize_by_directory(first_dir: str, filepath: str) -> str | None:
117 """Categorize file based on its first directory.
119 Args:
120 first_dir: First directory in the path
121 filepath: Full file path
123 Returns:
124 Category name or None if no match
125 """
126 if first_dir == ".github":
127 path_parts = Path(filepath).parts
128 if len(path_parts) > 1 and path_parts[1] == "workflows":
129 return "GitHub Actions Workflows"
130 return "GitHub Configuration"
132 if first_dir == ".rhiza":
133 if "script" in filepath.lower():
134 return "Rhiza Scripts"
135 if "Makefile" in filepath:
136 return "Makefiles"
137 return "Rhiza Configuration"
139 if first_dir in _DIR_CATEGORIES:
140 return _DIR_CATEGORIES[first_dir]
142 if first_dir in _DOC_DIRS:
143 return "Documentation"
145 return None
148def _categorize_single_file(filepath: str) -> str:
149 """Categorize a single file path.
151 Args:
152 filepath: File path to categorize
154 Returns:
155 Category name
156 """
157 path_parts = Path(filepath).parts
159 if not path_parts:
160 return "Other"
162 # Try directory-based categorization first
163 category = _categorize_by_directory(path_parts[0], filepath)
164 if category:
165 return category
167 # Check file-based categories
168 if filepath.endswith(".md"):
169 return "Documentation"
171 if filepath in _CONFIG_FILES:
172 return "Configuration Files"
174 return "Other"
177def categorize_files(files: list[str]) -> dict[str, list[str]]:
178 """Categorize files by type.
180 Args:
181 files: List of file paths
183 Returns:
184 Dictionary mapping category names to file lists
185 """
186 categories = defaultdict(list)
188 for filepath in files:
189 category = _categorize_single_file(filepath)
190 categories[category].append(filepath)
192 return dict(categories)
195def get_template_info(repo_path: Path) -> tuple[str, str]:
196 """Get template repository and branch from template.lock or template.yml.
198 Prefers ``template.lock`` as the authoritative record of the last sync.
199 Falls back to ``template.yml`` if the lock file is absent or incomplete.
200 Returns empty strings when no configuration is found, rather than
201 defaulting to any hardcoded repository name.
203 Args:
204 repo_path: Path to the repository
206 Returns:
207 Tuple of (template_repo, template_branch)
208 """
209 # Prefer template.lock - it is the authoritative record of what was synced
210 lock_file = repo_path / ".rhiza" / "template.lock"
211 if lock_file.exists():
212 try:
213 lock = TemplateLock.from_yaml(lock_file)
214 if lock.repo:
215 return lock.repo, lock.ref
216 except (yaml.YAMLError, ValueError, TypeError, KeyError):
217 logger.warning("Failed to read template.lock; falling back to template.yml")
219 # Fall back to template.yml, using the proper model which handles both
220 # 'template-repository'/'repository' and 'template-branch'/'ref' key variants
221 template_file = repo_path / ".rhiza" / "template.yml"
222 if not template_file.exists():
223 return ("", "")
225 try:
226 template = RhizaTemplate.from_yaml(template_file)
227 except (yaml.YAMLError, ValueError, TypeError, KeyError):
228 logger.warning("Failed to read template.yml")
229 return ("", "")
231 return template.template_repository, template.template_branch
234def get_last_sync_date(repo_path: Path, template_repo: str = "") -> str | None:
235 """Get the date of the last sync.
237 Checks ``template.lock`` for a recorded sync timestamp first, then falls
238 back to searching the git log. The template repository name (when given)
239 is used to build more accurate grep patterns so that projects using a
240 non-rhiza template are still matched correctly.
242 Args:
243 repo_path: Path to the repository
244 template_repo: Template repository name (e.g. ``"my-org/my-template"``)
245 used to derive the short name for git-log grep patterns.
247 Returns:
248 ISO format date string or None if not found
249 """
250 # Prefer template.lock synced_at - it is the most reliable source
251 lock_file = repo_path / ".rhiza" / "template.lock"
252 if lock_file.exists():
253 try:
254 lock = TemplateLock.from_yaml(lock_file)
255 if lock.synced_at:
256 return lock.synced_at
257 except (yaml.YAMLError, ValueError, TypeError, KeyError):
258 pass
260 # Derive the short name from the template repo for targeted grepping
261 template_short_name = template_repo.rsplit("/", 1)[-1] if template_repo else ""
263 grep_args = ["log", "--format=%cI", "-1"]
264 if template_short_name:
265 grep_args.extend(["--grep", template_short_name])
266 grep_args.extend(["--grep=Sync", "--grep=template", "-i"])
268 output = run_git_command(grep_args, cwd=repo_path)
269 if output:
270 return output
272 return None