diff options
Diffstat (limited to 'scripts')
| -rw-r--r-- | scripts/README.org | 56 | ||||
| -rw-r--r-- | scripts/import_org_backlog.py | 301 | ||||
| -rw-r--r-- | scripts/tests/test_import_org_backlog.py | 163 |
3 files changed, 520 insertions, 0 deletions
diff --git a/scripts/README.org b/scripts/README.org index e6bacd2..5d5ab6d 100644 --- a/scripts/README.org +++ b/scripts/README.org @@ -82,6 +82,62 @@ the repo. A convenient way to pass it: default new-issue state at Triage (the intake column) after the columns reconcile, so a fresh issue lands there. +* import_org_backlog.py + +** Purpose + +Imports an org-mode backlog into the Pearl Linear workspace. It reads the +level-2 =TODO=/=DOING= headings under the "Pearl Open Work" section and creates +one Linear issue per task, carrying the priority, tags, and body across. + +The mapping: + +- priority cookie =[#A]= .. =[#D]= becomes Linear priority Urgent .. Low, +- a =[#D]= "someday" task lands in Icebox, everything above it in Backlog, +- tags become labels (missing ones are created); the =discuss=, =next=, + =cleanup=, and =pearl= workflow markers are dropped rather than turned into + labels, +- the heading body becomes the issue description, minus the property drawer and + any sub-entries. + +It's idempotent: a task whose title already exists is skipped, so a re-run only +fills in what's missing. The two umbrella headings (the dogfooding parent and +the manual-testing checklist) and everything under the Resolved section are left +out. + +** Usage + +The script reads the Linear API key from =LINEAR_API_KEY= and takes the org file +as its argument. + +#+begin_src bash + # preview the tasks without creating anything + python3 scripts/import_org_backlog.py todo.org --dry-run + + # run it + LINEAR_API_KEY="$(tr -d '[:space:]' < apikey.txt)" python3 scripts/import_org_backlog.py todo.org +#+end_src + +Run the tests with pytest: + +#+begin_src bash + pytest scripts/tests/test_import_org_backlog.py +#+end_src + +** Notes + +- *Dry-run first.* =--dry-run= parses the file and prints each task with its + target state and surviving labels, plus the full label set it would create. It + touches the network only on a real run, so it's the safe way to confirm scope + before importing. + +- *Idempotent by title.* The skip check matches on exact issue title. Editing a + task's title in the org file and re-running creates a second issue rather than + updating the first, so treat titles as stable once imported. + +- *Tags are matched case-insensitively.* An existing =Bug= label is reused for a + =:bug:= tag; only genuinely new labels are created. + * coverage-summary.el ** Purpose diff --git a/scripts/import_org_backlog.py b/scripts/import_org_backlog.py new file mode 100644 index 0000000..4ea2406 --- /dev/null +++ b/scripts/import_org_backlog.py @@ -0,0 +1,301 @@ +#!/usr/bin/env python3 +"""Import an org-mode backlog into the Pearl Linear workspace. + +The org file keeps open tasks under a "Pearl Open Work" section as level-2 +TODO/DOING headings, each with a priority cookie, optional tags, and a body. +This brings them into Linear as issues. + +The work splits into a pure parser/mapper (no I/O) and a thin GraphQL client: + + - parse_open_work_tasks pulls the level-2 tasks out of the section, dropping + the two meta umbrellas (the dogfooding parent and the manual-testing + checklist) and everything under the Resolved section, + - label_names_needed / task_to_issue turn a task into an issueCreate input -- + priority A-D maps to Linear 1-4, a [#D] task lands in Icebox and the rest in + Backlog, and tags become label ids (the discuss/next/cleanup/pearl tags are + dropped on the way in), + - import_backlog creates any missing labels, then the issues, skipping a task + whose title already exists so a re-run is safe. + +Usage: + LINEAR_API_KEY=lin_api_... python3 import_org_backlog.py todo.org [--dry-run] +""" + +import argparse +import json +import os +import re +import sys +import urllib.request + +LINEAR_GRAPHQL_URL = "https://api.linear.app/graphql" + +TEAM_NAME = "Pearl" +PROJECT_NAME = "Pearl" +SECTION_HEADER = "Pearl Open Work" + +# Umbrella headings that organize other work rather than being tasks themselves. +SKIP_TITLES = {"Personal Account Dogfooding", "Manual testing and validation"} + +# Tags that carry no meaning as Linear labels -- workflow markers, not topics. +DROPPED_TAGS = {"discuss", "next", "cleanup", "pearl"} + +PRIORITY_MAP = {"A": 1, "B": 2, "C": 3, "D": 4} + +# A [#D] "someday" task is icebox material; everything above it is backlog. +ICEBOX_PRIORITY = "D" + +LABEL_COLOR = "#bec2c8" + +_HEADING_RE = re.compile(r"^\*\*\s+(TODO|DOING)\s+(.*)$") +_PRIORITY_RE = re.compile(r"^\[#([A-Z])\]\s*") +_TAGS_RE = re.compile(r"\s+(:[\w@#%]+(?::[\w@#%]+)*:)\s*$") + + +# --- pure: parser ------------------------------------------------------------ + +def _parse_heading(rest): + """Split a heading's text into (priority, title, tags). + + REST is everything after the TODO/DOING keyword. The priority cookie leads, + the tags trail; the title is what's left between them. The tag regex anchors + to the line end so a colon-laden title (e.g. an =#+title:= reference) isn't + mistaken for tags. + """ + rest = rest.strip() + priority = None + pm = _PRIORITY_RE.match(rest) + if pm: + priority = pm.group(1) + rest = rest[pm.end():] + tags = [] + tm = _TAGS_RE.search(rest) + if tm: + tags = [t for t in tm.group(1).split(":") if t] + rest = rest[:tm.start()] + return priority, rest.strip(), tags + + +def _clean_body(lines): + """Join body LINES, dropping the :PROPERTIES: drawer and trimming blanks.""" + out, in_drawer = [], False + for line in lines: + s = line.strip() + if not in_drawer and s == ":PROPERTIES:": + in_drawer = True + continue + if in_drawer: + if s == ":END:": + in_drawer = False + continue + out.append(line.rstrip()) + return "\n".join(out).strip() + + +def parse_open_work_tasks(text): + """Return the open-work tasks as dicts: keyword, priority, title, tags, body. + + Only level-2 TODO/DOING headings under the "Pearl Open Work" section are + returned. Sub-entries (level 3+), the property drawer, the two meta + umbrellas, and anything after the section ends are excluded. + """ + lines = text.splitlines() + start = None + for i, line in enumerate(lines): + if re.match(r"^\*\s", line) and line.lstrip("* ").strip() == SECTION_HEADER: + start = i + 1 + break + if start is None: + return [] + end = len(lines) + for i in range(start, len(lines)): + if re.match(r"^\*\s", lines[i]): # next level-1 heading ends the section + end = i + break + region = lines[start:end] + + tasks = [] + i = 0 + while i < len(region): + m = _HEADING_RE.match(region[i]) + if not m: + i += 1 + continue + priority, title, tags = _parse_heading(m.group(2)) + j = i + 1 + body_lines = [] + while j < len(region) and not re.match(r"^\*+\s", region[j]): + body_lines.append(region[j]) + j += 1 + if title not in SKIP_TITLES: + tasks.append({"keyword": m.group(1), "priority": priority, + "title": title, "tags": tags, "body": _clean_body(body_lines)}) + i = j + return tasks + + +# --- pure: label set + mapping ----------------------------------------------- + +def label_names_needed(tasks): + """Return the set of Linear label names the TASKS need (capitalized tags).""" + return {tag.capitalize() + for task in tasks for tag in task["tags"] if tag not in DROPPED_TAGS} + + +def task_to_issue(task, states, label_ids): + """Map a TASK to an issueCreate input. + + STATES maps state name -> id (needs Backlog and Icebox); LABEL_IDS maps a + lowercase tag -> label id. Dropped tags and tags without a known label id + are left off. + """ + state = "Icebox" if task["priority"] == ICEBOX_PRIORITY else "Backlog" + return { + "title": task["title"], + "description": task["body"], + "priority": PRIORITY_MAP.get(task["priority"], 0), + "stateId": states[state], + "labelIds": [label_ids[t] for t in task["tags"] + if t not in DROPPED_TAGS and t in label_ids], + } + + +# --- boundary: a thin GraphQL client ----------------------------------------- + +class LinearError(RuntimeError): + """A GraphQL request returned an errors array.""" + + +def _http_transport(api_key): + def transport(query, variables): + body = json.dumps({"query": query, "variables": variables}).encode() + req = urllib.request.Request( + LINEAR_GRAPHQL_URL, data=body, + headers={"Authorization": api_key, "Content-Type": "application/json"}) + with urllib.request.urlopen(req) as resp: + return json.loads(resp.read().decode()) + return transport + + +class LinearClient: + """Thin Linear GraphQL client; the only network boundary in this module. + + Each method names its GraphQL operation distinctly (Teams, Projects, + ProjectIssues, LabelCreate, IssueCreate) so a fake transport can route on + the operation name. + """ + + def __init__(self, api_key, transport=None): + self._transport = transport or _http_transport(api_key) + + def execute(self, query, variables): + resp = self._transport(query, variables) + if resp.get("errors"): + raise LinearError("; ".join(e.get("message", str(e)) for e in resp["errors"])) + return resp["data"] + + def find_team(self): + data = self.execute( + "query Teams { teams { nodes { id name " + "labels { nodes { id name } } states { nodes { id name } } } } }", {}) + nodes = data["teams"]["nodes"] + return next((t for t in nodes if t.get("name") == TEAM_NAME), nodes[0]) + + def find_project(self): + data = self.execute("query Projects { projects { nodes { id name } } }", {}) + nodes = data["projects"]["nodes"] + return next((p for p in nodes if p.get("name") == PROJECT_NAME), + nodes[0] if nodes else None) + + def project_issue_titles(self, project_id): + data = self.execute( + "query ProjectIssues($id: String!) " + "{ project(id: $id) { issues { nodes { title } } } }", {"id": project_id}) + return [n["title"] for n in data["project"]["issues"]["nodes"]] + + def create_label(self, team_id, name): + data = self.execute( + "mutation LabelCreate($input: IssueLabelCreateInput!) " + "{ issueLabelCreate(input: $input) { success issueLabel { id } } }", + {"input": {"teamId": team_id, "name": name, "color": LABEL_COLOR}}) + return data["issueLabelCreate"]["issueLabel"]["id"] + + def create_issue(self, team_id, project_id, issue): + data = self.execute( + "mutation IssueCreate($input: IssueCreateInput!) " + "{ issueCreate(input: $input) { success issue { id } } }", + {"input": {"teamId": team_id, "projectId": project_id, **issue}}) + return data["issueCreate"]["issue"]["id"] + + +def import_backlog(client, text): + """Create labels and issues for the org backlog in TEXT; idempotent by title. + + Returns {created, skipped, labels_created} -- lists of titles created, + titles skipped (already present), and label names freshly created. + """ + tasks = parse_open_work_tasks(text) + team = client.find_team() + team_id = team["id"] + + # Existing labels, matched case-insensitively; create the ones we lack. + label_ids = {l["name"].lower(): l["id"] + for l in team.get("labels", {}).get("nodes", [])} + labels_created = [] + for name in sorted(label_names_needed(tasks)): + if name.lower() not in label_ids: + label_ids[name.lower()] = client.create_label(team_id, name) + labels_created.append(name) + + states = {s["name"]: s["id"] for s in team.get("states", {}).get("nodes", [])} + + project = client.find_project() + project_id = project["id"] + existing = set(client.project_issue_titles(project_id)) + + created, skipped = [], [] + for task in tasks: + if task["title"] in existing: + skipped.append(task["title"]) + continue + client.create_issue(team_id, project_id, task_to_issue(task, states, label_ids)) + created.append(task["title"]) + return {"created": created, "skipped": skipped, "labels_created": labels_created} + + +def main(argv=None): + parser = argparse.ArgumentParser(description="Import an org backlog into Pearl Linear.") + parser.add_argument("orgfile", help="path to the org file (e.g. todo.org)") + parser.add_argument("--dry-run", action="store_true", + help="print the tasks that would be imported without creating anything") + args = parser.parse_args(argv) + + with open(args.orgfile, encoding="utf-8") as fh: + text = fh.read() + + if args.dry_run: + tasks = parse_open_work_tasks(text) + print(f"Would import {len(tasks)} task(s):") + for t in tasks: + state = "Icebox" if t["priority"] == ICEBOX_PRIORITY else "Backlog" + kept = [tag for tag in t["tags"] if tag not in DROPPED_TAGS] + print(f" [#{t['priority']}] -> {state:7} {t['title']}" + + (f" :{':'.join(kept)}:" if kept else "")) + print(f"Labels needed: {sorted(label_names_needed(tasks)) or 'none'}") + return 0 + + api_key = os.environ.get("LINEAR_API_KEY") + if not api_key: + print("LINEAR_API_KEY is not set", file=sys.stderr) + return 2 + + summary = import_backlog(LinearClient(api_key), text) + print(f"Created {len(summary['created'])} issue(s); " + f"skipped {len(summary['skipped'])} already present.") + if summary["labels_created"]: + print(f" labels created: {summary['labels_created']}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/tests/test_import_org_backlog.py b/scripts/tests/test_import_org_backlog.py new file mode 100644 index 0000000..783e0a6 --- /dev/null +++ b/scripts/tests/test_import_org_backlog.py @@ -0,0 +1,163 @@ +"""Tests for import_org_backlog. + +The importer splits into a pure org parser + mapper (no I/O) and a thin Linear +GraphQL client. The parser pulls top-level TODO/DOING tasks out of the "Pearl +Open Work" section; the mapper turns each into an issueCreate input (priority, +state, labels); the client creates missing labels and the issues, skipping a +task whose title already exists so a re-run is safe. +""" + +import os +import sys + +import pytest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import import_org_backlog as imp + + +SAMPLE = """\ +* Pearl Priority Scheme +- =[#A]= urgent (not a task) + +* Pearl Open Work + +** DOING [#A] Personal Account Dogfooding +Meta task, skipped. +*** TODO a child +child body + +** TODO [#B] Fix the thing :bug:discuss: +:PROPERTIES: +:LAST_REVIEWED: 2026-06-06 +:END: +The body explaining the bug. +More body. +*** 2026-06-06 dated log +log entry, excluded from the description + +** TODO [#C] A docs task :docs:quick: +Body here. + +** TODO [#B] Manual testing and validation :test: +checklist umbrella, skipped +*** TODO a verify +verify body + +** TODO [#D] Someday feature :feature: +Someday body. + +* Pearl Resolved +** DONE [#A] old finished thing +done body, excluded (resolved section) +""" + + +# --- pure: parser ------------------------------------------------------------ + +def test_parse_pulls_open_work_tasks_skipping_meta_and_resolved(): + tasks = imp.parse_open_work_tasks(SAMPLE) + assert [t["title"] for t in tasks] == [ + "Fix the thing", "A docs task", "Someday feature"] + + +def test_parse_captures_priority_and_tags(): + tasks = {t["title"]: t for t in imp.parse_open_work_tasks(SAMPLE)} + assert tasks["Fix the thing"]["priority"] == "B" + assert tasks["Fix the thing"]["tags"] == ["bug", "discuss"] + assert tasks["Someday feature"]["priority"] == "D" + assert tasks["A docs task"]["tags"] == ["docs", "quick"] + + +def test_parse_body_excludes_drawer_and_subtrees(): + tasks = {t["title"]: t for t in imp.parse_open_work_tasks(SAMPLE)} + body = tasks["Fix the thing"]["body"] + assert body == "The body explaining the bug.\nMore body." + assert "LAST_REVIEWED" not in body # drawer dropped + assert "dated log" not in body # sub-entry dropped + + +def test_parse_handles_a_colon_laden_title(): + text = ("* Pearl Open Work\n\n" + "** TODO [#C] Reconsider how =#+title:= displays :bug:quick:\nBody.\n") + t = imp.parse_open_work_tasks(text)[0] + assert t["title"] == "Reconsider how =#+title:= displays" + assert t["tags"] == ["bug", "quick"] + + +# --- pure: label set + mapping ----------------------------------------------- + +def test_label_names_needed_drops_the_four_tags(): + tasks = imp.parse_open_work_tasks(SAMPLE) + # bug/docs/quick/feature survive (capitalized); discuss is dropped + assert imp.label_names_needed(tasks) == {"Bug", "Docs", "Quick", "Feature"} + + +def test_task_to_issue_maps_priority_state_and_labels(): + label_ids = {"bug": "L-bug", "feature": "L-feat", "docs": "L-docs", "quick": "L-quick"} + states = {"Backlog": "S-back", "Icebox": "S-ice"} + by = {t["title"]: t for t in imp.parse_open_work_tasks(SAMPLE)} + + bug = imp.task_to_issue(by["Fix the thing"], states, label_ids) + assert bug["priority"] == 2 # [#B] + assert bug["stateId"] == "S-back" + assert bug["labelIds"] == ["L-bug"] # discuss dropped + assert "explaining the bug" in bug["description"] + + someday = imp.task_to_issue(by["Someday feature"], states, label_ids) + assert someday["priority"] == 4 # [#D] + assert someday["stateId"] == "S-ice" # [#D] -> Icebox + + +# --- boundary: a fake GraphQL transport -------------------------------------- + +class FakeTransport: + def __init__(self, responses): + self.responses = responses + self.calls = [] + + def __call__(self, query, variables): + op = next(name for name in self.responses if name in query) + self.calls.append((op, variables)) + resp = self.responses[op] + return resp(variables) if callable(resp) else resp + + def ops(self): + return [op for op, _ in self.calls] + + +def test_import_creates_missing_labels_and_new_issues_only(): + created_labels = {"i": 0} + + def make_label(_v): + created_labels["i"] += 1 + return {"data": {"issueLabelCreate": + {"success": True, "issueLabel": {"id": f"NL{created_labels['i']}"}}}} + + fake = FakeTransport({ + "Teams": {"data": {"teams": {"nodes": [ + {"id": "t1", + "labels": {"nodes": [{"id": "L-bug", "name": "Bug"}, + {"id": "L-feat", "name": "Feature"}]}, + "states": {"nodes": [{"id": "S-back", "name": "Backlog"}, + {"id": "S-ice", "name": "Icebox"}]}}]}}}, + "Projects": {"data": {"projects": {"nodes": [{"id": "p1", "name": "Pearl"}]}}}, + "ProjectIssues": {"data": {"project": {"issues": {"nodes": [{"title": "A docs task"}]}}}}, + "LabelCreate": make_label, + "IssueCreate": {"data": {"issueCreate": {"success": True, "issue": {"id": "i1"}}}}, + }) + summary = imp.import_backlog(imp.LinearClient("key", transport=fake), SAMPLE) + ops = fake.ops() + # Bug + Feature already exist; only Docs + Quick get created (discuss dropped) + assert ops.count("LabelCreate") == 2 + # "A docs task" already exists -> skipped; the other two created + assert ops.count("IssueCreate") == 2 + assert set(summary["created"]) == {"Fix the thing", "Someday feature"} + assert summary["skipped"] == ["A docs task"] + + +def test_client_raises_on_graphql_errors(): + fake = FakeTransport({"Teams": {"errors": [{"message": "nope"}]}}) + with pytest.raises(imp.LinearError, match="nope"): + imp.LinearClient("key", transport=fake).execute("query Teams { teams { nodes { id } } }", {}) |