HEARTBEAT.md queue — Diapason-style pending task drain.
Functions:
parse_heartbeat(
text: str,
) -> tuple[list[HeartbeatTask], list[HeartbeatTask]]
Parse ## Now and ## Watching task lists.
Source code in src/diapason/heartbeat/markdown.py
| def parse_heartbeat(text: str) -> tuple[list[HeartbeatTask], list[HeartbeatTask]]:
"""Parse ## Now and ## Watching task lists."""
lines = text.splitlines()
now: list[HeartbeatTask] = []
watching: list[HeartbeatTask] = []
for section, bucket in (("## Now", now), ("## Watching", watching)):
start, end = _section_bounds(lines, section)
if start < 0:
continue
for i in range(start, end):
m = _TASK_RE.match(lines[i])
if not m:
continue
indent, mark, body = m.group(1), m.group(2), m.group(3).strip()
if not body:
continue
bucket.append(
HeartbeatTask(
text=body,
done=mark.lower() == "x",
line_index=i,
indent=indent,
)
)
return now, watching
|
clear_done(workspace: Path | str | None = None) -> int
Remove checked items under ## Now. Returns count removed.
Source code in src/diapason/heartbeat/markdown.py
| def clear_done(workspace: Path | str | None = None) -> int:
"""Remove checked items under ## Now. Returns count removed."""
path = ensure_heartbeat_file(workspace)
lines = path.read_text(encoding="utf-8").splitlines()
start, end = _section_bounds(lines, "## Now")
if start < 0:
return 0
kept: list[str] = []
removed = 0
for i, line in enumerate(lines):
if start <= i < end:
m = _TASK_RE.match(line)
if m and m.group(2).lower() == "x":
removed += 1
continue
kept.append(line)
path.write_text("\n".join(kept) + "\n", encoding="utf-8")
return removed
|