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

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 warnings 

9from pathlib import Path 

10 

11from little_loops.cli_args import add_config_arg, add_dry_run_arg 

12from little_loops.frontmatter import parse_frontmatter 

13from little_loops.issue_lifecycle import _is_git_tracked 

14 

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

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

17 

18 

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

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

21 

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

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

24 """ 

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

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

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

28 

29 result = content 

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

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

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

33 if key_re.search(result): 

34 result = key_re.sub(line, result) 

35 else: 

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

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

38 if len(markers) >= 2: 

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

40 pos = markers[1].start() 

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

42 return result 

43 

44 

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

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

47 m = _FILENAME_PREFIX_RE.match(name) 

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

49 

50 

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

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

53 try: 

54 result = subprocess.run( 

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

56 capture_output=True, 

57 text=True, 

58 timeout=30, 

59 ) 

60 if result.returncode == 0: 

61 date_str = result.stdout.strip() 

62 if date_str: 

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

64 except (subprocess.TimeoutExpired, OSError): 

65 pass 

66 return None 

67 

68 

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

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

71 if _is_git_tracked(src): 

72 result = subprocess.run( 

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

74 capture_output=True, 

75 text=True, 

76 timeout=30, 

77 ) 

78 if result.returncode == 0: 

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

80 return 

81 # Untracked or git mv failed — write then rename 

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

83 src.rename(dst) 

84 

85 

86def main_migrate() -> int: 

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

88 

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

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

91 frontmatter. 

92 

93 Returns: 

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

95 """ 

96 parser = argparse.ArgumentParser( 

97 prog="ll-migrate", 

98 description=( 

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

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

101 ), 

102 formatter_class=argparse.RawDescriptionHelpFormatter, 

103 epilog=""" 

104Examples: 

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

106 %(prog)s # Execute migration 

107""", 

108 ) 

109 add_dry_run_arg(parser) 

110 add_config_arg(parser) 

111 args = parser.parse_args() 

112 

113 dry_run: bool = args.dry_run 

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

115 

116 from little_loops.config import BRConfig 

117 

118 config = BRConfig(repo_root) 

119 

120 with warnings.catch_warnings(): 

121 warnings.simplefilter("ignore", DeprecationWarning) 

122 completed_dir: Path = config.get_completed_dir() 

123 deferred_dir: Path = config.get_deferred_dir() 

124 

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

126 prefix_to_key: dict[str, str] = { 

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

128 } 

129 

130 if dry_run: 

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

132 

133 moved = 0 

134 backfilled = 0 

135 untyped: list[str] = [] 

136 errors: list[str] = [] 

137 

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

139 if completed_dir.exists(): 

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

141 if deferred_dir.exists(): 

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

143 

144 for file_path, status in sources: 

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

146 fm = parse_frontmatter(content) 

147 

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

149 

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

151 date = _get_git_completion_date(file_path) 

152 if date: 

153 updates["completed_at"] = date 

154 backfilled += 1 

155 

156 # Determine type prefix from frontmatter or filename 

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

158 if not type_prefix: 

159 type_prefix = _extract_prefix_from_filename(file_path.name) 

160 

161 if not type_prefix: 

162 untyped.append(str(file_path)) 

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

164 continue 

165 

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

167 if category_key is None: 

168 untyped.append(str(file_path)) 

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

170 continue 

171 

172 target_dir = config.get_issue_dir(category_key) 

173 target_path = target_dir / file_path.name 

174 

175 if target_path.exists(): 

176 errors.append(str(file_path)) 

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

178 continue 

179 

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

181 

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

183 print( 

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

185 ) 

186 

187 if not dry_run: 

188 try: 

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

190 _move_file(file_path, target_path, updated_content) 

191 moved += 1 

192 except Exception as exc: 

193 errors.append(str(file_path)) 

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

195 else: 

196 moved += 1 

197 

198 print() 

199 print( 

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

201 f"{backfilled} completed_at fields backfilled." 

202 ) 

203 if untyped: 

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

205 for p in untyped: 

206 print(f" {p}") 

207 if errors: 

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

209 return 1 

210 return 0