path: root/scripts/import_org_backlog.py

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