1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
#!/usr/bin/env python3
"""cj-remove-block — Remove a cj annotation block from an org file by line range.
Idempotently deletes lines [start, end] (1-indexed, inclusive) from the file,
but only after validating that those lines actually look like a cj annotation
(either a `#+begin_src cj: ... #+end_src` fence pair or a single `cj:` line).
The validation step is the point — it protects against accidentally trimming
the wrong block when line numbers drift between a `cj-scan` call and a remove call.
Usage:
cj-remove-block --file FILE.org --start N --end M
Companion to the /respond-to-cj-comments skill and to cj-scan.py.
"""
from __future__ import annotations
import argparse
import os
import re
import shutil
import sys
import tempfile
from datetime import datetime
from pathlib import Path
SRC_OPEN_RE = re.compile(r"^\s*#\+begin_src\s+cj:", re.IGNORECASE)
SRC_CLOSE_RE = re.compile(r"^\s*#\+end_src\s*$", re.IGNORECASE)
LEGACY_CJ_RE = re.compile(r"^\s*cj:\s")
def looks_like_cj_range(lines: list[str], start: int, end: int) -> tuple[bool, str]:
"""Return (ok, reason). Validates start..end (1-indexed, inclusive) is a cj range."""
if end < start:
return False, f"Range end ({end}) is before start ({start})"
if start < 1 or end > len(lines):
return False, (
f"Range {start}..{end} is out of bounds for a file with {len(lines)} lines"
)
first = lines[start - 1]
last = lines[end - 1]
if start == end:
# Single-line removal must look like legacy inline.
if LEGACY_CJ_RE.match(first):
return True, ""
return False, (
f"Line {start} does not look like a legacy inline cj: line "
f"(got: {first[:60]!r})"
)
# Multi-line removal must look like a source-block fence pair.
if not SRC_OPEN_RE.match(first):
return False, (
f"Line {start} does not look like a #+begin_src cj: opening fence "
f"(got: {first[:60]!r})"
)
if not SRC_CLOSE_RE.match(last):
return False, (
f"Line {end} does not look like a #+end_src closing fence "
f"(got: {last[:60]!r})"
)
# The range must hold exactly ONE block. Checking only the first and last
# lines let a drifted range run from one block's opener to a *later* block's
# closer: validation passed and the removal silently deleted everything
# between, prose and headings included. Drift is the case this check exists
# for, so it has to look inside the range, not just at its ends.
for offset, line in enumerate(lines[start:end - 1], start=start + 1):
if SRC_CLOSE_RE.match(line):
return False, (
f"Range {start}..{end} covers more than one cj block — "
f"a #+end_src appears at line {offset}, before the range ends. "
f"Re-scan for current line numbers; removing this range would "
f"delete everything between the two blocks."
)
if SRC_OPEN_RE.match(line):
return False, (
f"Range {start}..{end} covers more than one cj block — "
f"a second #+begin_src cj: appears at line {offset}. "
f"Re-scan for current line numbers."
)
return True, ""
def _backup(path: Path) -> Path:
"""Copy path to /tmp before mutating it, mirroring lint-org.el's convention.
These are Craig's org files. lint-org.el, the other tool that rewrites them,
leaves a /tmp copy before touching anything; this matches it so a bad edit is
always recoverable without reaching for git (which only reaches the last
commit, losing intra-session work).
"""
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
base = Path(tempfile.gettempdir()) / f"{path.name}.before-cj-remove.{stamp}"
# Never overwrite an earlier backup. The skill removes several annotations
# in quick succession, so a second-resolution stamp collides and the later
# copy would replace the earlier one with already-mutated content — losing
# the pre-session original the backup exists to preserve.
dest = base
n = 2
while dest.exists():
dest = base.with_name(f"{base.name}-{n}")
n += 1
shutil.copy2(path, dest)
return dest
def _atomic_write(path: Path, text: str) -> None:
"""Write text to path via a temp sibling and os.replace.
A bare write_text truncates the target on open, so a mid-write failure left
the org file truncated with no complete copy on disk. Writing a temp sibling
and renaming means the file is either its old content or its new content,
never a partial.
"""
# Follow a symlink to the file it names. os.replace would otherwise swap the
# symlink itself for a regular file, leaving the real target holding the old
# content — the edit silently goes nowhere. Resolving also puts the temp
# sibling on the same filesystem as the real file, which os.replace needs.
path = path.resolve()
fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp")
os.close(fd)
tmp_path = Path(tmp)
# Carry the original's permissions across. mkstemp creates 0600, and
# defaulting to the umask instead widened a deliberately-restricted file
# (a 0600 org file came back 0644).
shutil.copymode(path, tmp_path)
try:
tmp_path.write_text(text, encoding="utf-8")
os.replace(tmp_path, path)
except BaseException:
tmp_path.unlink(missing_ok=True)
raise
def remove_range(path: Path, start: int, end: int) -> None:
"""Read path, validate range looks like cj content, remove the range, write back."""
text = path.read_text(encoding="utf-8")
had_trailing_newline = text.endswith("\n")
lines = text.splitlines(keepends=False)
ok, reason = looks_like_cj_range(lines, start, end)
if not ok:
print(f"cj-remove-block: refusing to remove — {reason}", file=sys.stderr)
sys.exit(1)
new_lines = lines[: start - 1] + lines[end:]
new_text = "\n".join(new_lines)
if new_lines and had_trailing_newline:
new_text += "\n"
elif not new_lines and had_trailing_newline:
new_text = ""
_backup(path)
_atomic_write(path, new_text)
def main() -> int:
parser = argparse.ArgumentParser(
description="Remove a cj annotation block from an org file by line range.",
)
parser.add_argument("--file", required=True, type=Path, help="Path to the org file.")
parser.add_argument("--start", required=True, type=int, help="Start line (1-indexed, inclusive).")
parser.add_argument("--end", required=True, type=int, help="End line (1-indexed, inclusive).")
args = parser.parse_args()
if not args.file.is_file():
print(f"Not a file: {args.file}", file=sys.stderr)
return 2
remove_range(args.file, args.start, args.end)
return 0
if __name__ == "__main__":
sys.exit(main())
|