Coverage for little_loops / issues / anchor_sweep.py: 89%

1"""Backlog sweeper: rewrite file:line references in active issue files — ENH-1300. 

2 

3Two-phase sweep: 

4 1. Scan: find file:line patterns outside code fences, resolve each to an anchor. 

5 2. Apply: rewrite matches in-place using atomic_write(). 

6 

7Respects --dry-run: always records what would change, gates actual writes. 

8""" 

9 

10from __future__ import annotations 

11 

12import re 

13import sys 

14import warnings 

15from dataclasses import dataclass, field 

16from pathlib import Path 

17 

18from little_loops.file_utils import atomic_write 

19from little_loops.issues.anchors import resolve_anchor 

20from little_loops.text_utils import _CODE_FENCE 

21 

22# Captures file path (group 1) and line number (group 2) together. 

23# Unlike _STANDALONE_PATH, the :NNN suffix is required and captured separately. 

24_FILE_LINE = re.compile( 

25 r"(?:(?<=\s)|^)([a-zA-Z_][\w/.-]*\.[a-z]{2,4}):(\d+)(?=\s|$|:|\))", 

26 re.MULTILINE, 

27) 

28 

29_ACTIVE_CATEGORIES = ("bugs", "features", "enhancements", "epics") 

30 

31 

32@dataclass 

33class SweepResult: 

34 """Result of a sweep pass.""" 

35 

36 changes: list[str] = field(default_factory=list) 

37 modified_files: set[str] = field(default_factory=set) 

38 skipped_refs: int = 0 

39 

40 

41def _format_anchor_ref(file_path: str, anchor: str) -> str: 

42 """Format a resolved anchor reference. 

43 

44 Examples: 

45 "near function foo" -> "`file.py` (near function `foo`)" 

46 "near class Bar" -> "`file.py` (near class `Bar`)" 

47 'under section "X"' -> '`file.py` (under section "X")' 

48 """ 

49 # Backtick the name in "near function foo" → near function `foo` 

50 # but leave section titles as-is (already quoted with double-quotes). 

51 parts = anchor.split(" ", 2) # e.g. ["near", "function", "foo"] 

52 if len(parts) == 3 and parts[0] in ("near",): 

53 anchor_display = f"{parts[0]} {parts[1]} `{parts[2]}`" 

54 else: 

55 anchor_display = anchor 

56 return f"`{file_path}` ({anchor_display})" 

57 

58 

59def _sweep_file(path: Path, dry_run: bool, result: SweepResult) -> None: 

60 """Scan one issue file and rewrite file:line references.""" 

61 content = path.read_text(encoding="utf-8", errors="replace") 

62 

63 fence_spans = [(m.start(), m.end()) for m in _CODE_FENCE.finditer(content)] 

64 

65 def _in_fence(start: int, end: int) -> bool: 

66 return any(fs <= start and end <= fe for fs, fe in fence_spans) 

67 

68 replacements: list[tuple[int, int, str]] = [] 

69 for m in _FILE_LINE.finditer(content): 

70 if _in_fence(m.start(), m.end()): 

71 continue 

72 ref_path = m.group(1) 

73 line_no = int(m.group(2)) 

74 anchor = resolve_anchor(ref_path, line_no) 

75 if anchor is None: 

76 result.skipped_refs += 1 

77 warnings.warn( 

78 f"{path}: could not resolve anchor for {ref_path}:{line_no}", 

79 stacklevel=2, 

80 ) 

81 continue 

82 replacement = _format_anchor_ref(ref_path, anchor) 

83 replacements.append((m.start(), m.end(), replacement)) 

84 

85 if not replacements: 

86 return 

87 

88 desc = f"{path}: rewrote {len(replacements)} file:line reference(s)" 

89 result.changes.append(desc) 

90 

91 if not dry_run: 

92 # Apply replacements in reverse order so positions stay valid 

93 new_content = content 

94 for start, end, replacement in reversed(replacements): 

95 new_content = new_content[:start] + replacement + new_content[end:] 

96 atomic_write(path, new_content) 

97 result.modified_files.add(str(path)) 

98 

99 

100def sweep_issues(issues_dir: Path, dry_run: bool = False) -> SweepResult: 

101 """Sweep all active issue files in issues_dir, rewriting file:line refs. 

102 

103 Args: 

104 issues_dir: Base directory containing bugs/, features/, enhancements/ subdirs. 

105 dry_run: If True, report changes without modifying files. 

106 

107 Returns: 

108 SweepResult with changes, modified_files, and skipped_refs counts. 

109 """ 

110 result = SweepResult() 

111 for category in _ACTIVE_CATEGORIES: 

112 cat_dir = issues_dir / category 

113 if not cat_dir.is_dir(): 

114 continue 

115 for issue_file in sorted(cat_dir.glob("*.md")): 

116 try: 

117 _sweep_file(issue_file, dry_run, result) 

118 except OSError as exc: 

119 print(f"Warning: skipping {issue_file}: {exc}", file=sys.stderr) 

120 return result