#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ticrypt-snapshot.py - capture a tiCrypt system snapshot over the REST API. Reproduces the JSON exports offered by the tiCrypt management UI so a system inventory can be taken on a schedule rather than by hand, one browser click at a time. Standard library only. Python 3.9+. No pip installs. python3 ticrypt-snapshot.py --base-url https://ticrypt.yourinstitution.edu --token KEY Create the API key in the tiCrypt front end under Management -> Miscellaneous -> API Keys, and mark it read-only: this script never writes. As of backend 3.14.1 a key carries several types at once. Select every type (Drive, VM, Team, Project, User) and one key covers every route called here. On earlier versions a key carries exactly one type, so pass five: python3 ticrypt-snapshot.py --base-url https://ticrypt.yourinstitution.edu \ --token-user USERKEY --token-team TEAMKEY --token-drive DRIVEKEY \ --token-vm VMKEY --token-project PROJKEY Prefer the environment over the command line so keys stay out of shell history and the process table. TICRYPT_BASE_URL, TICRYPT_TOKEN, and TICRYPT_TOKEN_ are read in preference to the matching flags: export TICRYPT_BASE_URL=https://ticrypt.yourinstitution.edu export TICRYPT_TOKEN=KEY python3 ticrypt-snapshot.py Reference: https://ticrypt.com/docs/admin-guide/operations/rest-api Some columns in the UI exports have no REST route behind them yet. This script never invents them: it writes null (or an empty file) and records exactly what was missing, and why, in _snapshot-manifest.json. """ from __future__ import annotations import argparse import json import os import sys import urllib.error import urllib.parse import urllib.request from datetime import datetime __version__ = "1.0.0" # Token scopes as documented in the REST API reference. The internal "all" key # holds a single token carrying every type; when present it is used throughout. SCOPES = ("user", "team", "vm", "drive", "project") # UserInfo.role carries the rank name; the numeric rank is its index. RANKS = ("User", "Sub-admin", "Admin", "Super-admin") # There is no route to enumerate realms. Override with --realms when a # deployment uses something other than the default. DEFAULT_REALMS = ("libvirt",) # -------------------------------------------------------------------------- # Gap tracking # -------------------------------------------------------------------------- GAPS: list[dict] = [] _SEEN_GAPS: set[tuple] = set() def gap(export: str, field: str, reason: str, proposed: str = "") -> None: """Record a field this snapshot cannot populate. Deduplicated.""" key = (export, field) if key in _SEEN_GAPS: return _SEEN_GAPS.add(key) GAPS.append( { "export": export, "field": field, "reason": reason, "proposedRoute": proposed or None, } ) # -------------------------------------------------------------------------- # Transport # -------------------------------------------------------------------------- class ApiError(RuntimeError): pass class Client: """Thin wrapper over urllib for the documented token-in-query scheme.""" def __init__(self, base_url, tokens, timeout=30, replay=None): self.base_url = (base_url or "").rstrip("/") self.tokens = tokens self.timeout = timeout self.replay = replay def token_for(self, scope: str) -> str: tok = self.tokens.get("all") or self.tokens.get(scope) if not tok: raise ApiError( "no token available for %s routes - pass --token with a key " "carrying every type, or --token-%s" % (scope, scope) ) return tok @staticmethod def _slug(path: str, params: dict | None) -> str: stem = path.strip("/").replace("/", "-") if params: extra = "-".join(str(params[k]) for k in sorted(params)) stem = "%s-%s" % (stem, extra) return "".join(c if (c.isalnum() or c in "-_") else "_" for c in stem) + ".json" def get(self, path: str, scope: str, params: dict | None = None): """GET a documented route. Returns decoded JSON.""" if self.replay: src = os.path.join(self.replay, self._slug(path, params)) if not os.path.exists(src): raise ApiError("replay file missing: %s" % src) with open(src, "r", encoding="utf-8") as fh: return json.load(fh) query = dict(params or {}) query["token"] = self.token_for(scope) url = "%s%s?%s" % (self.base_url, path, urllib.parse.urlencode(query)) req = urllib.request.Request(url, method="GET") req.add_header("Accept", "application/json") try: with urllib.request.urlopen(req, timeout=self.timeout) as r: body = r.read().decode("utf-8") except urllib.error.HTTPError as e: detail = "" try: detail = e.read().decode("utf-8", "replace").strip()[:400] except Exception: pass if e.code in (401, 403): raise ApiError( "%s on %s - the key is missing, expired, or not of a type that " "grants %s routes. Check the key type in Management -> " "Miscellaneous -> API Keys. %s" % (e.code, path, scope, detail) ) raise ApiError("HTTP %s on %s. %s" % (e.code, path, detail)) except urllib.error.URLError as e: raise ApiError("cannot reach %s%s - %s" % (self.base_url, path, e.reason)) if not body.strip(): return None try: return json.loads(body) except ValueError: raise ApiError("%s did not return JSON" % path) # -------------------------------------------------------------------------- # Fetchers - one per documented route # -------------------------------------------------------------------------- def collect(client: Client, realms) -> dict: """Call every readable documented route. Returns raw responses by key.""" raw: dict = {} def attempt(key, fn, optional=False): try: raw[key] = fn() except ApiError as e: if not optional: raise warn("skipping %s - %s" % (key, e)) raw[key] = [] log("fetching users") attempt("users", lambda: client.get("/api/users", "user")) log("fetching teams") attempt("teams", lambda: client.get("/api/teams", "team")) log("fetching drives") attempt("drives", lambda: client.get("/api/drives", "drive"), optional=True) log("fetching projects") attempt("projects", lambda: client.get("/api/projects", "project"), optional=True) raw["vms"] = [] raw["vmConfigs"] = [] raw["images"] = [] for realm in realms: log("fetching realm %s" % realm) for key, path, params in ( ("vms", "/api/vms/%s/active" % realm, None), ("vmConfigs", "/api/vms/%s/configs" % realm, None), ("images", "/api/vms/images", {"realm": realm}), ): try: raw[key].extend(client.get(path, "vm", params) or []) except ApiError as e: warn("skipping %s for realm %s - %s" % (key, realm, e)) return raw # -------------------------------------------------------------------------- # Helpers # -------------------------------------------------------------------------- def pct(used, quota): """Percentage of quota consumed. 0 when there is no quota to measure.""" if not quota: return 0 return (used or 0) * 100 / quota def team_list(names) -> str: """UI summary column: none, up to two names, or a count.""" names = sorted(names, key=lambda s: (s or "").lower()) if not names: return "NO TEAMS" if len(names) <= 2: return ", ".join(names) return "%d teams" % len(names) def display_name(first, last, uid) -> str: full = ("%s %s" % (first or "", last or "")).strip() return full if full else "[unknown-user:%s]" % uid def storage_quota(quotas: dict, *keys): """Pull a byte quota out of QuotasTyped.storage by any of several keys. The storage map's key names are not documented; try the plausible ones. """ store = (quotas or {}).get("storage") or {} for k in keys: if k in store: return store[k] return None # -------------------------------------------------------------------------- # Normalizers - raw API shapes into the UI export shapes # -------------------------------------------------------------------------- def build_users(raw) -> list: users = raw.get("users") or [] teams = raw.get("teams") or [] membership: dict = {} for t in teams: for m in t.get("members") or []: membership.setdefault(m.get("id"), []).append(t.get("name") or "") gap( "users", "roleName", "profile names come from user profiles, which are a front-end " "construct with no backing store", ) rows = [] for u in users: perms = u.get("permissions") or [] role = u.get("role") state = u.get("state") or "" row = { "id": u.get("id"), "name": ("%s %s" % (u.get("firstName") or "", u.get("lastName") or "")).strip(), "firstName": u.get("firstName"), "lastName": u.get("lastName"), "email": u.get("email"), "contactEmail": u.get("contactEmail", ""), "state": state, "rank": RANKS.index(role) if role in RANKS else None, "rankName": role, } # Row-styling hint reproduced from the UI export. if state == "Deactivated": row["__class__"] = "error" elif state == "New, not activated": row["__class__"] = "ok" row["joined"] = u.get("joined") if u.get("lastLogin") is not None: row["lastLogin"] = u.get("lastLogin") row["roleName"] = None # deactivationType doubles as the UI's "reason" column. row["reason"] = u.get("deactivationType") or "Active" row["canEscrow"] = not bool(u.get("neverEscrow")) for opt in ("passChanged", "tosAccepted", "lastEscrow", "expiration"): if u.get(opt) is not None: row[opt] = u[opt] row["permissionList"] = perms row["stateReason"] = u.get("stateReason", "") row["teams"] = {"list": team_list(membership.get(u.get("id"), []))} rows.append(row) if users and not any(u.get("role") in RANKS for u in users): gap( "users", "rank / rankName", "UserInfo.role did not contain a recognized rank name", "add rank:number and rankName:string to UserInfo", ) return rows def build_teams(raw) -> list: teams = raw.get("teams") or [] drives = raw.get("drives") or [] vms = raw.get("vms") or [] drive_by_team: dict = {} for d in drives: drive_by_team.setdefault(d.get("team"), []).append(d) vm_by_team: dict = {} for v in vms: vm_by_team.setdefault(v.get("team"), []).append(v) gap( "teams", "files", "no route reports per-team vault usage", "GET /api/teams/usage", ) gap("teams", "intakeID", "no intake reference is stored on a team") gap( "teams", "customFields", "no deployment-defined fields are stored on a team", ) rows = [] for t in teams: tq = t.get("teamQuotas") or {} q_cores = tq.get("cores") q_mem = tq.get("memory") q_drive = storage_quota(tq, "drive", "drives", "driveSize") q_file = storage_quota(tq, "vault", "file", "files", "fileSize") q_total = None if q_drive is not None or q_file is not None: q_total = (q_drive or 0) + (q_file or 0) my_drives = drive_by_team.get(t.get("id"), []) my_vms = vm_by_team.get(t.get("id"), []) drive_size = sum(d.get("capacity") or 0 for d in my_drives) vm_cores = sum((v.get("resources") or {}).get("vcpus") or 0 for v in my_vms) vm_mem = sum((v.get("resources") or {}).get("memory") or 0 for v in my_vms) rows.append( { "id": t.get("id"), "name": t.get("name"), "qCores": q_cores, "qMemory": q_mem, "qTotal": q_total, "qDriveSize": q_drive, "qFileSize": q_file, "intakeID": None, "customFields": None, "members": {"cnt": len(t.get("members") or [])}, "files": {"cnt": None, "size": None}, "vms": {"cnt": len(my_vms), "mem": vm_mem, "cores": vm_cores}, "drives": { "cnt": len(my_drives), "size": drive_size, "used": sum(1 for d in my_drives if d.get("writer")), }, "corePct": pct(vm_cores, q_cores), "memPct": pct(vm_mem, q_mem), "drivePct": pct(drive_size, q_drive), # totalStorage = files.size + drives.size; files.size is unavailable. "totalStorage": None, } ) # The UI appends a synthetic row for resources owned by no team. orphan_drives = drive_by_team.get(None, []) + drive_by_team.get("", []) orphan_vms = vm_by_team.get(None, []) + vm_by_team.get("", []) if orphan_drives or orphan_vms: rows.append( { "members": {"cnt": 0}, "files": {"cnt": None, "size": None}, "vms": { "cnt": len(orphan_vms), "mem": sum((v.get("resources") or {}).get("memory") or 0 for v in orphan_vms), "cores": sum((v.get("resources") or {}).get("vcpus") or 0 for v in orphan_vms), }, "drives": { "cnt": len(orphan_drives), "size": sum(d.get("capacity") or 0 for d in orphan_drives), "used": sum(1 for d in orphan_drives if d.get("writer")), }, "name": "No Team Assigned", } ) return rows def build_team_memberships(raw) -> list: teams = raw.get("teams") or [] gap( "team_memberships", "joined / modified / perm", "TeamMemberInfo carries no join date or per-team permission", "add joined, modified, perm to TeamMemberInfo", ) rows = [] for t in teams: tid = t.get("id") for m in t.get("members") or []: uid = m.get("id") known = bool((m.get("firstName") or "").strip() or (m.get("lastName") or "").strip()) row = { "id": "%s*%s" % (tid, uid), "userID": uid, "user": display_name(m.get("firstName"), m.get("lastName"), uid), "teamID": tid, "team": t.get("name"), } # The UI omits lastLogin entirely for users it cannot resolve. if known: row["lastLogin"] = m.get("lastLogin", 0) row["joined"] = None row["modified"] = None row["perm"] = None rows.append(row) return rows def build_drives(raw) -> list: drives = raw.get("drives") or [] users = {u.get("id"): u for u in raw.get("users") or []} teams = {t.get("id"): t for t in raw.get("teams") or []} projects = {p.get("id"): p for p in raw.get("projects") or []} for field, why in ( ("sizeOnDisk", "DriveInfo reports capacity but not actual disk consumption"), ("keys", "key count is not exposed on DriveInfo"), ("snapshotPool", "snapshot pool is not exposed on DriveInfo"), ("hasMessage", "not exposed on DriveInfo"), ): gap("drives", field, why, "add %s to DriveInfo" % field) for field, why in ( ("shares", "share count is derived in the interface, not stored"), ): gap("drives", field, why) gap("drives", "__class__", "cosmetic row-styling hint; unlike users it does not follow from state") rows = [] for d in drives: owner_id = d.get("owner") team_id = d.get("team") proj_id = d.get("project") owner = users.get(owner_id) team = teams.get(team_id) proj = projects.get(proj_id) proj_name = proj.get("name") if proj else None row = { "id": d.get("id"), "name": d.get("name"), "state": d.get("state"), "stateChange": d.get("stateChange"), "realmID": d.get("realmID"), "capacity": d.get("capacity"), "sizeOnDisk": None, "format": d.get("format"), "type": d.get("diskType") or "raw", "diskType": d.get("diskType"), "ownerID": owner_id, "owner": display_name( (owner or {}).get("firstName"), (owner or {}).get("lastName"), owner_id ), "teamID": team_id, "team": team.get("name") if team else None, "project": proj_id, "projectName": proj_name, "projectNameEnhanced": proj_name or "No project", "poolID": d.get("poolID"), "pool": d.get("poolID"), "noBackup": d.get("noBackup"), "backup": not bool(d.get("noBackup")), "hasSnapshot": d.get("hasSnapshot"), "readers": d.get("readers") or [], "created": d.get("created"), "keys": None, "shares": None, "snapshotPool": None, "hasMessage": None, } for src, dst in (("attached", "attached"), ("attached", "lastAttached"), ("changed", "changed"), ("writable", "writable"), ("writer", "writer"), ("cache", "cache"), ("io", "io")): if d.get(src) is not None: row[dst] = d[src] rows.append(row) return rows def build_projects(raw) -> list: projects = raw.get("projects") or [] by_id = {p.get("id"): p for p in projects} gap("projects", "pi", "no principal investigator is stored on a project") gap("projects", "customFields", "no deployment-defined fields are stored on a project") gap("projects", "level", "securityLevel is a bare ID with no route to resolve the name", "GET /api/security-levels") gap("projects", "members", "project membership counts require a memberships route", "GET /api/projects/memberships") rows = [] for p in projects: row = { "id": p.get("id"), "name": p.get("name"), "pi": None, "levelID": p.get("securityLevel"), "level": None, "customFields": None, "members": {"mbr": None, "mnag": None}, } if p.get("parent"): row["parentID"] = p["parent"] row["parent"] = (by_id.get(p["parent"]) or {}).get("name") rows.append(row) return rows def build_resources_by_project(raw) -> list: projects = raw.get("projects") or [] drives = raw.get("drives") or [] vms = raw.get("vms") or [] gap("resources_by_project", "files", "no route reports per-project vault usage", "GET /api/projects/usage") by_drive: dict = {} for d in drives: by_drive.setdefault(d.get("project"), []).append(d) by_vm: dict = {} for v in vms: by_vm.setdefault(v.get("project"), []).append(v) def rollup(name, ds, vs, pid=None): row = {} if pid is not None: row["id"] = pid row["name"] = name row["files"] = {"cnt": None, "size": None} row["vms"] = { "cnt": len(vs), "mem": sum((v.get("resources") or {}).get("memory") or 0 for v in vs), "cores": sum((v.get("resources") or {}).get("vcpus") or 0 for v in vs), } row["drives"] = { "cnt": len(ds), "size": sum(d.get("capacity") or 0 for d in ds), "used": sum(1 for d in ds if d.get("writer")), } return row rows = [ rollup(p.get("name"), by_drive.get(p.get("id"), []), by_vm.get(p.get("id"), []), p.get("id")) for p in projects ] # The UI appends a synthetic row for resources tagged to no project. loose_d = by_drive.get(None, []) + by_drive.get("", []) loose_v = by_vm.get(None, []) + by_vm.get("", []) if loose_d or loose_v: rows.append(rollup("No project", loose_d, loose_v)) return rows def build_vm_images(raw) -> list: images = raw.get("images") or [] by_id = {i.get("id"): i for i in images} for field, why, route in ( ("realmName", "realms cannot be enumerated", "GET /api/realms"), ("secureBoot", "not exposed on ImageInfo", "add secureBoot to ImageInfo"), ("bricks", "hardware setups cannot be listed", "GET /api/vms/bricks"), ): gap("vm_images", field, why, route) for field, why in ( ("canModify", "not stored on an image; the interface evaluates it"), ("capabilities", "derived in the interface, not stored on an image"), ): gap("vm_images", field, why) rows = [] for i in images: info = i.get("info") or {} pool, volume = info.get("pool"), info.get("volume") settings = i.get("settings") or {} row = { "id": i.get("id"), "name": i.get("name"), "info": "%s::%s" % (pool, volume) if pool and volume else None, "parentName": (by_id.get(i.get("parent")) or {}).get("name", ""), "realmName": None, "size": i.get("size"), "canModify": None, "type": i.get("imageType"), "created": i.get("created"), "deviceBusType": info.get("deviceBusType"), "secureBoot": None, "driveFormats": settings.get("driveFormats"), "capabilities": None, "bricks": {"cnt": None}, } if i.get("modified") is not None: row["modified"] = i["modified"] if i.get("limit") is not None: row["limit"] = i["limit"] rows.append(row) return rows def _unreachable(export, reason, route): def build(raw, _e=export, _r=reason, _rt=route): gap(_e, "*", _r, _rt) return [] return build build_security_requirements = _unreachable( "security_requirements", "no route exposes security requirements", "GET /api/security-requirements", ) build_security_levels = _unreachable( "security_levels", "no route exposes security levels", "GET /api/security-levels" ) build_project_memberships = _unreachable( "project_memberships", "no route exposes project membership records", "GET /api/projects/memberships", ) build_user_certifications = _unreachable( "user_certifications", "no route exposes security-requirement certifications", "GET /api/users/certifications", ) build_licensing_servers = _unreachable( "licensing_servers", "external/licensing servers can be referenced but not listed", "GET /api/external-servers", ) build_vm_hardware_setups = _unreachable( "vm_hardware_setups", "bricks can be edited via PATCH but never listed", "GET /api/vms/bricks", ) def build_user_profiles(raw) -> list: gap( "user_profiles", "*", "user profiles are a front-end construct; the backend has no " "profile objects to serve", ) return [] def build_subadmin_managed_objects(raw) -> list: gap( "sub-admin_managed_objects", "*", "no route exposes sub-admin managed object assignments", "GET /api/users/managed-objects", ) return [] # Adding an export type is one entry here plus one build_* function. EXPORTS = ( {"name": "users", "build": build_users}, {"name": "user_profiles", "build": build_user_profiles}, {"name": "user_certifications", "build": build_user_certifications}, {"name": "sub-admin_managed_objects", "build": build_subadmin_managed_objects}, {"name": "teams", "build": build_teams}, {"name": "team_memberships", "build": build_team_memberships}, {"name": "projects", "build": build_projects}, {"name": "project_memberships", "build": build_project_memberships}, {"name": "resources_by_project", "build": build_resources_by_project}, {"name": "security_requirements", "build": build_security_requirements}, {"name": "security_levels", "build": build_security_levels}, {"name": "drives", "build": build_drives}, {"name": "vm_images", "build": build_vm_images}, {"name": "vm_hardware_setups", "build": build_vm_hardware_setups}, {"name": "licensing_servers", "build": build_licensing_servers}, ) # -------------------------------------------------------------------------- # Output # -------------------------------------------------------------------------- def log(msg): sys.stderr.write(" %s\n" % msg) def warn(msg): sys.stderr.write(" ! %s\n" % msg) def write_json(path, payload): with open(path, "w", encoding="utf-8") as fh: json.dump(payload, fh, indent=2, ensure_ascii=False) fh.write("\n") def export_filename(name, rows, when: datetime) -> str: """Build '- (--).json', the UI naming convention. Both parts vary per run: the count is the number of records written for this export, the date is the day the snapshot was taken. """ return "%d-%s (%d-%d-%d).json" % ( len(rows), name, when.month, when.day, when.year, ) # -------------------------------------------------------------------------- # Entry point # -------------------------------------------------------------------------- def parse_args(argv): p = argparse.ArgumentParser( prog="ticrypt-snapshot.py", description="Capture a tiCrypt system snapshot over the REST API.", epilog="Create API keys in Management -> Miscellaneous -> API Keys.", ) p.add_argument("--base-url", default=os.environ.get("TICRYPT_BASE_URL"), help="backend domain, e.g. https://ticrypt.yourinstitution.edu") p.add_argument("--token", default=os.environ.get("TICRYPT_TOKEN"), help="API key with every type selected (covers every " "route; needs backend 3.14.1+)") labels = {"user": "User", "team": "Team", "vm": "VM", "drive": "Drive", "project": "Project"} for s in SCOPES: p.add_argument("--token-%s" % s, dest="token_%s" % s, default=os.environ.get("TICRYPT_TOKEN_%s" % s.upper()), help="API key of type %s" % labels[s]) p.add_argument("--realms", default="libvirt", help="comma-separated realm IDs (default: libvirt)") p.add_argument("--out-dir", default=None, help="output directory (default: ./ticrypt-snapshot-)") p.add_argument("--raw", action="store_true", help="also write raw/ with unmodified API responses") p.add_argument("--timeout", type=int, default=30, help="per-request timeout (default: 30)") p.add_argument("--replay", default=None, help="read saved responses from a directory instead of calling the API") p.add_argument("--version", action="version", version="%(prog)s " + __version__) return p.parse_args(argv) def main(argv=None): args = parse_args(argv if argv is not None else sys.argv[1:]) tokens = {} if args.token: tokens["all"] = args.token for s in SCOPES: v = getattr(args, "token_%s" % s, None) if v: tokens[s] = v if not args.replay: if not args.base_url: sys.stderr.write("error: --base-url is required (or set TICRYPT_BASE_URL)\n") return 2 if not tokens: sys.stderr.write( "error: no API key given. Pass --token with a key carrying every\n" " type (backend 3.14.1+), or one --token- per scope.\n" " Keys are created in the tiCrypt front end under\n" " Management -> Miscellaneous -> API Keys.\n" ) return 2 started = datetime.now() out_dir = args.out_dir or "ticrypt-snapshot-%s" % started.strftime("%Y-%m-%d") os.makedirs(out_dir, exist_ok=True) realms = [r.strip() for r in (args.realms or "").split(",") if r.strip()] or list(DEFAULT_REALMS) client = Client( base_url=args.base_url, tokens=tokens, timeout=args.timeout, replay=args.replay, ) sys.stderr.write("tiCrypt snapshot %s\n" % __version__) try: raw = collect(client, realms) except ApiError as e: sys.stderr.write("error: %s\n" % e) return 1 if args.raw: raw_dir = os.path.join(out_dir, "raw") os.makedirs(raw_dir, exist_ok=True) for key, payload in raw.items(): write_json(os.path.join(raw_dir, "%s.json" % key), payload) written = [] for spec in EXPORTS: rows = spec["build"](raw) fname = export_filename(spec["name"], rows, started) write_json(os.path.join(out_dir, fname), rows) written.append({"export": spec["name"], "file": fname, "rows": len(rows)}) log("wrote %s" % fname) manifest = { "tool": "ticrypt-snapshot.py", "version": __version__, "captured": started.isoformat(timespec="seconds"), "baseUrl": args.base_url, "realms": realms, "authMode": ("single key, all types" if "all" in tokens else "one key per type"), "exports": written, "gaps": GAPS, } write_json(os.path.join(out_dir, "_snapshot-manifest.json"), manifest) sys.stderr.write("\nSnapshot written to %s\n" % os.path.abspath(out_dir)) if GAPS: sys.stderr.write( "\n%d field(s) could not be captured from the documented API:\n" % len(GAPS) ) for g in GAPS: suffix = " -> needs %s" % g["proposedRoute"] if g["proposedRoute"] else "" sys.stderr.write(" - %s.%s: %s%s\n" % (g["export"], g["field"], g["reason"], suffix)) sys.stderr.write("\nSee _snapshot-manifest.json for the full list.\n") return 0 if __name__ == "__main__": try: sys.exit(main()) except KeyboardInterrupt: sys.exit(130)