#!/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())