feat(scripts): import an org backlog into the Pearl workspace

I had 44 open tasks in the org backlog and no way to get them into Linear short of typing each one in by hand. This adds a script that reads the level-2 TODO/DOING headings under the "Pearl Open Work" section and creates one issue per task.

The priority cookie maps to Linear priority, a [#D] "someday" task lands in Icebox and the rest in Backlog, tags become labels (missing ones created), and the heading body becomes the description. The discuss/next/cleanup/pearl workflow tags are dropped rather than turned into labels. It skips a task whose title already exists, so a re-run only fills in what's missing, and the two umbrella headings plus the Resolved section are left out.

Same shape as the seed script: a pure parser and mapper with no I/O, behind a thin GraphQL client that's the only network boundary, so a fake transport routing on the operation name covers the import end to end. 8 tests across parse, mapping, and the import flow.

1 files changed, 301 insertions, 0 deletions

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())