Coverage for little_loops / cli / migrate.py: 15%

1"""ll-migrate: One-time migration of completed/deferred issues to type-based directories.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import re 

7import subprocess 

8import sys 

9import warnings 

10from pathlib import Path 

11 

12from little_loops.cli_args import add_config_arg, add_dry_run_arg 

13from little_loops.frontmatter import parse_frontmatter 

14from little_loops.issue_lifecycle import _is_git_tracked 

15from little_loops.session_store import DEFAULT_DB_PATH, cli_event_context 

16 

17_FILENAME_PREFIX_RE = re.compile(r"^(?:P\d+-)?([A-Z]+)-\d+") 

18_FM_FIELD_RE = re.compile(r"^---\s*$", re.MULTILINE) 

19 

20 

21def _set_fields(content: str, fields: dict[str, str]) -> str: 

22 """Set frontmatter fields via direct string manipulation (avoids yaml roundtrip). 

23 

24 Replaces existing fields in-place; inserts missing fields before the closing 

25 ``---`` marker. Prepends a new frontmatter block if none exists. 

26 """ 

27 if not content.startswith("---\n"): 

28 lines = "\n".join(f"{k}: {v}" for k, v in fields.items()) 

29 return f"---\n{lines}\n---\n{content}" 

30 

31 result = content 

32 for key, value in fields.items(): 

33 line = f"{key}: {value}" 

34 key_re = re.compile(rf"^{re.escape(key)}:.*$", re.MULTILINE) 

35 if key_re.search(result): 

36 result = key_re.sub(line, result) 

37 else: 

38 # Insert before the closing --- of the frontmatter 

39 markers = list(_FM_FIELD_RE.finditer(result)) 

40 if len(markers) >= 2: 

41 # markers[0] is opening ---, markers[1] is closing --- 

42 pos = markers[1].start() 

43 result = result[:pos] + f"{line}\n" + result[pos:] 

44 return result 

45 

46 

47def _extract_prefix_from_filename(name: str) -> str | None: 

48 """Extract issue type prefix from filename (e.g. 'P2-ENH-1420-...' → 'ENH').""" 

49 m = _FILENAME_PREFIX_RE.match(name) 

50 return m.group(1) if m else None 

51 

52 

53def _get_git_completion_date(file_path: Path) -> str | None: 

54 """Return ISO-8601 date from git log for file_path, or None if unavailable.""" 

55 try: 

56 result = subprocess.run( 

57 ["git", "log", "--format=%as", "-1", "--", str(file_path)], 

58 capture_output=True, 

59 text=True, 

60 timeout=30, 

61 ) 

62 if result.returncode == 0: 

63 date_str = result.stdout.strip() 

64 if date_str: 

65 return f"{date_str}T00:00:00Z" 

66 except (subprocess.TimeoutExpired, OSError): 

67 pass 

68 return None 

69 

70 

71def _move_file(src: Path, dst: Path, updated_content: str) -> None: 

72 """Move src to dst, writing updated_content. Prefers git mv for tracked files.""" 

73 if _is_git_tracked(src): 

74 result = subprocess.run( 

75 ["git", "mv", str(src), str(dst)], 

76 capture_output=True, 

77 text=True, 

78 timeout=30, 

79 ) 

80 if result.returncode == 0: 

81 dst.write_text(updated_content, encoding="utf-8") 

82 return 

83 # Untracked or git mv failed — write then rename 

84 src.write_text(updated_content, encoding="utf-8") 

85 src.rename(dst) 

86 

87 

88def main_migrate() -> int: 

89 """Entry point for ll-migrate command. 

90 

91 Moves files from completed/ and deferred/ into their type-based directories, 

92 backfills completed_at for completed files missing it, and sets correct status 

93 frontmatter. 

94 

95 Returns: 

96 Exit code (0 = success, 1 = error) 

97 """ 

98 with cli_event_context(DEFAULT_DB_PATH, "ll-migrate", sys.argv[1:]): 

99 parser = argparse.ArgumentParser( 

100 prog="ll-migrate", 

101 description=( 

102 "Migrate completed/ and deferred/ issues to type-based directories " 

103 "with correct status frontmatter. One-time migration for ENH-1390." 

104 ), 

105 formatter_class=argparse.RawDescriptionHelpFormatter, 

106 epilog=""" 

107Examples: 

108 %(prog)s --dry-run # Preview all planned moves (strongly advised first) 

109 %(prog)s # Execute migration 

110""", 

111 ) 

112 add_dry_run_arg(parser) 

113 add_config_arg(parser) 

114 args = parser.parse_args() 

115 

116 dry_run: bool = args.dry_run 

117 repo_root: Path = args.config or Path.cwd() 

118 

119 from little_loops.config import BRConfig 

120 

121 config = BRConfig(repo_root) 

122 

123 with warnings.catch_warnings(): 

124 warnings.simplefilter("ignore", DeprecationWarning) 

125 completed_dir: Path = config.get_completed_dir() 

126 deferred_dir: Path = config.get_deferred_dir() 

127 

128 # Build prefix → category-key mapping (e.g. "BUG" → "bugs") 

129 prefix_to_key: dict[str, str] = { 

130 cat.prefix.upper(): key for key, cat in config._issues.categories.items() 

131 } 

132 

133 if dry_run: 

134 print("[DRY RUN] No files will be moved or modified.") 

135 

136 moved = 0 

137 backfilled = 0 

138 untyped: list[str] = [] 

139 errors: list[str] = [] 

140 

141 sources: list[tuple[Path, str]] = [] 

142 if completed_dir.exists(): 

143 sources += [(f, "done") for f in sorted(completed_dir.glob("*.md"))] 

144 if deferred_dir.exists(): 

145 sources += [(f, "deferred") for f in sorted(deferred_dir.glob("*.md"))] 

146 

147 for file_path, status in sources: 

148 content = file_path.read_text(encoding="utf-8") 

149 fm = parse_frontmatter(content) 

150 

151 updates: dict[str, str | int] = {"status": status} 

152 

153 if status == "done" and "completed_at" not in fm: 

154 date = _get_git_completion_date(file_path) 

155 if date: 

156 updates["completed_at"] = date 

157 backfilled += 1 

158 

159 # Determine type prefix from frontmatter or filename 

160 type_prefix: str | None = fm.get("type") if fm else None 

161 if not type_prefix: 

162 type_prefix = _extract_prefix_from_filename(file_path.name) 

163 

164 if not type_prefix: 

165 untyped.append(str(file_path)) 

166 print(f" [UNTYPED] {file_path.name} — cannot determine type, skipped") 

167 continue 

168 

169 category_key = prefix_to_key.get(type_prefix.upper()) 

170 if category_key is None: 

171 untyped.append(str(file_path)) 

172 print(f" [UNTYPED] {file_path.name} — unknown prefix '{type_prefix}', skipped") 

173 continue 

174 

175 target_dir = config.get_issue_dir(category_key) 

176 target_path = target_dir / file_path.name 

177 

178 if target_path.exists(): 

179 errors.append(str(file_path)) 

180 print(f" [SKIP] {file_path.name} — target already exists: {target_path}") 

181 continue 

182 

183 updated_content = _set_fields(content, {k: str(v) for k, v in updates.items()}) 

184 

185 prefix = "[DRY RUN] " if dry_run else "" 

186 print( 

187 f" {prefix}MOVE {file_path.relative_to(repo_root)} → {target_path.relative_to(repo_root)}" 

188 ) 

189 

190 if not dry_run: 

191 try: 

192 target_dir.mkdir(parents=True, exist_ok=True) 

193 _move_file(file_path, target_path, updated_content) 

194 moved += 1 

195 except Exception as exc: 

196 errors.append(str(file_path)) 

197 print(f" [ERROR] {file_path.name}: {exc}") 

198 else: 

199 moved += 1 

200 

201 print() 

202 print( 

203 f"Results: {moved} files {'would be ' if dry_run else ''}moved, " 

204 f"{backfilled} completed_at fields backfilled." 

205 ) 

206 if untyped: 

207 print(f" Untyped (needs manual review): {len(untyped)}") 

208 for p in untyped: 

209 print(f" {p}") 

210 if errors: 

211 print(f" Errors: {len(errors)}") 

212 return 1 

213 return 0