scripts/tasks
2026-07-11 13:31:39 +02:00

701 lines
23 KiB
Python
Executable file

#!/usr/bin/env python3
"""
tasks — a terminal UI for managing your task files in a single store.
THE STORE
Tasks live under $TASKS_HOME (set from your `tasks_dir` config). Status is the
FOLDER a task lives in (backlog -> next -> in-progress -> done); priority and
context live in frontmatter; the FILENAME date is the *last-updated* date; the
`id:` is stable and never changes. This tool enforces those rules so they can't
drift — it re-dates filenames on every mutation, stamps the ## Log, uses `git mv`
(in whichever repo the file lives), and never touches `id:`.
USAGE
tasks Launch the TUI
tasks list [--status S] Print a table (headless)
tasks show <id|file> Print one task
tasks new --title T [--priority P2] [--context @admin]
[--project X] [--status backlog] [--goal ...] [--next ...]
[--notes ...] [--links ...]
tasks move <id|file> <status> Move between status folders (git mv)
tasks selftest Exercise the core logic on a temp store; print OK
tasks help
Env: TASKS_HOME = your tasks directory (from `tasks_dir` in your config)
"""
import curses
import datetime
import os
import re
import shutil
import subprocess
import sys
import textwrap
# ---------------------------------------------------------------------------
# Config / format constants (kept in sync with the task file format)
# ---------------------------------------------------------------------------
STATUSES = ["backlog", "next", "in-progress", "done"]
PRIORITIES = ["P0", "P1", "P2", "P3"]
CONTEXTS = ["@code", "@calls", "@errand", "@home", "@admin", "@waiting"]
# Canonical frontmatter field order (matches the task template).
FIELD_ORDER = [
"id", "title", "priority", "context", "project", "created", "links",
]
DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}-")
def tasks_dir():
root = os.environ.get("TASKS_HOME")
if not root:
sys.exit("tasks: TASKS_HOME is not set — point it at your tasks_dir "
"(see example.conf / ayo.sh).")
return root
def today():
return datetime.date.today().isoformat()
def ensure_dirs():
for s in STATUSES:
os.makedirs(os.path.join(tasks_dir(), s), exist_ok=True)
def slugify(title):
s = re.sub(r"[^a-z0-9]+", "-", title.strip().lower()).strip("-")
return s or "task"
# ---------------------------------------------------------------------------
# Task model — parse / serialize
# ---------------------------------------------------------------------------
class Task:
def __init__(self, path):
self.path = path
self.status = os.path.basename(os.path.dirname(path))
self.fields = {} # frontmatter key -> value (strings)
self.body = "" # everything after the frontmatter block
self._parse()
@property
def filename(self):
return os.path.basename(self.path)
@property
def slug(self):
name = self.filename[:-3] if self.filename.endswith(".md") else self.filename
return DATE_RE.sub("", name)
def _parse(self):
try:
with open(self.path, "r", encoding="utf-8") as fh:
text = fh.read()
except OSError:
return
if text.startswith("---"):
parts = text.split("\n")
end = None
for i in range(1, len(parts)):
if parts[i].strip() == "---":
end = i
break
if end is not None:
for line in parts[1:end]:
m = re.match(r"^([A-Za-z0-9_-]+):\s?(.*)$", line)
if m:
self.fields[m.group(1)] = m.group(2).strip()
self.body = "\n".join(parts[end + 1:]).lstrip("\n")
return
self.body = text
# --- convenience accessors -------------------------------------------
def get(self, key, default=""):
return self.fields.get(key, default)
@property
def title(self):
return self.get("title") or "(no title)"
@property
def priority(self):
return self.get("priority") or "?"
@property
def context(self):
return self.get("context") or "-"
def prio_rank(self):
try:
return PRIORITIES.index(self.priority)
except ValueError:
return len(PRIORITIES)
# --- serialization ----------------------------------------------------
def serialize(self):
lines = ["---"]
for key in FIELD_ORDER:
lines.append(f"{key}: {self.fields.get(key, '')}".rstrip())
for key, val in self.fields.items(): # preserve any extra keys
if key not in FIELD_ORDER:
lines.append(f"{key}: {val}".rstrip())
lines.append("---")
body = self.body.rstrip("\n")
return "\n".join(lines) + ("\n\n" + body if body else "") + "\n"
def save(self):
with open(self.path, "w", encoding="utf-8") as fh:
fh.write(self.serialize())
# --- mutations (each re-dates the filename to today) ------------------
def _target_name(self, when=None):
return f"{when or today()}-{self.slug}.md"
def redate(self):
"""Rename the file so its date prefix == today (the last-updated rule)."""
new_name = self._target_name()
if new_name == self.filename:
return
newpath = _unique_path(os.path.join(os.path.dirname(self.path), new_name),
keep=self.path)
_rename(self.path, newpath)
self.path = newpath
def set_field(self, key, value):
self.fields[key] = value
self.save()
self.redate()
def append_log(self, msg):
entry = f"- {today()}{msg}"
if re.search(r"(?m)^##\s+Log\s*$", self.body):
self.body = self.body.rstrip("\n") + "\n" + entry
else:
self.body = self.body.rstrip("\n") + f"\n\n## Log\n{entry}"
self.save()
self.redate()
def move_to(self, new_status):
if new_status not in STATUSES:
raise ValueError(f"unknown status: {new_status}")
if new_status == self.status:
return
dest_dir = os.path.join(tasks_dir(), new_status)
os.makedirs(dest_dir, exist_ok=True)
newpath = _unique_path(os.path.join(dest_dir, self._target_name()))
_rename(self.path, newpath)
self.path = newpath
self.status = new_status
self.append_log(f"moved to {new_status}.")
def _unique_path(path, keep=None):
"""Avoid clobbering an existing file; `keep` is the path allowed to match."""
if not os.path.exists(path) or path == keep:
return path
base, ext = os.path.splitext(path)
i = 2
while os.path.exists(f"{base}-{i}{ext}"):
i += 1
return f"{base}-{i}{ext}"
def _git_root(path):
try:
out = subprocess.run(
["git", "-C", os.path.dirname(path), "rev-parse", "--show-toplevel"],
capture_output=True, text=True, check=True)
return out.stdout.strip()
except (subprocess.CalledProcessError, FileNotFoundError):
return None
def _rename(old, new):
"""Prefer `git mv` in whichever repo the file lives; else a plain rename."""
if old == new:
return
root = _git_root(old)
if root:
try:
subprocess.run(["git", "-C", root, "mv", old, new],
check=True, capture_output=True)
return
except (subprocess.CalledProcessError, FileNotFoundError):
pass # untracked file — fall through to a plain rename
os.rename(old, new)
def load_all():
ensure_dirs()
tasks = []
for s in STATUSES:
d = os.path.join(tasks_dir(), s)
if not os.path.isdir(d):
continue
for name in sorted(os.listdir(d)):
if name.endswith(".md") and not name.startswith("_"):
tasks.append(Task(os.path.join(d, name)))
return tasks
def find_one(ref):
"""Resolve a task by `id:`, filename, slug, or path."""
if os.path.isfile(ref):
return Task(ref)
for t in load_all():
if t.get("id") == ref or t.filename == ref or t.slug == ref:
return t
return None
def create_task(title, priority="P2", context="@admin",
project="", status="backlog", goal="", next_action="",
notes="", links=""):
ensure_dirs()
if status not in STATUSES:
raise ValueError(f"unknown status: {status}")
slug = slugify(title)
tid = f"{today()}-{slug}"
path = _unique_path(os.path.join(tasks_dir(), status, f"{today()}-{slug}.md"))
t = Task.__new__(Task)
t.path = path
t.status = status
t.fields = {
"id": tid, "title": title, "priority": priority, "context": context,
"project": project, "created": today(), "links": links or "[]",
}
t.body = (
"## Goal\n" + (goal or "What done looks like, in one or two sentences.") + "\n\n"
"## Next action\n" + (next_action or "The single concrete next step — a physical verb.") + "\n\n"
"## Notes\n" + (notes or "Context, sub-steps, blockers.") + "\n\n"
"## Log\n" + f"- {today()} — created.\n"
)
t.save()
return t
# ---------------------------------------------------------------------------
# Headless CLI (scriptable + testable without a TTY)
# ---------------------------------------------------------------------------
def cli_list(argv):
status = None
it = iter(argv)
for a in it:
if a == "--status":
status = next(it, None)
tasks = [t for t in load_all() if not status or t.status == status]
tasks.sort(key=lambda t: (STATUSES.index(t.status), t.prio_rank()))
print(f"{'PRI':3} {'STATUS':11} {'CONTEXT':9} TASK")
for t in tasks:
print(f"{t.priority:3} {t.status:11} {t.context:9} {t.title}")
print(f" {t.path}")
return 0
def cli_show(argv):
if not argv:
print("usage: tasks show <id|file>", file=sys.stderr)
return 2
t = find_one(argv[0])
if not t:
print(f"not found: {argv[0]}", file=sys.stderr)
return 1
print(t.serialize())
return 0
def cli_new(argv):
kw, it = {}, iter(argv)
keymap = {"--title": "title", "--priority": "priority",
"--context": "context", "--project": "project", "--status": "status",
"--goal": "goal", "--next": "next_action", "--notes": "notes",
"--links": "links"}
for a in it:
if a in keymap:
kw[keymap[a]] = next(it, "")
if not kw.get("title"):
print("usage: tasks new --title T [--priority P2] [--context @admin] ...",
file=sys.stderr)
return 2
t = create_task(**kw)
print(f"created: {t.path}")
return 0
def cli_move(argv):
if len(argv) < 2:
print("usage: tasks move <id|file> <status>", file=sys.stderr)
return 2
t = find_one(argv[0])
if not t:
print(f"not found: {argv[0]}", file=sys.stderr)
return 1
if argv[1] not in STATUSES:
print(f"unknown status: {argv[1]} (choose from {', '.join(STATUSES)})",
file=sys.stderr)
return 2
t.move_to(argv[1])
print(f"moved -> {t.path}")
return 0
def cli_selftest(argv):
import tempfile
base = tempfile.mkdtemp(prefix="tasks-selftest-")
os.environ["TASKS_HOME"] = os.path.join(base, "tasks")
os.makedirs(os.environ["TASKS_HOME"], exist_ok=True)
subprocess.run(["git", "-C", base, "init", "-q"], check=False, capture_output=True)
try:
t = create_task("Book the blood-sugar appointment", priority="P1",
context="@calls", status="backlog", goal="Appt booked",
next_action="Call the clinic")
assert t.get("id").endswith("-book-the-blood-sugar-appointment"), t.get("id")
assert t.path.startswith(os.environ["TASKS_HOME"]), t.path
original_id = t.get("id")
t.set_field("priority", "P0") # update field -> re-date, id kept
assert t.priority == "P0" and t.get("id") == original_id
assert t.filename.startswith(today()), t.filename
t.append_log("bumped to P0.")
assert "bumped to P0." in Task(t.path).body
t.move_to("in-progress") # move -> folder + log stamp
assert t.status == "in-progress"
assert "moved to in-progress." in Task(t.path).body
assert len(load_all()) == 1
print("selftest: OK (create/redate/log/move pass)")
return 0
except AssertionError as e:
print(f"selftest: FAIL — {e}", file=sys.stderr)
return 1
finally:
shutil.rmtree(base, ignore_errors=True)
# ---------------------------------------------------------------------------
# curses TUI
# ---------------------------------------------------------------------------
HELP_TEXT = [
"tasks — keys",
"",
" j / ↓ down k / ↑ up",
" Tab cycle status filter (all → each folder)",
" s cycle sort (status+priority / priority / updated)",
" Enter view full task",
" c create (opens $EDITOR pre-filled from the template)",
" o open in $EDITOR (re-dates filename, logs 'edited')",
" move: e / b → backlog n → next i → in-progress d → done",
" p cycle priority @ set context",
" l add a log entry",
" r reload from disk",
" ? this help q quit",
"",
" Status = folder · filename date = last-updated · id never changes.",
"",
" (press any key)",
]
class TUI:
def __init__(self, stdscr):
self.scr = stdscr
self.tasks = []
self.sel = 0
self.top = 0
self.filter_status = "in-progress" # what you're actively working on, first
self.sort_mode = 0
self.msg = ""
self.reload()
# --- data ------------------------------------------------------------
def reload(self):
self.tasks = load_all()
self._sort()
self.sel = min(self.sel, max(0, len(self.visible()) - 1))
def _sort(self):
if self.sort_mode == 1:
self.tasks.sort(key=lambda t: (t.prio_rank(), t.title.lower()))
elif self.sort_mode == 2:
self.tasks.sort(key=lambda t: t.filename, reverse=True)
else:
self.tasks.sort(key=lambda t: (STATUSES.index(t.status), t.prio_rank()))
def visible(self):
if self.filter_status is None:
return self.tasks
return [t for t in self.tasks if t.status == self.filter_status]
def current(self):
vis = self.visible()
return vis[self.sel] if vis and 0 <= self.sel < len(vis) else None
# --- drawing ---------------------------------------------------------
def draw(self):
self.scr.erase()
h, w = self.scr.getmaxyx()
vis = self.visible()
flt = self.filter_status or "all"
sortname = ["status+prio", "priority", "updated"][self.sort_mode]
header = f" tasks filter:{flt} sort:{sortname} {len(vis)} tasks "
self.scr.addnstr(0, 0, header.ljust(w), w - 1, curses.A_REVERSE)
list_h = max(3, (h - 4) * 3 // 5)
if self.sel < self.top:
self.top = self.sel
elif self.sel >= self.top + list_h:
self.top = self.sel - list_h + 1
row = 1
for i in range(self.top, min(len(vis), self.top + list_h)):
t = vis[i]
line = f" {t.priority:2} {t.status:11} {t.context:8} {t.title}"
attr = curses.A_REVERSE if i == self.sel else curses.A_NORMAL
self.scr.addnstr(row, 0, line.ljust(w - 1), w - 1, attr)
row += 1
div = list_h + 1
if div < h - 1:
self.scr.addnstr(div, 0, "" * (w - 1), w - 1)
self._draw_preview(div + 1, h - 2, w)
footer = self.msg or " c:create o:open b/e:backlog n:next i:in-prog d:done p:prio @:ctx l:log Enter:view ?:help q:quit "
self.scr.addnstr(h - 1, 0, footer.ljust(w - 1), w - 1, curses.A_REVERSE)
self.scr.refresh()
def _draw_preview(self, y0, y1, w):
t = self.current()
if not t:
self.scr.addnstr(y0, 0, " (no task selected)", w - 1)
return
lines = [f" {t.filename}", f" id: {t.get('id')} created: {t.get('created')}", ""]
for bl in t.body.splitlines():
if bl.strip():
lines.extend(" " + s for s in textwrap.wrap(bl, w - 4))
else:
lines.append("")
y = y0
for ln in lines:
if y >= y1:
break
self.scr.addnstr(y, 0, ln, w - 1)
y += 1
# --- input helpers ---------------------------------------------------
def prompt(self, label, default=""):
h, w = self.scr.getmaxyx()
curses.echo()
curses.curs_set(1)
self.scr.addnstr(h - 1, 0, (" " + label + " ").ljust(w - 1), w - 1, curses.A_REVERSE)
try:
raw = self.scr.getstr(h - 1, len(label) + 2, w - len(label) - 4)
val = raw.decode("utf-8").strip()
except Exception:
val = ""
curses.noecho()
curses.curs_set(0)
return val or default
def choose(self, label, options):
"""Single-key chooser: shows options[0..], returns chosen or None."""
h, w = self.scr.getmaxyx()
opts = " ".join(f"{i+1}:{o}" for i, o in enumerate(options))
self.scr.addnstr(h - 1, 0, f" {label} {opts} (Esc cancel) ".ljust(w - 1),
w - 1, curses.A_REVERSE)
self.scr.refresh()
ch = self.scr.getch()
if options and ord("1") <= ch <= ord("1") + min(9, len(options)) - 1:
idx = ch - ord("1")
if 0 <= idx < len(options):
return options[idx]
return None
def flash(self, msg):
self.msg = " " + msg + " "
# --- actions ---------------------------------------------------------
def _run_editor(self, path):
editor = os.environ.get("EDITOR") or shutil.which("nano") or shutil.which("vi") or "vi"
curses.endwin()
subprocess.call([editor, path])
self.scr = curses.initscr()
curses.curs_set(0)
def act_create(self):
title = self.prompt("New task title:")
if not title:
self.flash("cancelled")
return
t = create_task(title) # template-filled draft in backlog/
self._run_editor(t.path) # flesh it out in $EDITOR
self.reload()
self.flash(f"created {t.filename}")
def act_open(self):
t = self.current()
if not t:
return
self._run_editor(t.path)
Task(t.path).append_log("edited.") # re-dates filename, logs 'edited'
self.reload()
self.flash("saved & re-dated")
def act_move_to(self, status):
t = self.current()
if not t:
return
if t.status == status:
self.flash(f"already in {status}")
return
t.move_to(status)
self.reload()
self.flash(f"moved -> {status}")
def act_field(self, key, options=None):
t = self.current()
if not t:
return
val = self.choose(f"Set {key}", options) if options else self.prompt(f"Set {key}:", t.get(key))
if val is None:
return
t.set_field(key, val)
self.reload()
self.flash(f"{key} = {val}")
def act_cycle_priority(self):
t = self.current()
if not t:
return
try:
nxt = PRIORITIES[(PRIORITIES.index(t.priority) + 1) % len(PRIORITIES)]
except ValueError:
nxt = "P2"
t.set_field("priority", nxt)
self.reload()
self.flash(f"priority = {nxt}")
def act_log(self):
t = self.current()
if not t:
return
msg = self.prompt("Log entry:")
if msg:
t.append_log(msg)
self.reload()
self.flash("logged")
def view_full(self):
t = self.current()
if not t:
return
self.scr.erase()
h, w = self.scr.getmaxyx()
y = 0
for ln in t.serialize().splitlines():
if y >= h - 1:
break
self.scr.addnstr(y, 0, ln[:w - 1], w - 1)
y += 1
self.scr.addnstr(h - 1, 0, " (press any key) ".ljust(w - 1), w - 1, curses.A_REVERSE)
self.scr.getch()
def help(self):
self.scr.erase()
h, w = self.scr.getmaxyx()
for i, ln in enumerate(HELP_TEXT):
if i >= h - 1:
break
self.scr.addnstr(i, 2, ln, w - 3)
self.scr.getch()
# --- main loop -------------------------------------------------------
def run(self):
curses.curs_set(0)
while True:
self.draw()
self.msg = ""
ch = self.scr.getch()
vis = self.visible()
if ch == ord("q"):
break
elif ch in (ord("j"), curses.KEY_DOWN):
self.sel = min(self.sel + 1, max(0, len(vis) - 1))
elif ch in (ord("k"), curses.KEY_UP):
self.sel = max(self.sel - 1, 0)
elif ch == ord("\t"):
order = [None] + STATUSES
self.filter_status = order[(order.index(self.filter_status) + 1) % len(order)]
self.sel = 0
elif ch == ord("s"):
self.sort_mode = (self.sort_mode + 1) % 3
self._sort()
elif ch in (curses.KEY_ENTER, 10, 13):
self.view_full()
elif ch == ord("c"):
self.act_create()
elif ch == ord("o"):
self.act_open()
elif ch in (ord("e"), ord("b")):
self.act_move_to("backlog")
elif ch == ord("n"):
self.act_move_to("next")
elif ch == ord("i"):
self.act_move_to("in-progress")
elif ch == ord("d"):
self.act_move_to("done")
elif ch == ord("p"):
self.act_cycle_priority()
elif ch == ord("@"):
self.act_field("context", CONTEXTS)
elif ch == ord("l"):
self.act_log()
elif ch == ord("r"):
self.reload()
self.flash("reloaded")
elif ch == ord("?"):
self.help()
def run_tui():
if not sys.stdout.isatty():
print("tasks: not a TTY — use the headless subcommands "
"(list/show/new/move) or run in a terminal.", file=sys.stderr)
return 2
curses.wrapper(lambda scr: TUI(scr).run())
return 0
# ---------------------------------------------------------------------------
# entry point
# ---------------------------------------------------------------------------
def main(argv):
if not argv:
return run_tui()
cmd, rest = argv[0], argv[1:]
dispatch = {
"list": cli_list, "show": cli_show, "new": cli_new,
"move": cli_move, "selftest": cli_selftest,
}
if cmd in dispatch:
return dispatch[cmd](rest)
if cmd in ("help", "-h", "--help"):
print(__doc__)
return 0
if cmd in ("tui", "ui"):
return run_tui()
print(f"tasks: unknown command '{cmd}' (try 'help')", file=sys.stderr)
return 2
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))