diff options
Diffstat (limited to 'wflib/tasks.py')
| -rw-r--r-- | wflib/tasks.py | 728 |
1 files changed, 728 insertions, 0 deletions
diff --git a/wflib/tasks.py b/wflib/tasks.py new file mode 100644 index 0000000..80264c7 --- /dev/null +++ b/wflib/tasks.py @@ -0,0 +1,728 @@ +"""TASKS.md as data: parse into sections and items, change, render back. + +Pure text in, text out; no file access. Format: docs/design.md (section 3). +""" +from __future__ import annotations + +import difflib +import re +from dataclasses import dataclass, field + +SECTIONS = { + "awaiting": "Awaiting your decision", + "pending": "Pending", + "human": "Needs human", + "deferred": "Deferred", +} +TASK_KEYS = ("pending", "human", "deferred") +EFFORTS = ("<1h", "1h", "5h", "10h", "100h") +INDENT = " " + +ID_RE = re.compile(r"^[ta]-[a-z0-9]+(?:-[a-z0-9]+)*$") +LINK_RE = re.compile(r"\[\[([^\]|#]+)\]\]") +HEADER_RE = re.compile( + r"^- \*\*(?P<id>[^*\s]+)\*\*" + r"(?: \[P(?P<prio>\d)\])?" + r"(?: \((?!in progress:|blocked:)(?P<effort>[^)]*)\))?" + r"(?: \((?P<status>(?:in progress|blocked):[^)]*)\))?" + r": (?P<text>.*)$" +) +ITEM_START = "- **" +AFTER_RE = re.compile(r"^\s*(?:-\s+)?After:\s*(.*)$") +REF_RE = re.compile(r"^\s*(?:-\s+)?Ref:\s*(.*)$") +SLICES_RE = re.compile(r"^\s*-\s+Slices:\s*(.*)$") +MODEL_RE = re.compile(r"^\s*(?:-\s+)?Model:\s*(.*)$") +MODELS = ("haiku", "sonnet", "opus") # low → high; no Model line = opus +DONE_RE = re.compile(r"^\s*(?:-\s+)?Done:\s*(.*)$") +HUMAN_DONE_RE = re.compile( + r"\bowner(?:'s)?\s+(?:approves?|approval|tests?|checks?|runs?|says?|confirms?|verif(?:y|ies)|decides?|reviews?|signs?|sees?|ok)\b" + r"|\breport to\b|\bconfirm with\b|\bplease confirm\b|^\s*confirm\b", re.I) +SESSIONS_RE = re.compile(r"^\s*(?:-\s+)?Sessions:\s*(.*)$") +SESSIONS = ("parallel", "solo", "owner") # no Sessions line = parallel +CLOUD_RE = re.compile(r"^\s*(?:-\s+)?Cloud:\s*(.*)$") +CLOUDS = ("yes", "no") # no Cloud line = decided by the fit rules (wflib/cloud.py) +LEADING_ID_RE = re.compile(r"^([ta]-[a-z0-9]+(?:-[a-z0-9]+)*):\s+(.*)$") + + +class TaskError(ValueError): + pass + + +@dataclass +class Item: + id: str + prio: int | None = None + effort: str | None = None + interactive: bool = False + status: str | None = None + text: str = "" + body: list[str] = field(default_factory=list) + error: str | None = None + raw: str | None = None + line: int | None = None # 1-based line of the header in the parsed text + + def header(self) -> str: + if self.error: + return self.raw or "" + out = f"- **{self.id}**" + if self.prio is not None: + out += f" [P{self.prio}]" + if self.effort is not None: + out += f" ({self.effort}{', interactive' if self.interactive else ''})" + if self.status: + out += f" ({self.status})" + return f"{out}: {self.text}" + + def lines(self) -> list[str]: + return [self.header(), *self.body] + + @property + def title(self) -> str: + return self.text.split(". ", 1)[0].rstrip(".").strip() + + @property + def goal(self) -> str: + return self.text.split(". ", 1)[1].strip() if ". " in self.text else "" + + @property + def blocked_on(self) -> str | None: + if self.status and self.status.startswith("blocked:"): + found = LINK_RE.findall(self.status) + return found[0] if found else None + return None + + def _line(self, pattern: re.Pattern) -> int | None: + return next((i for i, l in enumerate(self.body) if pattern.match(l)), None) + + @property + def after(self) -> list[str]: + i = self._line(AFTER_RE) + return LINK_RE.findall(self.body[i]) if i is not None else [] + + @property + def slices(self) -> list[str]: + i = self._line(SLICES_RE) + return LINK_RE.findall(self.body[i]) if i is not None else [] + + @property + def model(self) -> str: + i = self._line(MODEL_RE) + words = MODEL_RE.match(self.body[i]).group(1).split() if i is not None else [] + return words[0] if words else "opus" + + @property + def done_text(self) -> str | None: + """Text of the Done line, None without one.""" + i = self._line(DONE_RE) + return DONE_RE.match(self.body[i]).group(1) if i is not None else None + + @property + def runner_ready(self) -> bool: + """Headless worker can finish it: has a Done line, not owner-bound, Done not about the owner/confirming.""" + done = self.done_text + return bool(done and done.strip()) and self.sessions != "owner" and not HUMAN_DONE_RE.search(done) + + @property + def human_done_match(self) -> str | None: + """The Done phrase that makes it not runner-ready (human action), None if none.""" + m = HUMAN_DONE_RE.search(self.done_text or "") + return m.group(0) if m else None + + @property + def sessions(self) -> str: + """parallel | solo (no other live session) | owner (owner present); header ', interactive' = owner.""" + i = self._line(SESSIONS_RE) + words = SESSIONS_RE.match(self.body[i]).group(1).split() if i is not None else [] + if words and words[0] in SESSIONS: + return words[0] + return "owner" if self.interactive else "parallel" + + @property + def cloud(self) -> str | None: + """yes | no | None (no Cloud line, or an unknown word).""" + i = self._line(CLOUD_RE) + words = CLOUD_RE.match(self.body[i]).group(1).split() if i is not None else [] + return words[0] if words and words[0] in CLOUDS else None + + @property + def refs(self) -> list[tuple[str, str | None]]: + i = self._line(REF_RE) + if i is None: + return [] + out = [] + for part in split_refs(REF_RE.match(self.body[i]).group(1)): + target = re.sub(r"\s*\(.*$", "", part).strip() + if target: + path, _, anchor = target.partition("#") + out.append((path, anchor or None)) + return out + + +def split_refs(text: str) -> list[str]: + """Split on commas that are not inside parentheses.""" + parts, depth, cur = [], 0, "" + for ch in text: + if ch == "(": + depth += 1 + elif ch == ")": + depth = max(0, depth - 1) + if ch == "," and depth == 0: + parts.append(cur) + cur = "" + else: + cur += ch + return [p.strip() for p in parts + [cur] if p.strip()] + + +@dataclass +class Section: + heading: str + key: str | None + prefix: list[str] = field(default_factory=list) + items: list[Item] = field(default_factory=list) + suffix: list[str] = field(default_factory=list) + line: int | None = None # 1-based line of the heading + suffix_line: int | None = None + + def lines(self) -> list[str]: + out = [f"## {self.heading}", *self.prefix] + for item in self.items: + out += [*item.lines(), ""] + return out + self.suffix + + +@dataclass +class Doc: + preamble: list[str] + sections: list[Section] + newline: str = "\n" + + def section(self, key: str) -> Section: + for s in self.sections: + if s.key == key: + return s + raise TaskError(f"no '## {SECTIONS[key]}' section") + + def all_items(self) -> list[Item]: + return [i for s in self.sections for i in s.items] + + def ids(self) -> set[str]: + return {i.id for i in self.all_items() if i.id} + + def find(self, id: str) -> tuple[Section, int]: + for s in self.sections: + for n, item in enumerate(s.items): + if item.id == id: + return s, n + raise TaskError(unknown_id(id, self.ids())) + + def item(self, id: str) -> Item: + s, n = self.find(id) + return s.items[n] + + +def unknown_id(id: str, known: set[str]) -> str: + near = difflib.get_close_matches(id, sorted(known), n=3, cutoff=0.0) + return f"unknown id '{id}'" + (f" (nearest: {', '.join(near)})" if near else "") + + +def parse_header(line: str) -> Item: + m = HEADER_RE.match(line) + if not m: + found = re.match(r"^- \*\*([^*]+)\*\*", line) + return Item(id=found.group(1).strip() if found else "", raw=line, + error="bad header: want '- **id** [Pn] (effort) [(status)]: Title. Goal.'") + effort, interactive = m.group("effort"), False + if effort is not None: + parts = [p.strip() for p in effort.split(",")] + interactive = "interactive" in parts[1:] + effort = parts[0] + status = m.group("status") + return Item(id=m.group("id"), prio=int(m.group("prio")) if m.group("prio") else None, + effort=effort, interactive=interactive, + status=status.strip() if status else None, text=m.group("text").strip()) + + +def _strip_blank_tail(lines: list[str]) -> list[str]: + while lines and not lines[-1].strip(): + lines.pop() + return lines + + +def _parse_section(heading: str, lines: list[str], at: int = 0) -> Section: + """`at` = 1-based line number of the heading.""" + key = next((k for k, h in SECTIONS.items() if h == heading), None) + section = Section(heading, key, line=at) + if key is None: + section.prefix = lines + return section + for n, line in enumerate(lines, at + 1): + if section.suffix: + section.suffix.append(line) + elif line.startswith(ITEM_START): + section.items.append(parse_header(line)) + section.items[-1].line = n + elif not section.items: + section.prefix.append(line) + elif line.strip() and not line[0].isspace(): + section.suffix.append(line) + section.suffix_line = n + else: + section.items[-1].body.append(line) + for item in section.items: + _strip_blank_tail(item.body) + return section + + +def parse(text: str) -> Doc: + newline = "\r\n" if "\r\n" in text else "\n" + lines = _strip_blank_tail(text.replace("\r\n", "\n").split("\n")) + starts = [i for i, l in enumerate(lines) if l.startswith("## ")] + doc = Doc(lines[: starts[0]] if starts else lines, [], newline) + for n, start in enumerate(starts): + end = starts[n + 1] if n + 1 < len(starts) else len(lines) + doc.sections.append(_parse_section(lines[start][3:].strip(), lines[start + 1:end], start + 1)) + return doc + + +def render(doc: Doc) -> str: + lines = list(doc.preamble) + for s in doc.sections: + lines += s.lines() + _strip_blank_tail(lines) + return doc.newline.join(lines) + doc.newline + + +# ---------------------------------------------------------------- ids + +def make_id(title: str, taken: set[str], prefix: str = "t-") -> str: + words = [w for w in re.sub(r"[^a-z0-9]+", " ", title.lower().encode("ascii", "ignore").decode()).split() if w] + slug = "" + for w in words: + longer = f"{slug}-{w}" if slug else w + if len(longer) > 40: + break + slug = longer + if not slug and words: + slug = words[0][:40] + if not slug: + raise TaskError(f"no id can be made from title '{title}': give one with --id") + base, n, id = prefix + slug, 2, prefix + slug + while id in taken: + id = f"{base}-{n}" + n += 1 + return id + + +def slice_id(parent: str, taken: set[str]) -> str: + n = 1 + while f"{parent}-{n}" in taken: + n += 1 + return f"{parent}-{n}" + + +def open_slices(doc: Doc, id: str) -> list[str]: + """Open items listed in the Slices: line of `id`, then open `<id>-N` items.""" + pattern = re.compile(re.escape(id) + r"-\d+$") + open_ = doc.ids() + named = [s for s in doc.item(id).slices if s in open_] + return named + [i.id for i in doc.all_items() if pattern.match(i.id) and i.id not in named] + + +def parent_of(doc: Doc, id: str) -> str | None: + for item in doc.all_items(): + if id in item.slices: + return item.id + m = re.match(r"(.+)-\d+$", id) + return m.group(1) if m and m.group(1) in doc.ids() else None + + +def rename(doc: Doc, old: str, new: str, taken: set[str]) -> int: + """Change the id of an open item and every [[old]] link; returns the link count. + `taken` = ids used in the archive.""" + item = doc.item(old) + if not ID_RE.match(new): + raise TaskError(f"bad id '{new}' (want t-… or a-…, lowercase a-z 0-9 -)") + if new[:2] != old[:2]: + raise TaskError(f"a rename keeps the kind ({old[:2]}…)") + if new in doc.ids(): + raise TaskError(f"id '{new}' already exists") + if new in taken: + raise TaskError(f"id '{new}' already used in the archive (ids are never reused)") + link, n = re.compile(r"\[\[" + re.escape(old) + r"\]\]"), 0 + for other in doc.all_items(): + other.text, k = link.subn(f"[[{new}]]", other.text) + n += k + if other.status: + other.status, k = link.subn(f"[[{new}]]", other.status) + n += k + for i, line in enumerate(other.body): + other.body[i], k = link.subn(f"[[{new}]]", line) + n += k + item.id = new + return n + + +# ---------------------------------------------------------------- placing + +def _prio(item: Item) -> int: + return 9 if item.prio is None else item.prio + + +def place(section: Section, item: Item) -> int: + """Insert rule: after the last item of the same or higher priority, and + after every item named in the new item's After:.""" + at = max((n + 1 for n, other in enumerate(section.items) if _prio(other) <= _prio(item)), default=0) + deps = set(item.after) + return max([at, *(n + 1 for n, other in enumerate(section.items) if other.id in deps)]) + + +def _check_kind(item: Item, key: str) -> None: + if key not in SECTIONS: + raise TaskError(f"unknown section '{key}' (want {', '.join(SECTIONS)})") + if item.id.startswith("a-") != (key == "awaiting"): + raise TaskError("a- items belong in Awaiting, t- items in Pending / Needs human / Deferred") + if key != "awaiting" and item.prio is None: + raise TaskError(f"task '{item.id}' needs a priority [P0]-[P3]") + + +def insert(doc: Doc, item: Item, key: str) -> None: + if item.error: + raise TaskError(item.error) + if item.id in doc.ids(): + raise TaskError(f"id '{item.id}' already exists") + _check_kind(item, key) + section = doc.section(key) + section.items.insert(place(section, item), item) + + +def remove(doc: Doc, id: str) -> Item: + section, n = doc.find(id) + return section.items.pop(n) + + +def _section_key(doc: Doc, id: str) -> str: + return doc.find(id)[0].key + + +def set_prio(doc: Doc, id: str, prio: int) -> None: + if prio not in (0, 1, 2, 3): + raise TaskError("priority 0-3") + key = _section_key(doc, id) + item = doc.item(id) + if item.id.startswith("a-"): + raise TaskError("Awaiting items have no priority") + remove(doc, id) + item.prio = prio + insert(doc, item, key) + + +def move_to(doc: Doc, id: str, key: str) -> None: + item = doc.item(id) + _check_kind(item, key) + doc.section(key) + remove(doc, id) + insert(doc, item, key) + + +def move_rel(doc: Doc, id: str, other: str, before: bool, force: bool = False) -> None: + if id == other: + raise TaskError("cannot move an item relative to itself") + item = doc.item(id) + target, _ = doc.find(other) + _check_kind(item, target.key) + source, n = doc.find(id) + source.items.pop(n) + at = next(i for i, o in enumerate(target.items) if o.id == other) + (0 if before else 1) + target.items.insert(at, item) + + def undo(): + target.items.pop(at) + source.items.insert(n, item) + + later = {o.id for o in target.items[at + 1:]} + for dep in item.after: + if dep in later: + undo() + raise TaskError(f"'{id}' is After: [[{dep}]], it cannot go before it") + for o in target.items[:at]: + if id in o.after: + undo() + raise TaskError(f"'{o.id}' is After: [[{id}]], '{id}' cannot go after it") + prev = target.items[at - 1] if at else None + nxt = target.items[at + 1] if at + 1 < len(target.items) else None + if not force and ((prev and _prio(prev) > _prio(item)) or (nxt and _prio(nxt) < _prio(item))): + undo() + raise TaskError(f"moving '{id}' there breaks priority order (wf prio, or --force)") + + +# ---------------------------------------------------------------- fields + +def set_status(doc: Doc, id: str, status: str | None) -> None: + item = doc.item(id) + if item.id.startswith("a-"): + raise TaskError("Awaiting items have no status") + if status: + status = status.strip() + if status.startswith("blocked:"): + on = LINK_RE.findall(status) + open_awaiting = {i.id for s in doc.sections if s.key == "awaiting" for i in s.items} + if len(on) != 1 or on[0] not in open_awaiting: + raise TaskError(f"'{on[0] if on else status}' is not an open Awaiting item") + elif not status.startswith("in progress:"): + raise TaskError("status is 'in progress: <note>' or 'blocked: [[a-id]]'") + if ")" in status or "(" in status: + raise TaskError("status note cannot contain parentheses (the status line ends in one); use - or :") + item.status = status or None + + +def _tail_start(item: Item) -> int: + """Index where the trailing Sessions:/Model:/After:/Ref: lines of the body begin.""" + n = len(item.body) + while n and any(p.match(item.body[n - 1]) for p in (AFTER_RE, REF_RE, MODEL_RE, SESSIONS_RE, CLOUD_RE)): + n -= 1 + return n + + +def _set_line(item: Item, pattern: re.Pattern, line: str | None, at_end: bool) -> None: + i = item._line(pattern) + if i is not None: + if line is None: + item.body.pop(i) + else: + item.body[i] = line + elif line is not None: + if at_end: + item.body.append(line) + else: + ref = item._line(REF_RE) + item.body.insert(len(item.body) if ref is None else ref, line) + + +def set_fields(doc: Doc, id: str, *, title: str | None = None, effort: str | None = None, + interactive: bool | None = None, after: list[str] | None = None, + refs: list[str] | None = None, model: str | None = None, + sessions: str | None = None, done: str | None = None, + cloud: str | None = None) -> None: + item = doc.item(id) + if done is not None: + done = done.strip() + if (i := item._line(DONE_RE)) is not None: + if done: + item.body[i] = INDENT + "Done: " + done + else: + item.body.pop(i) + elif done: + item.body.insert(_tail_start(item), INDENT + "Done: " + done) + if sessions is not None: + if sessions not in ("", *SESSIONS): + raise TaskError(f"sessions '{sessions}' (want {', '.join(SESSIONS)})") + item.interactive = False + line = INDENT + "Sessions: " + sessions if sessions else None + if (i := item._line(SESSIONS_RE)) is not None: + if line: + item.body[i] = line + else: + item.body.pop(i) + elif line: + m = item._line(MODEL_RE) + item.body.insert(m if m is not None and m >= _tail_start(item) else _tail_start(item), line) + if cloud is not None: + if cloud not in ("", *CLOUDS): + raise TaskError(f"cloud '{cloud}' (want {', '.join(CLOUDS)})") + line = INDENT + "Cloud: " + cloud if cloud else None + if (i := item._line(CLOUD_RE)) is not None: + if line: + item.body[i] = line + else: + item.body.pop(i) + elif line: + item.body.insert(_tail_start(item), line) + if model: + if model not in MODELS: + raise TaskError(f"model '{model}' (want {', '.join(MODELS)})") + line = INDENT + "Model: " + model + if (i := item._line(MODEL_RE)) is not None: + item.body[i] = line + else: + item.body.insert(_tail_start(item), line) + elif model is not None and (i := item._line(MODEL_RE)) is not None: + item.body.pop(i) + if title is not None: + title = title.strip().rstrip(".") + if not title: + raise TaskError("empty title") + if ". " in title or title.endswith(("?", "!")): + item.text = title if title.endswith(("?", "!")) else f"{title}." + else: + item.text = f"{title}. {item.goal}" if item.goal else f"{title}." + if effort is not None: + if effort not in EFFORTS: + raise TaskError(f"effort '{effort}' (want {', '.join(EFFORTS)})") + item.effort = effort + if interactive is not None: + item.interactive = interactive + if after is not None: + line = INDENT + "- After: " + ", ".join(f"[[{a}]]" for a in after) if after else None + _set_line(item, AFTER_RE, line, at_end=False) + if refs is not None: + _set_line(item, REF_RE, INDENT + "Ref: " + ", ".join(refs) if refs else None, at_end=True) + + +def add_note(doc: Doc, id: str, line: str) -> None: + item = doc.item(id) + line = line.strip() + if not line: + raise TaskError("empty note") + item.body.insert(_tail_start(item), f"{INDENT}- {line}") + + +def _link_after(line: str) -> str: + """'After: t-a, t-b' -> 'After: [[t-a]], [[t-b]]' (bare ids accepted, written as links).""" + m = re.match(r"^(\s*(?:-\s+)?After:\s*)(.*)$", line) + if not m: + return line + parts = [w for w in re.split(r"[\s,;]+", m.group(2)) if w] + if not parts or not all(LINK_RE.fullmatch(w) or ID_RE.match(w) for w in parts): + return line + return m.group(1) + ", ".join(w if w.startswith("[[") else f"[[{w}]]" for w in parts) + + +def indent_body(lines: list[str]) -> list[str]: + import textwrap + text = textwrap.dedent("\n".join(_link_after(l.rstrip()) for l in lines)) + return _strip_blank_tail([INDENT + l if l.strip() else "" for l in text.split("\n")]) if text.strip() else [] + + +def set_body(doc: Doc, id: str, lines: list[str]) -> None: + item = doc.item(id) + new = indent_body(lines) + kept = [l for l in item.body[_tail_start(item):] + if not any(p.match(l) and any(p.match(n) for n in new) for p in (AFTER_RE, REF_RE))] + item.body = new + kept + + +BOX_RE = re.compile(r"^(\s*- \[)([ xX])(\] )(.*)$") + + +def tick(doc: Doc, id: str, which: str) -> str: + item = doc.item(id) + boxes = [(i, BOX_RE.match(l)) for i, l in enumerate(item.body) if BOX_RE.match(l)] + if which.isdigit(): + if not 1 <= int(which) <= len(boxes): + raise TaskError(f"no box {which} in '{id}' ({len(boxes)} boxes)") + hits = [boxes[int(which) - 1]] + else: + hits = [(i, m) for i, m in boxes if which.lower() in m.group(4).lower()] + if not hits: + raise TaskError(f"no box matches '{which}' in '{id}'") + if len(hits) > 1: + raise TaskError(f"'{which}' matches {len(hits)} boxes in '{id}'") + i, m = hits[0] + if m.group(2) != " ": + raise TaskError(f"box already ticked: {m.group(4)}") + item.body[i] = f"{m.group(1)}x{m.group(3)}{m.group(4)}" + return m.group(4) + + +def unblock(doc: Doc, a_id: str) -> list[str]: + out = [] + for item in doc.all_items(): + if item.blocked_on == a_id: + item.status = None + out.append(item.id) + return out + + +# ---------------------------------------------------------------- pick + +def why_not(doc: Doc, item: Item, archived: set[str], held: dict[str, str] | None = None, + others: int = 0, owner: bool = True) -> str | None: + """Why a Pending item can't be picked now; None = pickable. held: id → holder of a live + claim by another session; others: other live sessions in the project; owner: owner present.""" + if item.error: + return "bad header" + if held and item.id in held: + return f"in progress by {held[item.id]}" + if item.sessions == "owner" and not owner: + return "owner: needs the owner (wf next --owner)" + if item.sessions == "solo" and others: + return f"solo: {others} other live session{'s' if others > 1 else ''}" + if item.status and item.status.startswith("blocked:"): + return f"blocked: {item.blocked_on}" + if open_slices(doc, item.id): + return "open slices: " + ", ".join(open_slices(doc, item.id)) + if [a for a in item.after if a not in archived]: + return "after: " + ", ".join(a for a in item.after if a not in archived) + return None + + +def waiters(doc: Doc) -> dict[str, set[str]]: + """Open task id → every open task (Pending, Needs human) that waits on it, transitively: + via After:, and a parent waits on its open slices.""" + items = {i.id: i for s in doc.sections if s.key in ("pending", "human") for i in s.items if not i.error} + direct: dict[str, set[str]] = {id: set() for id in items} + for item in items.values(): + for dep in item.after: + if dep in direct: + direct[dep].add(item.id) + for s in item.slices: + if s in direct: + direct[s].add(item.id) + out: dict[str, set[str]] = {} + for id in direct: + seen, todo = set(), list(direct[id]) + while todo: + w = todo.pop() + if w not in seen and w != id: + seen.add(w) + todo += direct[w] + out[id] = seen + return out + + +def solo_running(doc: Doc, held: dict[str, str]) -> tuple[str, str] | None: + """(id, holder) of a solo task in progress by another live session; held: see why_not.""" + return next(((i.id, held[i.id]) for i in doc.section("pending").items + if i.id in held and not i.error and i.sessions == "solo"), None) + + +# ---------------------------------------------------------------- archive + +ARCHIVE_ID_RE = re.compile(r"^- .*?\*\*([ta]-[a-z0-9-]+)\*\*") + + +def archive_ids(text: str) -> set[str]: + return {m.group(1) for l in text.replace("\r\n", "\n").split("\n") if (m := ARCHIVE_ID_RE.match(l))} + + +def archive_line(date: str, item: Item, entry: str) -> str: + entry = " ".join(entry.split()) + return f"- {date} **{item.id}** {item.title}" + (f" — {entry}" if entry else "") + + +def archive_prepend(text: str, line: str) -> str: + newline = "\r\n" if "\r\n" in text else "\n" + lines = _strip_blank_tail(text.replace("\r\n", "\n").split("\n")) + at = next((i for i, l in enumerate(lines) if l.startswith("- ")), None) + if at is None: + lines += ["", line] + else: + lines.insert(at, line) + return newline.join(lines) + newline + + +# ---------------------------------------------------------------- blocks + +def parse_block(text: str) -> Item: + lines = _strip_blank_tail(text.replace("\r\n", "\n").strip("\n").split("\n")) + if not lines or not lines[0].strip(): + raise TaskError("empty item") + first = lines[0].strip() + if re.match(r"^\d+\. ", first): + raise TaskError("old numbered format: write '- **id** [Pn] (effort): Title. Goal.'") + item = parse_header(first) + if item.error: + raise TaskError(item.error) + item.body = indent_body(lines[1:]) + return item |