Coverage for little_loops / cli / issues / search.py: 6%
232 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-04-11 23:20 -0500
1"""ll-issues search: Search issues with filters and sorting."""
3from __future__ import annotations
5import argparse
6import re
7from datetime import date
8from typing import TYPE_CHECKING
10from little_loops.cli.output import PRIORITY_COLOR, TYPE_COLOR, colorize, print_json
12if TYPE_CHECKING:
13 from pathlib import Path
15 from little_loops.config import BRConfig
16 from little_loops.issue_parser import IssueInfo
19def _parse_discovered_date(content: str) -> date | None:
20 """Extract discovered_date from YAML frontmatter."""
21 match = re.match(r"^---\n(.*?)\n---", content, re.DOTALL)
22 if not match:
23 return None
24 date_match = re.search(r"discovered_date:\s*(\S+)", match.group(1))
25 if not date_match:
26 return None
27 date_str = date_match.group(1).strip("\"'")
28 try:
29 # Handle ISO datetime strings like 2026-03-13T21:11:34Z
30 return date.fromisoformat(date_str[:10])
31 except ValueError:
32 return None
35def _parse_updated_date(content: str, file_path: Path) -> date | None:
36 r"""Extract last-activity date from the ## Session Log section, or fall back to file mtime.
38 Scans for the most recent timestamp entry in the Session Log section.
39 Entry format: `- \`/ll:cmd\` - YYYY-MM-DDTHH:MM:SS - \`path\``
40 Falls back to file mtime when no session log timestamps are found.
41 """
42 import re as _re
44 _SESSION_LOG_RE = _re.compile(
45 r"^## Session Log\s*\n+(.*?)(?:\n##|\n---|\Z)", _re.MULTILINE | _re.DOTALL
46 )
47 _TIMESTAMP_RE = _re.compile(r"- `[^`]+` - (\d{4}-\d{2}-\d{2})T\d{2}:\d{2}:\d{2}")
49 match = _SESSION_LOG_RE.search(content)
50 if match:
51 timestamps = _TIMESTAMP_RE.findall(match.group(1))
52 if timestamps:
53 try:
54 return date.fromisoformat(timestamps[-1])
55 except ValueError:
56 pass
58 # Fallback to file mtime
59 try:
60 return date.fromtimestamp(file_path.stat().st_mtime)
61 except OSError:
62 return None
65def _parse_labels_from_content(content: str) -> list[str]:
66 """Extract labels from ## Labels section (backtick-wrapped items)."""
67 match = re.search(r"## Labels\s*\n(.*?)(?:\n##|\Z)", content, re.DOTALL)
68 if not match:
69 return []
70 return [m.lower() for m in re.findall(r"`([^`]+)`", match.group(1))]
73def _parse_priority_filter(priority_values: list[str]) -> set[str]:
74 """Parse priority filter values, supporting ranges like P0-P2."""
75 priorities: set[str] = set()
76 for val in priority_values:
77 range_match = re.match(r"^(P\d)-(P\d)$", val)
78 if range_match:
79 start = int(range_match.group(1)[1:])
80 end = int(range_match.group(2)[1:])
81 for i in range(start, end + 1):
82 priorities.add(f"P{i}")
83 elif re.match(r"^P\d$", val):
84 priorities.add(val)
85 return priorities
88def _load_issues_with_status(
89 config: BRConfig,
90 include_active: bool,
91 include_completed: bool,
92 include_deferred: bool,
93) -> list[tuple[IssueInfo, str]]:
94 """Load issues from the relevant directories, tagged with their status.
96 Returns:
97 List of (IssueInfo, status) where status is 'active', 'completed', or 'deferred'.
98 """
99 from little_loops.issue_parser import IssueParser
101 parser = IssueParser(config)
102 results: list[tuple[IssueInfo, str]] = []
104 if include_active:
105 for category in config.issue_categories:
106 issue_dir = config.get_issue_dir(category)
107 if issue_dir.exists():
108 for f in sorted(issue_dir.glob("*.md")):
109 try:
110 results.append((parser.parse_file(f), "active"))
111 except Exception:
112 continue
114 if include_completed:
115 completed_dir = config.get_completed_dir()
116 if completed_dir.exists():
117 for f in sorted(completed_dir.glob("*.md")):
118 try:
119 results.append((parser.parse_file(f), "completed"))
120 except Exception:
121 continue
123 if include_deferred:
124 deferred_dir = config.get_deferred_dir()
125 if deferred_dir.exists():
126 for f in sorted(deferred_dir.glob("*.md")):
127 try:
128 results.append((parser.parse_file(f), "deferred"))
129 except Exception:
130 continue
132 return results
135def _sort_issues(
136 items: list[tuple[IssueInfo, str, date | None, date | None]],
137 sort_field: str,
138 descending: bool,
139) -> list[tuple[IssueInfo, str, date | None, date | None]]:
140 """Sort issues by the requested field."""
141 sentinel = date(9999, 12, 31)
143 def key(item: tuple) -> tuple:
144 issue, _status, disc_date, comp_date = item
145 if sort_field == "priority":
146 return (issue.priority_int, issue.issue_id)
147 if sort_field == "id":
148 m = re.search(r"-(\d+)$", issue.issue_id)
149 num = int(m.group(1)) if m else 0
150 return (issue.issue_id.split("-", 1)[0], num)
151 if sort_field in ("date", "created"):
152 return (disc_date or sentinel,)
153 if sort_field == "completed":
154 return (comp_date or sentinel,)
155 if sort_field == "type":
156 return (issue.issue_id.split("-", 1)[0], issue.priority_int, issue.issue_id)
157 if sort_field == "title":
158 return (issue.title.lower(),)
159 if sort_field == "confidence":
160 score = issue.confidence_score if issue.confidence_score is not None else 9999
161 return (score,)
162 if sort_field == "outcome":
163 score = issue.outcome_confidence if issue.outcome_confidence is not None else 9999
164 return (score,)
165 if sort_field == "refinement":
166 refinement_commands = {
167 "/ll:verify-issues",
168 "/ll:refine-issue",
169 "/ll:tradeoff-review-issues",
170 "/ll:map-dependencies",
171 "/ll:ready-issue",
172 }
173 counts: dict[str, int] = getattr(issue, "session_command_counts", {}) or {}
174 total = sum(counts.get(cmd, 0) for cmd in refinement_commands)
175 return (total,)
176 return (issue.priority_int, issue.issue_id)
178 return sorted(items, key=key, reverse=descending)
181def cmd_search(config: BRConfig, args: argparse.Namespace) -> int:
182 """Search issues with optional text query, filters, and sorting.
184 Args:
185 config: Project configuration
186 args: Parsed arguments
188 Returns:
189 Exit code (0 = success)
190 """
191 # Resolve status flags
192 status = getattr(args, "status", "active")
193 if getattr(args, "include_completed", False):
194 status = "all"
196 include_active = status in ("active", "all")
197 include_completed = status in ("completed", "all")
198 include_deferred = status in ("deferred", "all")
200 # Load issues
201 raw = _load_issues_with_status(config, include_active, include_completed, include_deferred)
203 # Parse additional metadata (dates, labels) only when needed
204 query: str | None = getattr(args, "query", None)
205 since_date: date | None = None
206 until_date: date | None = None
207 raw_since = getattr(args, "since", None)
208 raw_until = getattr(args, "until", None)
209 if raw_since:
210 try:
211 since_date = date.fromisoformat(raw_since)
212 except ValueError:
213 print(f"Invalid --since date: {raw_since!r}. Use YYYY-MM-DD.")
214 return 1
215 if raw_until:
216 try:
217 until_date = date.fromisoformat(raw_until)
218 except ValueError:
219 print(f"Invalid --until date: {raw_until!r}. Use YYYY-MM-DD.")
220 return 1
222 sort_field = getattr(args, "sort", "priority") or "priority"
223 need_content = bool(
224 query
225 or since_date
226 or until_date
227 or getattr(args, "label", None)
228 or sort_field in {"date", "created", "completed"}
229 or getattr(args, "date_field", "discovered") == "updated"
230 )
232 # Build enriched list: (IssueInfo, status, discovered_date, completed_date)
233 enriched: list[tuple[IssueInfo, str, date | None, date | None]] = []
234 for issue, stat in raw:
235 if need_content:
236 try:
237 content = issue.path.read_text(encoding="utf-8")
238 except Exception:
239 content = ""
240 disc_date = _parse_discovered_date(content)
241 labels = _parse_labels_from_content(content)
242 else:
243 content = ""
244 disc_date = None
245 labels = []
247 # --- Filter: text query ---
248 if query:
249 haystack = (issue.title + "\n" + content).lower()
250 if query.lower() not in haystack:
251 continue
253 # --- Filter: type ---
254 type_filters: list[str] = getattr(args, "type", None) or []
255 if type_filters:
256 issue_type = issue.issue_id.split("-", 1)[0]
257 if issue_type not in type_filters:
258 continue
260 # --- Filter: priority ---
261 priority_filters: list[str] = getattr(args, "priority", None) or []
262 if priority_filters:
263 allowed = _parse_priority_filter(priority_filters)
264 if issue.priority not in allowed:
265 continue
267 # --- Filter: label ---
268 label_filters: list[str] = getattr(args, "label", None) or []
269 if label_filters:
270 if not any(lf.lower() in labels for lf in label_filters):
271 continue
273 # --- Filter: date range ---
274 date_field = getattr(args, "date_field", "discovered")
275 if date_field == "discovered":
276 ref_date = disc_date
277 else:
278 ref_date = _parse_updated_date(content, issue.path)
280 if since_date and (ref_date is None or ref_date < since_date):
281 continue
282 if until_date and (ref_date is None or ref_date > until_date):
283 continue
285 comp_date: date | None = None
286 if sort_field == "completed" and need_content:
287 from little_loops.issue_history.parsing import _parse_completion_date
289 comp_date = _parse_completion_date(content, issue.path)
290 enriched.append((issue, stat, disc_date, comp_date))
292 # --- Sort ---
293 # Default direction: desc for date/created/completed (newest first), asc for everything else
294 if getattr(args, "desc", False):
295 descending = True
296 elif getattr(args, "asc", False):
297 descending = False
298 else:
299 descending = sort_field in {"date", "created", "completed"}
301 enriched = _sort_issues(enriched, sort_field, descending)
303 # --- Limit ---
304 limit = getattr(args, "limit", None)
305 if limit and limit > 0:
306 enriched = enriched[:limit]
308 if not enriched:
309 print("No issues found.")
310 return 0
312 # --- Output ---
313 issues_out = [item[0] for item in enriched]
314 statuses_out = [item[1] for item in enriched]
315 dates_out = [item[2] for item in enriched]
317 if getattr(args, "json", False):
318 print_json(
319 [
320 {
321 "id": issue.issue_id,
322 "priority": issue.priority,
323 "type": issue.issue_id.split("-", 1)[0],
324 "title": issue.title,
325 "path": str(issue.path),
326 "status": stat,
327 "discovered_date": str(d) if d else None,
328 }
329 for issue, stat, d in zip(issues_out, statuses_out, dates_out, strict=True)
330 ]
331 )
332 return 0
334 fmt = getattr(args, "format", "table") or "table"
336 if fmt == "ids":
337 for issue in issues_out:
338 print(issue.issue_id)
339 return 0
341 if fmt == "list":
342 for issue in issues_out:
343 print(f"{issue.path.name} {issue.title}")
344 return 0
346 # Default: table (grouped by type, similar to ll-issues list)
347 buckets: dict[str, list[tuple[IssueInfo, str]]] = {"BUG": [], "FEAT": [], "ENH": []}
348 for issue, stat in zip(issues_out, statuses_out, strict=True):
349 prefix = issue.issue_id.split("-", 1)[0]
350 if prefix in buckets:
351 buckets[prefix].append((issue, stat))
353 type_labels = {"BUG": "Bugs", "FEAT": "Features", "ENH": "Enhancements"}
354 lines: list[str] = []
355 for prefix, label in type_labels.items():
356 group = buckets[prefix]
357 if not group:
358 continue
359 header = colorize(f"{label} ({len(group)})", f"{TYPE_COLOR.get(prefix, '0')};1")
360 lines.append(header)
361 for issue, stat in group:
362 issue_type = issue.issue_id.split("-", 1)[0]
363 colored_id = colorize(issue.issue_id, TYPE_COLOR.get(issue_type, "0"))
364 colored_priority = colorize(issue.priority, PRIORITY_COLOR.get(issue.priority, "0"))
365 status_tag = f" [{stat}]" if stat != "active" else ""
366 lines.append(f" {colored_priority} {colored_id} {issue.title}{status_tag}")
367 lines.append("")
368 lines.append(f"Total: {len(issues_out)} issue(s) found")
369 print("\n".join(lines))
370 return 0