Coverage for little_loops / cli / migrate_relationships.py: 16%

1"""ll-migrate-relationships: Rename parent_issue: -> parent: and related: -> relates_to: in issue files.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import re 

7import sys 

8from pathlib import Path 

9 

10from little_loops.cli_args import add_config_arg, add_dry_run_arg 

11from little_loops.frontmatter import parse_frontmatter 

12from little_loops.session_store import DEFAULT_DB_PATH, cli_event_context 

13 

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

15 

16_PARENT_ISSUE_RE = re.compile(r"^parent_issue:(.*)$", re.MULTILINE) 

17_RELATED_RE = re.compile(r"^related:(.*)$", re.MULTILINE) 

18 

19 

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

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

22 

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

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

25 """ 

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

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

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

29 

30 result = content 

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

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

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

34 if key_re.search(result): 

35 result = key_re.sub(line, result) 

36 else: 

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

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

39 if len(markers) >= 2: 

40 pos = markers[1].start() 

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

42 return result 

43 

44 

45def _migrate_content(content: str) -> tuple[str, list[str]]: 

46 """Rename relationship keys in frontmatter. Returns (updated_content, list_of_renames).""" 

47 fm = parse_frontmatter(content) 

48 if not fm: 

49 return content, [] 

50 

51 renames: list[str] = [] 

52 result = content 

53 

54 if "parent_issue" in fm: 

55 value = str(fm["parent_issue"]) 

56 result = _set_fields(result, {"parent": value}) 

57 result = _PARENT_ISSUE_RE.sub("", result) 

58 # Clean up blank line left by removal 

59 result = re.sub(r"\n{3,}", "\n\n", result) 

60 renames.append(f"parent_issue: {value!r} → parent: {value!r}") 

61 

62 if "related" in fm: 

63 value_raw = _RELATED_RE.search(content) 

64 raw_suffix = value_raw.group(1) if value_raw else "" 

65 result = _set_fields(result, {"relates_to": fm["related"]}) 

66 result = _RELATED_RE.sub("", result) 

67 result = re.sub(r"\n{3,}", "\n\n", result) 

68 renames.append(f"related:{raw_suffix} → relates_to:{raw_suffix}") 

69 

70 return result, renames 

71 

72 

73def main_migrate_relationships() -> int: 

74 """Entry point for ll-migrate-relationships command. 

75 

76 Renames parent_issue: -> parent: and related: -> relates_to: in all issue files. 

77 

78 Returns: 

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

80 """ 

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

82 parser = argparse.ArgumentParser( 

83 prog="ll-migrate-relationships", 

84 description=( 

85 "Rename parent_issue: → parent: and related: → relates_to: " 

86 "in all issue frontmatter files. One-time migration for ENH-1431." 

87 ), 

88 formatter_class=argparse.RawDescriptionHelpFormatter, 

89 epilog=""" 

90Examples: 

91 %(prog)s --dry-run # Preview all planned renames (strongly advised first) 

92 %(prog)s # Execute migration 

93""", 

94 ) 

95 add_dry_run_arg(parser) 

96 add_config_arg(parser) 

97 args = parser.parse_args() 

98 

99 dry_run: bool = args.dry_run 

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

101 

102 issues_dir = repo_root / ".issues" 

103 if not issues_dir.exists(): 

104 print(f"No .issues/ directory found at {repo_root}") 

105 return 1 

106 

107 if dry_run: 

108 print("[DRY RUN] No files will be modified.") 

109 

110 renamed = 0 

111 errors: list[str] = [] 

112 

113 for file_path in sorted(issues_dir.rglob("*.md")): 

114 try: 

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

116 except OSError as exc: 

117 errors.append(str(file_path)) 

118 print(f" [ERROR] {file_path}: {exc}") 

119 continue 

120 

121 updated, renames = _migrate_content(content) 

122 if not renames: 

123 continue 

124 

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

126 rel = file_path.relative_to(repo_root) 

127 for rename in renames: 

128 print(f" {prefix}RENAME {rel}: {rename}") 

129 

130 if not dry_run: 

131 try: 

132 file_path.write_text(updated, encoding="utf-8") 

133 renamed += 1 

134 except OSError as exc: 

135 errors.append(str(file_path)) 

136 print(f" [ERROR] {file_path}: {exc}") 

137 else: 

138 renamed += 1 

139 

140 print() 

141 print(f"Results: {renamed} files {'would be ' if dry_run else ''}updated.") 

142 if errors: 

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

144 return 1 

145 return 0