#!/usr/bin/env python3 """Coordinate account data export. Downloads everything reachable through the Coordinate REST API into a folder of JSON files -- one file per project, with attachments embedded as base64. Requires nothing but a standard Python 3.8+ install. No pip, no virtualenv. python3 coordinate_export.py --api-key secret_xxxxxxxx --out ./my-export The export can be interrupted at any time; re-running the same command resumes where it left off. `export_status.json` in the output folder records whether the export actually completed, and lists anything that failed. Run with --help for all options. """ import argparse import base64 import errno import getpass import hashlib import http.client import json import os import re import shutil import signal import socket import ssl import sys import time import traceback import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone SCRIPT_VERSION = "1.0" EXPORT_FORMAT_VERSION = 1 DEFAULT_BASE_URL = "https://app.coordinatehq.com" DEFAULT_INLINE_MAX_BYTES = 10 * 1024 * 1024 # 10 MB DOWNLOAD_CHUNK = 64 * 1024 HTTP_TIMEOUT = 120 HTTP_RETRIES = 3 STATUS_FILENAME = "export_status.json" EXIT_OK = 0 EXIT_COMPLETED_WITH_ERRORS = 1 EXIT_INTERRUPTED = 2 EXIT_FAILED = 3 # Recorded verbatim into export_status.json so whoever reads the export knows what is not # in it. These are limits of the public API, not of this script. KNOWN_GAPS = [ "Account/vendor settings are not exposed by the API: company name, branding, tag list, " "project status list, project roles, custom field definitions and form definitions are " "all absent. Field *values* are exported, definitions are not.", "The internal user list is not exposed by the API. Team members appear only where they " "are referenced (project manager, comment author, task assignee).", "The activity feed is not available through the API.", "Project templates and playbooks are not available through the API, and template " "projects are excluded from the project list.", "Direct messages are not available through the API.", "Recurring task definitions and their schedules are not returned by the API.", "Task dependencies, goal-to-task links, call-to-action type and e-signature capture are " "not returned by the API.", "Group colour, archived state and auto-assign configuration are not returned by the API. " "Whether a group is internal is inferred from its tasks, so an empty group reports null.", "Answers to file-type form fields are omitted by the API; other form answers are included.", "Per-project notification preferences, tab order and reminder settings are not exposed.", "Sent email history, iCal feeds, webhook configuration, and logo/avatar images are not " "exposed by the API.", "Draft and deleted items are excluded, as are any records with no last-modified date.", "Images pasted inline into comments cannot be downloaded through the API, so comment " "HTML may reference images that are not present in this export.", "Archived and inactive projects ARE included; check the project_active field.", ] # -------------------------------------------------------------------------------------- # small helpers # -------------------------------------------------------------------------------------- def utcnow_iso(): return datetime.now(timezone.utc).isoformat() def fmt_bytes(n): if n is None: return "?" step = 1024.0 for unit in ("B", "KB", "MB", "GB", "TB"): if abs(n) < step or unit == "TB": if unit == "B": return "%d B" % n return "%.1f %s" % (n, unit) n /= step return "%.1f TB" % n def fmt_duration(seconds): if seconds is None: return "--:--:--" seconds = int(max(0, seconds)) return "%02d:%02d:%02d" % (seconds // 3600, (seconds % 3600) // 60, seconds % 60) _UNSAFE = re.compile(r"[^A-Za-z0-9 ._-]+") _WINDOWS_RESERVED = { "CON", "PRN", "AUX", "NUL", "COM1", "COM2", "COM3", "COM4", "COM5", "COM6", "COM7", "COM8", "COM9", "LPT1", "LPT2", "LPT3", "LPT4", "LPT5", "LPT6", "LPT7", "LPT8", "LPT9", } def sanitize_filename(name, fallback="untitled", max_length=80): """Make a string safe to use as a single path component on macOS, Linux and Windows.""" if not name: return fallback cleaned = _UNSAFE.sub("_", str(name)) cleaned = re.sub(r"[\s_]+", "_", cleaned).strip("._ ") cleaned = cleaned[:max_length].strip("._ ") if not cleaned: return fallback if cleaned.upper() in _WINDOWS_RESERVED: cleaned = cleaned + "_" return cleaned def write_json_atomic(path, obj): """Write JSON via a temp file + rename. Returns the sha256 of the bytes written. The rename is what makes a hard kill survivable: readers either see the previous complete file or the new complete file, never a half-written one. """ data = json.dumps(obj, indent=2, ensure_ascii=False, sort_keys=False).encode("utf-8") tmp = path + ".tmp" ensure_dir(os.path.dirname(path)) with open(tmp, "wb") as fh: fh.write(data) fh.flush() os.fsync(fh.fileno()) os.replace(tmp, path) return hashlib.sha256(data).hexdigest() def sha256_of_file(path): h = hashlib.sha256() try: with open(path, "rb") as fh: while True: chunk = fh.read(DOWNLOAD_CHUNK) if not chunk: break h.update(chunk) except OSError: return None return h.hexdigest() def ensure_dir(path): if not path: return try: os.makedirs(path) except OSError as exc: if exc.errno != errno.EEXIST: raise def short_exception(exc): """One-line description of any exception, for logs and the status file.""" return "%s: %s" % (exc.__class__.__name__, exc) class FatalError(Exception): """Something that means we cannot start or cannot safely continue.""" class ApiError(Exception): def __init__(self, message, status=None): Exception.__init__(self, message) self.status = status # -------------------------------------------------------------------------------------- # HTTP # -------------------------------------------------------------------------------------- class _NoRedirect(urllib.request.HTTPRedirectHandler): """Stop urllib from following redirects so we never forward the API key to S3.""" def redirect_request(self, req, fp, code, msg, headers, newurl): return None RETRYABLE_EXCEPTIONS = ( urllib.error.URLError, http.client.HTTPException, socket.timeout, socket.error, ssl.SSLError, ) class Client(object): def __init__(self, base_url, api_key, logger, timeout=HTTP_TIMEOUT, retries=HTTP_RETRIES, verbose=False): self.base_url = base_url.rstrip("/") self.api_key = api_key self.timeout = timeout self.retries = retries self.log = logger self.verbose = verbose self.request_count = 0 # Two openers: one that refuses redirects (used for anything carrying our API key) # and a bare one for following the presigned storage URL we get handed back. self._api_opener = urllib.request.build_opener(_NoRedirect) self._bare_opener = urllib.request.build_opener() # -- low level ---------------------------------------------------------------- def _open(self, url, opener, headers, label): """Open a URL with retries. Returns the response object (caller must close).""" last_error = None for attempt in range(1, self.retries + 1): req = urllib.request.Request(url, headers=headers or {}) try: self.request_count += 1 if self.verbose: self.log("GET %s (attempt %d)" % (url, attempt)) return opener.open(req, timeout=self.timeout) except urllib.error.HTTPError as exc: # Redirects reach us as HTTPError because of _NoRedirect; hand them back. if exc.code in (301, 302, 303, 307, 308): return exc if exc.code >= 500 and attempt < self.retries: last_error = exc self._sleep_backoff(attempt, label, "HTTP %d" % exc.code) continue raise ApiError("%s -> HTTP %d %s" % (label, exc.code, _read_error_body(exc)), status=exc.code) except RETRYABLE_EXCEPTIONS as exc: last_error = exc if attempt < self.retries: self._sleep_backoff(attempt, label, str(exc)) continue raise ApiError("%s -> %s (after %d attempts)" % (label, exc, self.retries)) raise ApiError("%s -> %s" % (label, last_error)) def _sleep_backoff(self, attempt, label, reason): delay = min(30, 3 ** (attempt - 1)) self.log("retrying %s in %ds after %s" % (label, delay, reason)) time.sleep(delay) # -- json ------------------------------------------------------------------------- def get_json(self, path, params=None): url = self.base_url + path if params: url = url + "?" + urllib.parse.urlencode(params) resp = self._open(url, self._api_opener, {"X-API-Key": self.api_key}, path) try: body = resp.read() content_type = (resp.headers.get("Content-Type") or "").lower() finally: resp.close() # This API returns plain text for several error paths, so never assume JSON. if "json" not in content_type: snippet = body[:300].decode("utf-8", "replace").strip() raise ApiError("%s -> expected JSON, got %r (%s)" % (path, content_type, snippet)) try: return json.loads(body.decode("utf-8")) except ValueError as exc: raise ApiError("%s -> malformed JSON: %s" % (path, exc)) # -- files ------------------------------------------------------------------------ def open_download(self, url): """Resolve a file download URL to an open stream. The API answers with a 302 to a short-lived presigned storage URL. We follow it by hand with a clean opener so the API key is never sent to the storage host. """ if url.startswith("/"): url = self.base_url + url resp = self._open(url, self._api_opener, {"X-API-Key": self.api_key}, "download") code = getattr(resp, "code", None) or getattr(resp, "status", None) if code in (301, 302, 303, 307, 308): location = resp.headers.get("Location") resp.close() if not location: raise ApiError("download -> %s with no Location header" % code) resp = self._open(location, self._bare_opener, {}, "storage download") code = getattr(resp, "code", None) or getattr(resp, "status", None) if code is not None and code >= 400: body = _read_error_body(resp) resp.close() raise ApiError("download -> HTTP %d %s" % (code, body), status=code) return resp def _read_error_body(resp): try: return resp.read(300).decode("utf-8", "replace").strip().replace("\n", " ") except Exception: return "" # -------------------------------------------------------------------------------------- # progress display # -------------------------------------------------------------------------------------- class Progress(object): """Single-line live progress on a terminal, plain appended lines everywhere else. Everything goes to stderr so stdout stays clean and a redirected log stays readable. """ def __init__(self, quiet=False, stream=None): self.stream = stream or sys.stderr self.quiet = quiet try: self.tty = bool(self.stream.isatty()) and not quiet except Exception: self.tty = False self.total = 0 self.done = 0 self.project_name = "" self.detail_text = "" self.byte_text = "" self._durations = [] self._project_started = None self._last_draw = 0.0 self._dirty = False # -- lifecycle --------------------------------------------------------------------- def banner(self, text): if not self.quiet: self.stream.write(text + "\n") self.stream.flush() def start(self, total): self.total = total self.done = 0 def start_project(self, name): self.project_name = name self.detail_text = "" self.byte_text = "" self._project_started = time.time() self._draw(force=True) def detail(self, text): self.detail_text = text self.byte_text = "" self._draw() def bytes_progress(self, done, total, name): if total: self.byte_text = "%s/%s %s" % (fmt_bytes(done), fmt_bytes(total), name) else: self.byte_text = "%s %s" % (fmt_bytes(done), name) self._draw() def finish_project(self, summary_line): if self._project_started is not None: self._durations.append(time.time() - self._project_started) del self._durations[:-10] self._project_started = None self.done += 1 self.byte_text = "" if self.quiet: return if self.tty: self._clear() self.stream.write(summary_line + "\n") self.stream.flush() self._draw(force=True) else: self.stream.write(summary_line + "\n") self.stream.flush() def note(self, text): """A message that must survive above the live line.""" if self.quiet: return if self.tty: self._clear() self.stream.write(text + "\n") self.stream.flush() if self.tty: self._draw(force=True) def close(self): if self.tty: self._clear() # -- drawing ----------------------------------------------------------------------- def _eta(self): if not self._durations or not self.total: return None remaining = self.total - self.done if remaining <= 0: return 0 mean = sum(self._durations) / float(len(self._durations)) return mean * remaining def _draw(self, force=False): if not self.tty: return now = time.time() if not force and (now - self._last_draw) < 0.1: return self._last_draw = now pct = int(100 * self.done / self.total) if self.total else 0 current = min(self.done + 1, self.total) if self.total else 0 parts = ["[%*d/%d] %3d%%" % (len(str(self.total)), current, self.total, pct)] if self.project_name: parts.append(self.project_name) if self.byte_text: parts.append("v " + self.byte_text) elif self.detail_text: parts.append(self.detail_text) eta = self._eta() if eta: parts.append("eta " + fmt_duration(eta)) line = " ".join(parts) width = shutil.get_terminal_size((100, 24)).columns if len(line) > width - 1: line = line[:max(0, width - 2)] + "…" self.stream.write("\r\x1b[K" + line) self.stream.flush() self._dirty = True def _clear(self): if self._dirty: self.stream.write("\r\x1b[K") self.stream.flush() self._dirty = False class Logger(object): """Plain timestamped log file, always written regardless of terminal state.""" def __init__(self, path): self.path = path self._fh = None if path: ensure_dir(os.path.dirname(path)) self._fh = open(path, "a", encoding="utf-8") def __call__(self, message): if self._fh is None: return self._fh.write("%s %s\n" % (utcnow_iso(), message)) self._fh.flush() def close(self): if self._fh is not None: self._fh.close() self._fh = None # -------------------------------------------------------------------------------------- # status ledger # -------------------------------------------------------------------------------------- class Status(object): """The central record: resume ledger during the run, completion record afterwards.""" def __init__(self, path, data): self.path = path self.data = data @classmethod def create(cls, path, base_url, vendor_id, vendor_name): return cls(path, { "export_format_version": EXPORT_FORMAT_VERSION, "script_version": SCRIPT_VERSION, "status": "in_progress", "base_url": base_url, "vendor_id": vendor_id, "vendor_name": vendor_name, "started_at_utc": utcnow_iso(), "updated_at_utc": utcnow_iso(), "finished_at_utc": None, "runs": [], "totals": {}, "projects": {}, "errors": [], "known_gaps": list(KNOWN_GAPS), }) @classmethod def load(cls, path): with open(path, "r", encoding="utf-8") as fh: return cls(path, json.load(fh)) def check_compatible(self, base_url, vendor_id): """Refuse to mix two different accounts, or two different formats, in one folder.""" found = self.data.get("export_format_version") if found != EXPORT_FORMAT_VERSION: raise FatalError( "%s was written by a different version of this script (format %s, this script " "writes format %s). Use a fresh --out directory." % (self.path, found, EXPORT_FORMAT_VERSION)) if self.data.get("base_url") != base_url: raise FatalError( "%s belongs to an export from %s but you passed --base-url %s. Use a fresh " "--out directory." % (self.path, self.data.get("base_url"), base_url)) if vendor_id and self.data.get("vendor_id") and self.data["vendor_id"] != vendor_id: raise FatalError( "%s belongs to account %s but this API key is for account %s. Use a fresh " "--out directory." % (self.path, self.data["vendor_id"], vendor_id)) def project(self, project_id): return self.data["projects"].get(project_id) def set_project(self, project_id, record): self.data["projects"][project_id] = record def add_error(self, scope, detail, project_id=None): self.data["errors"].append({ "when_utc": utcnow_iso(), "scope": scope, "project_id": project_id, "detail": detail, }) def recompute_totals(self, current_project_ids): """Totals are always relative to what the account holds right now. A project the ledger knows about but the account no longer has is counted as `disappeared` and contributes nothing else, so it can never make an incomplete export look complete. """ counters = { "projects": len(current_project_ids), "completed": 0, "failed": 0, "pending": 0, "disappeared": 0, "tasks": 0, "subtasks": 0, "comments": 0, "files_downloaded": 0, "files_failed": 0, "bytes_downloaded": 0, } for project_id in current_project_ids: record = self.data["projects"].get(project_id) state = (record or {}).get("status") if state == "completed": counters["completed"] += 1 elif state == "failed": counters["failed"] += 1 else: counters["pending"] += 1 counts = (record or {}).get("counts") or {} for key, target in (("tasks", "tasks"), ("subtasks", "subtasks"), ("comments", "comments"), ("files_ok", "files_downloaded"), ("files_failed", "files_failed"), ("bytes", "bytes_downloaded")): counters[target] += counts.get(key, 0) or 0 for project_id, record in self.data["projects"].items(): if project_id not in current_project_ids: counters["disappeared"] += 1 self.data["totals"] = counters return counters def save(self): self.data["updated_at_utc"] = utcnow_iso() write_json_atomic(self.path, self.data) # -------------------------------------------------------------------------------------- # attachments # -------------------------------------------------------------------------------------- class FileFetcher(object): def __init__(self, client, files_root, inline_max_bytes, progress, logger, skip_files=False): self.client = client self.files_root = files_root self.inline_max_bytes = inline_max_bytes self.progress = progress self.log = logger self.skip_files = skip_files self.reset_counters() def reset_counters(self): self.ok = 0 self.failed = 0 self.bytes = 0 self.errors = [] def hydrate(self, file_entries, project_id): """Return copies of the API's file entries with their bytes attached.""" out = [] for entry in (file_entries or []): out.append(self._fetch_one(entry, project_id)) return out def _fetch_one(self, entry, project_id): record = { "file_uid": entry.get("file_uid"), "file_name": entry.get("file_name"), "file_size": entry.get("file_size"), "file_content_type": entry.get("file_content_type"), "file_dt": entry.get("file_dt"), "sha256": None, "content_base64": None, "content_path": None, "download_error": None, } url = entry.get("download_url") display_name = entry.get("file_name") or entry.get("file_uid") or "file" if self.skip_files: record["download_error"] = "skipped (--skip-files)" return record if not url: record["download_error"] = "no download_url returned by the API" self.failed += 1 self.errors.append("%s: no download_url" % display_name) return record expected = entry.get("file_size") try: data, path, size, digest = self._stream(url, project_id, entry, expected, display_name) except Exception as exc: # Includes OSError from the spill path (disk full, permissions) as well as # anything the API or network threw. One bad file never stops the export. record["download_error"] = short_exception(exc) self.failed += 1 self.errors.append("%s: %s" % (display_name, short_exception(exc))) self.log("file download failed %s (%s): %s" % (display_name, project_id, short_exception(exc))) return record record["sha256"] = digest record["file_size"] = size if path is not None: record["content_path"] = os.path.relpath(path, os.path.dirname(self.files_root)) else: record["content_base64"] = base64.b64encode(data).decode("ascii") self.ok += 1 self.bytes += size return record def _spill_path(self, project_id, entry): directory = os.path.join(self.files_root, sanitize_filename(project_id, "project")) ensure_dir(directory) name = "%s__%s" % (sanitize_filename(entry.get("file_uid"), "file", 40), sanitize_filename(entry.get("file_name"), "attachment")) return os.path.join(directory, name) def _stream(self, url, project_id, entry, expected_size, display_name): """Stream a download, buffering in memory until it outgrows the inline limit. Returns (bytes_or_None, spill_path_or_None, size, sha256). """ resp = self.client.open_download(url) try: declared = resp.headers.get("Content-Length") total = int(declared) if declared and declared.isdigit() else expected_size digest = hashlib.sha256() buffered = bytearray() handle = None spill_path = None spill_tmp = None size = 0 try: while True: chunk = resp.read(DOWNLOAD_CHUNK) if not chunk: break digest.update(chunk) size += len(chunk) if handle is None: buffered.extend(chunk) if len(buffered) > self.inline_max_bytes: # Outgrew the inline budget: flush what we have to disk and # keep streaming there. spill_path = self._spill_path(project_id, entry) spill_tmp = spill_path + ".tmp" handle = open(spill_tmp, "wb") handle.write(buffered) buffered = bytearray() else: handle.write(chunk) self.progress.bytes_progress(size, total, display_name) except RETRYABLE_EXCEPTIONS as exc: if handle is not None: handle.close() handle = None if spill_tmp and os.path.exists(spill_tmp): os.remove(spill_tmp) # never leave a partial .tmp behind raise ApiError("download of %s interrupted: %s" % (display_name, exc)) finally: if handle is not None: handle.close() finally: resp.close() if spill_path is not None: os.replace(spill_tmp, spill_path) return None, spill_path, size, digest.hexdigest() return bytes(buffered), None, size, digest.hexdigest() # -------------------------------------------------------------------------------------- # project assembly # -------------------------------------------------------------------------------------- def bucket_comments(comments): buckets = {} for comment in comments: buckets.setdefault(comment.get("target_entity_id"), []).append(comment) for entries in buckets.values(): entries.sort(key=lambda c: c.get("discussion_timestamp_dt") or "") return buckets def decorate_comments(comments, files, project_id): """Add the derived `private` alias and attach downloaded bytes.""" out = [] for comment in comments: enriched = dict(comment) enriched["private"] = bool(comment.get("discussion_entry_internal")) enriched["files"] = files.hydrate(comment.get("files"), project_id) out.append(enriched) return out def sort_key_int(value, default=10 ** 9): return value if isinstance(value, int) else default def export_project(client, files, progress, project_id, org_map): """Fetch and assemble one project. Returns (document, counts, warnings, errors). Only an unreadable project record aborts the project; every other section degrades on its own, records why, and the rest of the project still gets exported. """ warnings = [] errors = [] def guard(label, fn, default): """Run a piece of assembly; on any failure record it and carry on.""" try: return fn() except Exception as exc: errors.append("%s failed (%s)" % (label, short_exception(exc))) return default def section(label, path, params=None): """Fetch a list endpoint. A failure costs that section, not the project.""" progress.detail(label) try: value = client.get_json(path, params) except Exception as exc: errors.append("%s could not be fetched (%s)" % (label, short_exception(exc))) return [] if not isinstance(value, list): errors.append("%s returned %s rather than a list; skipped." % (label, type(value).__name__)) return [] rows = [v for v in value if isinstance(v, dict)] if len(rows) != len(value): # Never drop records silently - that would make a partial export look complete. errors.append("%s: skipped %d entry/entries that were not objects." % (label, len(value) - len(rows))) return rows # The one genuinely fatal case: without the project record there is nothing to write. progress.detail("project") project = client.get_json("/api/v1/projects/%s" % project_id) if not isinstance(project, dict): raise ApiError("project record was %s, not an object" % type(project).__name__) groups = section("groups", "/api/v1/projects/%s/group" % project_id) tasks = section("tasks", "/api/v1/projects/%s/task" % project_id, {"include_subtasks": "true"}) goals = section("goals", "/api/v1/projects/%s/goal" % project_id) collaborators = section("collaborators", "/api/v1/projects/%s/stakeholder" % project_id) progress_reports = section("progress reports", "/api/v1/projects/%s/progress_report" % project_id) # Deliberately no last_modified_dt: that filter applies to the parent task/goal, so it # would silently drop new comments on old tasks. comments = section("comments", "/api/v1/projects/%s/comments" % project_id, {"include_subtasks": "true"}) progress.detail("pages") pages = guard("pages", lambda: fetch_pages(client, project_id, warnings), []) comment_buckets = guard("bucketing comments", lambda: bucket_comments(comments), {}) # -- tasks: attach comments and files, then nest subtasks under checklist items ------ progress.detail("attachments") tasks_by_id = {} for task in tasks: task_id = task.get("task_id") label = task.get("task_title") or task_id task["discussion"] = guard( "comments on task %r" % label, lambda: decorate_comments(comment_buckets.pop(task_id, []), files, project_id), []) task["files"] = guard( "attachments on task %r" % label, lambda: files.hydrate(task.get("files"), project_id), []) if task_id: tasks_by_id[task_id] = task subtask_ids = set(tid for tid, t in tasks_by_id.items() if t.get("task_parent_task_id")) attached = set() def nest_checklists(task): for checklist in (task.get("task_checklists") or []): if not isinstance(checklist, dict): continue items = [i for i in (checklist.get("items") or []) if isinstance(i, dict)] items.sort(key=lambda i: sort_key_int(i.get("order"))) checklist["items"] = items for item in items: subtask_id = item.get("subtask_task_id") if not subtask_id: item["subtask"] = None continue subtask = tasks_by_id.get(subtask_id) if subtask is None: item["subtask"] = None warnings.append( "Checklist item %r on task %r references subtask %s, which the API " "did not return (it may be deleted or recurring)." % (item.get("text"), task.get("task_title"), subtask_id)) continue item["subtask"] = subtask attached.add(subtask_id) for task in tasks: guard("checklists on task %r" % (task.get("task_title") or task.get("task_id")), lambda t=task: nest_checklists(t), None) orphaned_subtasks = [] for subtask_id in sorted(subtask_ids - attached): subtask = tasks_by_id[subtask_id] orphaned_subtasks.append(subtask) warnings.append( "Subtask %r (%s) has no checklist item pointing at it; it is listed under " "tasks_ungrouped so it is not lost." % (subtask.get("task_title"), subtask_id)) top_level = [t for t in tasks if not t.get("task_parent_task_id")] guard("sorting tasks", lambda: top_level.sort(key=lambda t: (sort_key_int(t.get("task_sort_order")), t.get("task_title") or "")), None) # -- groups -------------------------------------------------------------------------- groups = guard("sorting groups", lambda: sorted(groups, key=lambda g: (sort_key_int(g.get("group_sort_order")), g.get("group_title") or "")), groups) group_map = {} group_docs = [] for group in groups: doc = dict(group) doc["group_internal"] = None doc["tasks"] = [] group_map[group.get("group_id")] = doc group_docs.append(doc) ungrouped = list(orphaned_subtasks) for task in top_level: group_id = task.get("group_id") if group_id and group_id in group_map: group_map[group_id]["tasks"].append(task) else: if group_id: warnings.append( "Task %r references group %s which was not returned; it is listed as " "ungrouped." % (task.get("task_title"), group_id)) ungrouped.append(task) for doc in group_docs: if doc["tasks"]: # The API never exposes a group's private flag directly, but it does report it # per task as task_internal. doc["group_internal"] = bool(doc["tasks"][0].get("task_internal")) else: warnings.append( "Group %r has no tasks, so whether it is internal-only could not be " "determined (group_internal is null)." % doc.get("group_title")) # -- goals --------------------------------------------------------------------------- goal_docs = [] for goal in goals: doc = dict(goal) label = goal.get("goal_title") or goal.get("goal_id") doc["files"] = guard("attachments on goal %r" % label, lambda g=goal: files.hydrate(g.get("files"), project_id), []) doc["discussion"] = guard( "comments on goal %r" % label, lambda g=goal: decorate_comments(comment_buckets.pop(g.get("goal_id"), []), files, project_id), []) goal_docs.append(doc) # -- project level comments and leftovers --------------------------------------------- project_discussion = guard( "project comments", lambda: decorate_comments(comment_buckets.pop(project_id, []), files, project_id), []) orphan_comments = [] for entity_id, entries in sorted(comment_buckets.items(), key=lambda kv: str(kv[0])): orphan_comments.extend(guard( "orphaned comments on %s" % entity_id, lambda e=entries: decorate_comments(e, files, project_id), [])) warnings.append( "%d comment(s) target entity %s which is not a task, goal or the project itself; " "they are listed under discussion_orphaned." % (len(entries), entity_id)) project_doc = dict(project) project_doc["files"] = guard("project attachments", lambda: files.hydrate(project.get("files"), project_id), []) # `customers` on the project payload is a list of {customer_id, customer_name} references # to linked organizations. Swap in the full records we already fetched where we can. linked_orgs = [] for reference in (project.get("customers") or []): full = org_map.get(reference.get("customer_id")) linked_orgs.append(full if full else { "organization_id": reference.get("customer_id"), "organization_name": reference.get("customer_name"), }) document = { "export_format_version": EXPORT_FORMAT_VERSION, "exported_at_utc": utcnow_iso(), "source": { "base_url": client.base_url, "vendor_id": project.get("vendor_id"), "project_id": project_id, }, "project": project_doc, "pages": pages, "organizations": linked_orgs, "collaborators": collaborators, "groups": group_docs, "tasks_ungrouped": ungrouped, "goals": goal_docs, "progress_reports": progress_reports, "discussion": project_discussion, "discussion_orphaned": orphan_comments, "warnings": warnings, # Anything that went wrong while building this file. `partial` is the quick check. "errors": errors + ["attachment - " + e for e in files.errors], "partial": bool(errors or files.errors), } counts = { "tasks": len(top_level), "subtasks": len(subtask_ids), "comments": len(comments), "groups": len(group_docs), "goals": len(goal_docs), "files_ok": files.ok, "files_failed": files.failed, "bytes": files.bytes, } return document, counts, warnings, errors def fetch_pages(client, project_id, warnings): try: names = client.get_json("/api/v1/projects/%s/pages" % project_id) except ApiError as exc: warnings.append("Could not list project pages: %s" % exc) return [] pages = [] for name in names or []: try: quoted = urllib.parse.quote(str(name), safe="") pages.append(client.get_json("/api/v1/projects/%s/pages/%s" % (project_id, quoted))) except ApiError as exc: warnings.append("Could not fetch project page %r: %s" % (name, exc)) return pages # -------------------------------------------------------------------------------------- # account level # -------------------------------------------------------------------------------------- def export_organizations(client, logger): """Organizations, each with their collaborators and linked projects. A failure on any one organization costs only that organization's sub-records. """ orgs = client.get_json("/api/v1/organizations") if not isinstance(orgs, list): raise ApiError("organizations returned %s, not a list" % type(orgs).__name__) out = [] for org in orgs: if not isinstance(org, dict): continue org_id = org.get("organization_id") doc = dict(org) try: raw = client.get_json("/api/v1/organizations/%s/stakeholders" % org_id) # This endpoint uniquely returns an array of single-element arrays. flattened = [] for element in raw or []: if isinstance(element, list): flattened.extend(element) else: flattened.append(element) doc["stakeholders"] = flattened except Exception as exc: doc["stakeholders"] = [] doc["stakeholders_error"] = short_exception(exc) logger("organization stakeholders failed for %s: %s" % (org_id, short_exception(exc))) try: doc["projects"] = client.get_json("/api/v1/organizations/%s/projects" % org_id) except Exception as exc: doc["projects"] = [] doc["projects_error"] = short_exception(exc) logger("organization projects failed for %s: %s" % (org_id, short_exception(exc))) out.append(doc) return out def build_account_doc(client, vendor_id, vendor_name, vendor_name_source, logger): doc = { "export_format_version": EXPORT_FORMAT_VERSION, "exported_at_utc": utcnow_iso(), "source": {"base_url": client.base_url, "vendor_id": vendor_id}, "vendor_name": vendor_name, "vendor_name_source": vendor_name_source, "json_storage": None, "known_gaps": list(KNOWN_GAPS), } try: doc["json_storage"] = client.get_json("/api/v1/json_storage") except Exception as exc: doc["json_storage_error"] = short_exception(exc) logger("json_storage failed: %s" % short_exception(exc)) return doc # -------------------------------------------------------------------------------------- # output directory resolution # -------------------------------------------------------------------------------------- def find_existing_export(out_root, base_url, vendor_id): """Locate a previous export for this account so --resume works without --vendor-name.""" if not os.path.isdir(out_root): return None for name in sorted(os.listdir(out_root)): candidate = os.path.join(out_root, name, STATUS_FILENAME) if not os.path.isfile(candidate): continue try: with open(candidate, "r", encoding="utf-8") as fh: data = json.load(fh) except (ValueError, OSError): continue if data.get("base_url") == base_url and vendor_id and data.get("vendor_id") == vendor_id: return os.path.join(out_root, name) return None def resolve_vendor_name(args, vendor_id): if args.vendor_name: return args.vendor_name, "command line" if sys.stdin.isatty() and not args.quiet: try: typed = input("Company name for the export folder " "(press enter to skip): ").strip() except EOFError: typed = "" if typed: return typed, "prompted" # The API exposes no vendor name, so fall back to the account id. return "vendor_%s" % (vendor_id[:8] if vendor_id else "unknown"), "fallback (API does not expose the account name)" # -------------------------------------------------------------------------------------- # main # -------------------------------------------------------------------------------------- INTERRUPT = {"requested": False} def install_signal_handler(progress): def handler(signum, frame): if INTERRUPT["requested"]: raise KeyboardInterrupt() INTERRUPT["requested"] = True progress.note("\nInterrupt received - finishing the current project, then stopping. " "Press Ctrl-C again to stop immediately.") try: signal.signal(signal.SIGINT, handler) except (ValueError, OSError): pass # not on the main thread, or unsupported platform def parse_args(argv): parser = argparse.ArgumentParser( description="Export all of your Coordinate data to a folder of JSON files.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="The export resumes automatically if interrupted - just run the same command " "again. Check export_status.json to confirm it completed.") parser.add_argument("--api-key", help="API key. Defaults to the COORDINATE_API_KEY " "environment variable, or you will be prompted.") parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help="Site address (default: %s)" % DEFAULT_BASE_URL) parser.add_argument("--out", default="./coordinate-export", help="Output directory (default: ./coordinate-export)") parser.add_argument("--vendor-name", help="Name for the export folder. The API does not " "expose your company name, so supply it here.") parser.add_argument("--inline-max-bytes", type=int, default=DEFAULT_INLINE_MAX_BYTES, help="Attachments larger than this are written to files/ instead of " "being embedded as base64 (default: %d)" % DEFAULT_INLINE_MAX_BYTES) parser.add_argument("--skip-files", action="store_true", help="Record attachment metadata but do not download the contents.") parser.add_argument("--only-project", action="append", default=[], metavar="PROJECT_ID", help="Export only this project. May be repeated.") parser.add_argument("--force", action="store_true", help="Re-export projects that previously completed.") parser.add_argument("--quiet", action="store_true", help="Suppress live progress output.") parser.add_argument("--verbose", action="store_true", help="Log every HTTP request.") parser.add_argument("--timeout", type=int, default=HTTP_TIMEOUT, help="Per-request timeout in seconds (default: %d)" % HTTP_TIMEOUT) return parser.parse_args(argv) def resolve_api_key(args): key = args.api_key or os.environ.get("COORDINATE_API_KEY") if not key and sys.stdin.isatty(): key = getpass.getpass("API key: ").strip() if not key: raise FatalError("No API key. Pass --api-key, or set COORDINATE_API_KEY.") return key def project_filename(index, project_name, project_id): return "projects/%04d__%s__%s.json" % ( index, sanitize_filename(project_name, "project"), (project_id or "")[:8]) def main(argv): args = parse_args(argv) progress = Progress(quiet=args.quiet) logger = Logger(None) status = None run_started = time.time() run_outcome = "failed" try: api_key = resolve_api_key(args) base_url = args.base_url.rstrip("/") client = Client(base_url, api_key, logger, timeout=args.timeout, verbose=args.verbose) progress.banner("Coordinate export - contacting %s" % base_url) try: client.get_json("/api/v1/test") except ApiError as exc: if exc.status in (401, 403): raise FatalError("The API key was rejected (%s). Check that you copied the " "whole key from Settings > Integrations > API Keys." % exc) raise FatalError("Could not reach the API: %s" % exc) all_projects = client.get_json("/api/v1/projects") vendor_id = None for project in all_projects: if project.get("vendor_id"): vendor_id = project["vendor_id"] break # Reuse a previous export folder for this account so resume works without having to # remember exactly what --vendor-name was typed last time. out_dir = find_existing_export(args.out, base_url, vendor_id) if out_dir: vendor_name = os.path.basename(out_dir) vendor_name_source = "existing export folder" else: vendor_name, vendor_name_source = resolve_vendor_name(args, vendor_id) out_dir = os.path.join(args.out, sanitize_filename(vendor_name, "vendor")) ensure_dir(out_dir) ensure_dir(os.path.join(out_dir, "projects")) logger.close() logger = Logger(os.path.join(out_dir, "export.log")) client.log = logger logger("=== run start: %s projects visible, out=%s" % (len(all_projects), out_dir)) status_path = os.path.join(out_dir, STATUS_FILENAME) if os.path.isfile(status_path): status = Status.load(status_path) status.check_compatible(base_url, vendor_id) status.data["status"] = "in_progress" status.data["script_version"] = SCRIPT_VERSION status.data["known_gaps"] = list(KNOWN_GAPS) if vendor_id: status.data["vendor_id"] = vendor_id else: status = Status.create(status_path, base_url, vendor_id, vendor_name) status.data["finished_at_utc"] = None status.save() install_signal_handler(progress) all_projects.sort(key=lambda p: ((p.get("project_name") or "").lower(), p.get("project_id") or "")) current_ids = set(p.get("project_id") for p in all_projects) # Anything in the ledger that the account no longer has. Recorded rather than # deleted, so a project that vanishes mid-export is visible afterwards. for project_id, record in status.data["projects"].items(): if project_id not in current_ids and record.get("status") != "disappeared": record["status"] = "disappeared" logger("project %s no longer exists in the account" % project_id) # `targets` is what this run walks; totals stay relative to the whole account so # --only-project can never make a partial export report itself as complete. targets = all_projects if args.only_project: wanted = set(args.only_project) targets = [p for p in all_projects if p.get("project_id") in wanted] for project_id in sorted(wanted - current_ids): status.add_error("project", "requested project not found", project_id) progress.note("Project %s was not found in this account." % project_id) progress.banner("Exporting %s (%s) - %d of %d project(s)" % (vendor_name, vendor_id or "unknown account", len(targets), len(all_projects))) # -- account level ------------------------------------------------------------ account_doc = build_account_doc(client, vendor_id, vendor_name, vendor_name_source, logger) write_json_atomic(os.path.join(out_dir, "account.json"), account_doc) org_map = {} try: organizations = export_organizations(client, logger) write_json_atomic(os.path.join(out_dir, "organizations.json"), organizations) for org in organizations: if org.get("organization_id"): # Drop the org's project list before embedding it in each project file; # organizations.json already holds it, and repeating it per project would # duplicate every project payload once per linked org. org_map[org["organization_id"]] = { k: v for k, v in org.items() if k not in ("projects", "projects_error")} except Exception as exc: status.add_error("organizations", short_exception(exc)) progress.note("Organizations could not be exported: %s" % short_exception(exc)) logger("organizations failed: %s" % traceback.format_exc().replace("\n", " | ")) # -- projects ------------------------------------------------------------------ files_root = os.path.join(out_dir, "files") fetcher = FileFetcher(client, files_root, args.inline_max_bytes, progress, logger, skip_files=args.skip_files) next_index = 1 + max([(r.get("index") or 0) for r in status.data["projects"].values()] or [0]) progress.start(len(targets)) for project in targets: if INTERRUPT["requested"]: break project_id = project.get("project_id") project_name = project.get("project_name") or "(untitled)" record = status.project(project_id) or {} if not args.force and should_skip(record, out_dir): progress.finish_project("%-9s %s (already exported)" % ("[skip]", project_name)) continue index = record.get("index") or next_index if not record.get("index"): next_index += 1 relative = record.get("file") or project_filename(index, project_name, project_id) progress.start_project(project_name) fetcher.reset_counters() started = time.time() # Everything below is deliberately catch-all: a single unreadable project must # never take down a run that still has hundreds to go. try: document, counts, warnings, project_errors = export_project( client, fetcher, progress, project_id, org_map) digest = write_json_atomic(os.path.join(out_dir, relative), document) except Exception as exc: detail = short_exception(exc) status.set_project(project_id, { "project_name": project_name, "status": "failed", "index": index, "file": relative, "sha256": None, "attempts": (record.get("attempts") or 0) + 1, "last_error": detail, "completed_at_utc": None, "counts": record.get("counts") or {}, }) status.add_error("project", detail, project_id) safe_save(status, logger) progress.finish_project("%-9s %s - %s" % ("[FAIL]", project_name, detail)) logger("project %s failed: %s" % (project_id, traceback.format_exc().replace("\n", " | "))) continue for message in project_errors: status.add_error("project-section", message, project_id) for message in fetcher.errors: status.add_error("file", message, project_id) status.set_project(project_id, { "project_name": project_name, "status": "completed", "index": index, "file": relative, "sha256": digest, "attempts": (record.get("attempts") or 0) + 1, "last_error": None, "completed_at_utc": utcnow_iso(), "counts": counts, "warnings": len(warnings), "errors": len(project_errors) + len(fetcher.errors), }) status.recompute_totals(current_ids) safe_save(status, logger) elapsed = time.time() - started problems = len(project_errors) + len(fetcher.errors) progress.finish_project( "%-9s %s - %d tasks, %d subtasks, %d comments, %d/%d files (%s) in %.1fs%s" % ("[partial]" if problems else "[ok]", project_name, counts["tasks"], counts["subtasks"], counts["comments"], counts["files_ok"], counts["files_ok"] + counts["files_failed"], fmt_bytes(counts["bytes"]), elapsed, " - %d problem(s)" % problems if problems else "")) logger("project %s exported to %s (%d problem(s))" % (project_id, relative, problems)) # -- wrap up -------------------------------------------------------------------- totals = status.recompute_totals(current_ids) if INTERRUPT["requested"]: run_outcome = "interrupted" final_status = "interrupted" elif totals["pending"] or totals["failed"]: # Every project in the account must have a completed record before this export # can call itself complete. run_outcome = "incomplete" final_status = "completed_with_errors" elif status.data["errors"]: run_outcome = "completed_with_errors" final_status = "completed_with_errors" else: run_outcome = "completed" final_status = "completed" status.data["status"] = final_status status.data["finished_at_utc"] = utcnow_iso() status.data["runs"].append({ "started_at_utc": datetime.fromtimestamp(run_started, timezone.utc).isoformat(), "finished_at_utc": utcnow_iso(), "projects_completed": totals["completed"], "outcome": run_outcome, }) safe_save(status, logger, progress) progress.close() print_summary(progress, status, out_dir, time.time() - run_started, final_status) logger("=== run end: %s (%d requests)" % (final_status, client.request_count)) if final_status == "completed": return EXIT_OK if final_status == "interrupted": return EXIT_INTERRUPTED return EXIT_COMPLETED_WITH_ERRORS except FatalError as exc: progress.close() sys.stderr.write("\nCannot continue: %s\n" % exc) if status is not None: status.data["status"] = "failed" status.add_error("fatal", str(exc)) try: status.save() except OSError: pass return EXIT_FAILED except KeyboardInterrupt: progress.close() sys.stderr.write("\nStopped. Re-run the same command to resume.\n") if status is not None: status.data["status"] = "interrupted" try: status.save() except OSError: pass return EXIT_INTERRUPTED except Exception as exc: # Last line of defence. Anything reaching here is a bug, but the customer still gets # an accurate status file, the full traceback in the log, and a resumable export. progress.close() logger("UNEXPECTED FAILURE: %s" % traceback.format_exc().replace("\n", " | ")) sys.stderr.write("\nThe export stopped unexpectedly: %s\n" % short_exception(exc)) sys.stderr.write("Details were written to export.log. Re-run the same command to " "resume from where it stopped.\n") if status is not None: status.data["status"] = "failed" status.add_error("unexpected", short_exception(exc)) try: status.save() except OSError: pass return EXIT_FAILED finally: logger.close() def safe_save(status, logger, progress=None): """Persisting the ledger must not abort the run, but you need to hear about it.""" try: status.save() return True except OSError as exc: message = "could not write %s: %s" % (status.path, short_exception(exc)) logger(message) if progress is not None: progress.note("WARNING: %s - resume information may be out of date" % message) return False def should_skip(record, out_dir): """A project is only skipped if the ledger, the file, and its checksum all agree.""" if not record or record.get("status") != "completed": return False relative = record.get("file") if not relative: return False path = os.path.join(out_dir, relative) if not os.path.isfile(path): return False expected = record.get("sha256") if not expected: return False return sha256_of_file(path) == expected def print_summary(progress, status, out_dir, elapsed, final_status): totals = status.data.get("totals") or {} errors = status.data.get("errors") or [] lines = [] if final_status == "completed": lines.append("EXPORT COMPLETE") elif final_status == "interrupted": lines.append("EXPORT INCOMPLETE - re-run the same command to resume") else: lines.append("EXPORT FINISHED WITH ERRORS - re-run the same command to retry") lines.append(" projects %d completed, %d failed, %d not yet exported" % (totals.get("completed", 0), totals.get("failed", 0), totals.get("pending", 0))) lines.append(" tasks %d subtasks %d comments %d" % (totals.get("tasks", 0), totals.get("subtasks", 0), totals.get("comments", 0))) lines.append(" files %d downloaded, %d failed, %s" % (totals.get("files_downloaded", 0), totals.get("files_failed", 0), fmt_bytes(totals.get("bytes_downloaded", 0)))) lines.append(" elapsed %s" % fmt_duration(elapsed)) lines.append(" output %s" % os.path.abspath(out_dir)) if errors: lines.append(" %d problem(s) recorded - see %s" % (len(errors), os.path.join(out_dir, STATUS_FILENAME))) sys.stderr.write("\n" + "\n".join(lines) + "\n") if __name__ == "__main__": sys.exit(main(sys.argv[1:]))