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
|
#!/usr/bin/env python3
"""route-batch — the wrap-up router's mechanical go path.
The wrap-up cross-project router (wrap-it-up.org Step 3; wrapup-routing spec
D7/D8/D9) surfaces the local tasks that inbox process mode stamped with
:ROUTE_CANDIDATE: <destination> at file time, and on "go" delivers each to its
destination project's inbox. This script does the mechanical half so the
subtree surgery is deterministic:
route-batch --list [--todo todo.org]
One "<destination>\t<heading>" line per :ROUTE_CANDIDATE:-tagged task.
Silent with exit 0 when there are no candidates (the workflow's
empty-set-equals-zero-interaction rule). Read-only.
route-batch --go [--todo todo.org]
For each candidate, bottom-up: extract the task's whole subtree
(children ride along), drop the :ROUTE_CANDIDATE: line (and the
property drawer if that leaves it empty), promote the subtree so its
top heading is level 1, write it to a temp file, and deliver it via
the sibling inbox-send.py to the destination's inbox/ (one file per
task, from-<source> provenance stamped by inbox-send). Only after a
successful send is the subtree removed from the local todo.org — a
failed send leaves that task in place, is reported, and the run exits
non-zero after attempting the rest.
The candidate set is exactly the tagged tasks — never the standing backlog.
Discovery, roots, and the source-project name all come from inbox-send.py
(INBOX_SEND_ROOTS sandboxes it in tests). The reject-from-another-project
flow in inbox process mode is the mis-route recovery; that path is why
removing the local source after a successful send is safe.
"""
import argparse
import os
import re
import subprocess
import sys
import tempfile
from pathlib import Path
HEADING_RE = re.compile(r"^(\*+)\s+(.*)$")
MARKER_RE = re.compile(r"^\s*:ROUTE_CANDIDATE:\s+(\S+)\s*$")
def find_candidates(lines):
"""[(heading_idx, end_idx, marker_idx, destination, heading_text)] —
end_idx is one past the subtree's last line."""
candidates = []
for i, line in enumerate(lines):
m = MARKER_RE.match(line)
if not m:
continue
head_idx = None
for j in range(i, -1, -1):
hm = HEADING_RE.match(lines[j])
if hm:
head_idx = j
level = len(hm.group(1))
heading = hm.group(2)
break
if head_idx is None:
continue
end = len(lines)
for k in range(head_idx + 1, len(lines)):
km = HEADING_RE.match(lines[k])
if km and len(km.group(1)) <= level:
end = k
break
candidates.append((head_idx, end, i, m.group(1), heading))
return candidates
def extract_handoff(lines, head_idx, end):
"""The subtree as handoff text: every :ROUTE_CANDIDATE: line dropped
(a marker is meaningless at the destination), empty drawers pruned,
headings promoted so the task is level 1."""
sub = [l for l in lines[head_idx:end] if not MARKER_RE.match(l)]
pruned = []
i = 0
while i < len(sub):
if sub[i].strip() == ":PROPERTIES:" and i + 1 < len(sub) and sub[i + 1].strip() == ":END:":
i += 2
continue
pruned.append(sub[i])
i += 1
shift = len(HEADING_RE.match(pruned[0]).group(1)) - 1
if shift > 0:
pruned = [l[shift:] if HEADING_RE.match(l) else l for l in pruned]
return "\n".join(pruned).rstrip() + "\n"
def send(destination, handoff_text, slug):
inbox_send = Path(__file__).with_name("inbox-send.py")
with tempfile.NamedTemporaryFile(
"w", suffix=".org", prefix=f"route-{slug}-", delete=False, encoding="utf-8"
) as tf:
tf.write(handoff_text)
tmp = tf.name
try:
result = subprocess.run(
[sys.executable, str(inbox_send), destination, "--file", tmp],
capture_output=True, text=True,
)
return result.returncode == 0, (result.stderr or result.stdout).strip()
finally:
os.unlink(tmp)
def main():
ap = argparse.ArgumentParser(prog="route-batch")
mode = ap.add_mutually_exclusive_group(required=True)
mode.add_argument("--list", action="store_true", dest="list_mode")
mode.add_argument("--go", action="store_true")
ap.add_argument("--todo", default="todo.org")
args = ap.parse_args()
todo_path = Path(args.todo)
if not todo_path.is_file():
return 0 # no todo file, no candidates
lines = todo_path.read_text(encoding="utf-8").splitlines()
candidates = find_candidates(lines)
# Two markers in one task's drawer are one candidate, not two: same span +
# same destination dedupes. Everything else that overlaps — a tagged child
# inside a tagged parent, one task tagged for two destinations — is a
# conflict: routing either span would silently take the other (or, with a
# stale end index, a bystander task) along. Conflicts are left in place
# and reported; the human untangles which project the pieces belong to.
deduped = []
for cand in candidates:
if not any(c[0] == cand[0] and c[1] == cand[1] and c[3] == cand[3] for c in deduped):
deduped.append(cand)
conflicted = set()
for a in deduped:
for b in deduped:
if a is not b and a[0] <= b[0] and b[1] <= a[1]:
conflicted.add(a)
conflicted.add(b)
routable = [c for c in deduped if c not in conflicted]
if not deduped:
return 0
if args.list_mode:
for _h, _e, _m, dest, heading in deduped:
flag = "\tCONFLICT (overlapping candidates — resolve by hand)" if (_h, _e, _m, dest, heading) in conflicted else ""
print(f"{dest}\t{heading}{flag}")
return 0
failures = 0
for _h, _e, _m, dest, heading in sorted(conflicted):
failures += 1
print(f"CONFLICT: {dest}\t{heading}\t(overlapping candidate subtrees — left in place, resolve by hand)")
# Bottom-up so earlier indices stay valid as subtrees are removed; the
# file is rewritten after every successful send so a crash mid-run never
# leaves an already-sent task still present locally.
for head_idx, end, _marker_idx, dest, heading in sorted(routable, reverse=True):
handoff = extract_handoff(lines, head_idx, end)
slug = re.sub(r"[^a-z0-9]+", "-", heading.lower()).strip("-")[:40] or "task"
ok, detail = send(dest, handoff, slug)
if ok:
del lines[head_idx:end]
todo_path.write_text("\n".join(lines).rstrip("\n") + "\n", encoding="utf-8")
print(f"routed: {dest}\t{heading}")
else:
failures += 1
print(f"FAILED: {dest}\t{heading}\t({detail})")
return 1 if failures else 0
if __name__ == "__main__":
sys.exit(main())
|
