#!/usr/bin/env python3 """Publish private Tacksheet reviews with resumable, idempotent uploads. Python 3.10+.""" from __future__ import annotations import argparse from contextlib import contextmanager import datetime import email.utils import hashlib import http.client import json import math import os from pathlib import Path import re import sys import time import urllib.error import urllib.parse import urllib.request import uuid try: import tacksheet_sheet # Sheet-kit subcommands (single hook; implementation lives there). except ImportError: tacksheet_sheet = None # type: ignore[assignment] DEFAULT_ORIGIN = "https://tacksheet.com" # The largest plan's file (Ultra); each workspace's plan sets its own limit, enforced by the API. MAX_FILE = 10 * 1024 * 1024 * 1024 SHEET_MAX_FILE = 90 * 1024 * 1024 # Files above the single-request size upload in parts; the plan names the part size (5-64 MiB). MIN_PART = 5 * 1024 * 1024 MAX_PART = 64 * 1024 * 1024 PART_DIGEST_HEADER = "X-Part-Sha256" MAX_RESPONSE = 16 * 1024 * 1024 MAX_EXPORT_BYTES = 20 * 1024 * 1024 * 1024 MAX_EXPORT_FILES = 10000 ID = re.compile(r"[A-Za-z0-9_-]{16,64}\Z") HASH = re.compile(r"[a-f0-9]{64}\Z") TYPES = {".html": "text/html", ".htm": "text/html", ".txt": "text/plain", ".md": "text/markdown", ".json": "application/json", ".pdf": "application/pdf", ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".webp": "image/webp", ".gif": "image/gif", ".svg": "image/svg+xml", ".avif": "image/avif", ".csv": "text/csv", ".zip": "application/zip", ".mp4": "video/mp4", ".webm": "video/webm", ".mp3": "audio/mpeg", ".ogg": "audio/ogg", ".wav": "audio/wav", ".glb": "model/gltf-binary", ".gltf": "model/gltf+json"} HTML_TYPES = {"text/html"} class ClientError(Exception): """Safe, credential-free diagnostics only. retry_after: seconds the server asked to wait, if any.""" def __init__(self, message: str, status: int | None = None, code: str | None = None, retry_after: int | None = None): super().__init__(message) self.status, self.code, self.retry_after = status, code, retry_after # A request sleeps through the server's Retry-After up to this long at a time, and this long in all. RETRY_WAIT_CAP = 60 RETRY_WAIT_TOTAL = 120 # 429s that waiting does not clear: the workspace must make room or change plan first. # The plan's counted limits: what makes room, so an agent can offer it (tack-it SKILL.md "Running out of room"). ROOM_TIPS = { "plan_version_limit": "This sheet has no room for another version. Offer to delete its old versions " "(tackit.py versions REVIEW_ID, then versions REVIEW_ID --keep N), or give the owner {cleanup}.", "plan_review_limit": "The plan's limit on active sheets is reached. Offer to move old or unimportant sheets to " "Recently deleted, or give the owner {cleanup}.", "plan_storage_exceeded": "The plan's storage is full. Offer to delete old versions or move old sheets to Recently " "deleted, or give the owner {cleanup}.", } QUOTA_CODES = frozenset({"quota_exceeded", "workspace_capacity_exceeded", "recovery_capacity_exceeded", "feedback_limit", "plan_storage_exceeded"}) def retry_after(headers) -> int | None: """Seconds from a Retry-After header (delay-seconds or an HTTP date), or None.""" value = headers.get("Retry-After") if headers is not None else None if not isinstance(value, str) or not value.strip(): return None value = value.strip() if value.isdigit(): return min(int(value), 10 ** 7) try: when = email.utils.parsedate_to_datetime(value) except (TypeError, ValueError, IndexError): return None if when.tzinfo is None: when = when.replace(tzinfo=datetime.timezone.utc) return max(0, math.ceil(when.timestamp() - time.time())) def retry_note(seconds: int | None, start: bool = False) -> str: """When to try again, as words: in N seconds, or after a UTC time.""" if seconds is None: note = "try again in a minute" elif seconds < 120: note = f"try again in {seconds} seconds (after {time.strftime('%H:%M:%S UTC', time.gmtime(time.time() + seconds))})" else: note = f"try again after {time.strftime('%Y-%m-%d %H:%M UTC', time.gmtime(time.time() + seconds))} (in about {math.ceil(seconds / 60)} minutes)" return note[0].upper() + note[1:] if start else note def limit_message(code: str | None, seconds: int | None) -> str: """Why a 429 happened and when to try again: a quota needs room, a rate limit needs time.""" if code in QUOTA_CODES: return (f"A workspace or plan limit was reached ({code}); waiting will not clear it. Make room " "or check the plan, then retry.") return "Tacksheet's request rate limit was reached (a limit on requests, not on storage); " + retry_note(seconds) + "." PUBLICATION_CONFLICT = ("The review changed after this upload was planned. Inspect the latest version, " "then run a new update with a fresh manifest. This upload cannot be resumed.") class SafeArgumentParser(argparse.ArgumentParser): def error(self, _message): # A user may mistakenly paste a key into an unsupported argument. Do not echo argv. self.print_usage(sys.stderr) self.exit(2, "Tacksheet: Invalid arguments. Run --help for usage.\n") def require(condition: bool, message: str) -> None: if not condition: raise ClientError(message) def origin_value(value: str, allow_local: bool = False) -> str: try: parsed = urllib.parse.urlsplit(value) port = parsed.port except ValueError as exc: raise ClientError("Invalid API origin.") from exc require(not parsed.username and not parsed.password and not parsed.query and not parsed.fragment and parsed.path in ("", "/"), "Use an origin without credentials, path, query, or fragment.") if value.rstrip("/") == DEFAULT_ORIGIN: return DEFAULT_ORIGIN require(allow_local and parsed.hostname in ("localhost", "127.0.0.1", "::1") and parsed.scheme in ("http", "https") and (port is None or 1 <= port <= 65535), "Only https://tacksheet.com is supported; local development needs --allow-local.") return value.rstrip("/") class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *_args, **_kwargs): raise ClientError("API redirect refused; credentials were not forwarded.") def encode(value: object) -> bytes: return (json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + "\n").encode() def decode(data: bytes) -> dict: try: value = json.loads(data) except (UnicodeError, ValueError, RecursionError) as exc: raise ClientError("The API returned invalid JSON. Resume the saved upload to retry safely.") from exc require(isinstance(value, dict), "The API returned an invalid response object.") return value def bounded_read(response, limit: int, timeout: int) -> bytes: deadline = time.monotonic() + timeout chunks = [] received = 0 # HTTPResponse.read1 performs at most one underlying read, so a drip-fed response cannot # keep read(limit) blocked indefinitely while repeatedly resetting the socket timeout. while True: require(time.monotonic() <= deadline, "API response deadline exceeded. Resume the saved upload to retry safely.") part = response.read1(min(65536, limit + 1 - received)) if not part: return b"".join(chunks) received += len(part) require(received <= limit, "API response exceeds the client limit.") chunks.append(part) class Client: def __init__(self, origin: str, key: str, timeout: int = 75, attempts: int = 3): require(re.fullmatch(r"pbk_[A-Za-z0-9_-]{43}", key or "") is not None, "Set TACKSHEET_API_KEY to a scoped developer key from your workspace.") require(1 <= timeout <= 120 and 1 <= attempts <= 3, "Invalid request budget.") self.origin, self._key, self.timeout, self.attempts = origin, key, timeout, attempts self.opener = urllib.request.build_opener(NoRedirect()) def request(self, method: str, path: str, *, payload: dict | None = None, source: Path | None = None, file: dict | None = None, idempotency: str | None = None, limit: int = 65536, part: bytes | None = None) -> bytes: require((path.startswith("/api/v1/reviews") or path.startswith("/api/v1/categories") or path.startswith("/api/v1/workspaces") or path.startswith("/api/v1/account/export") or path.startswith("/api/v1/styles") or path.startswith("/api/v1/signatures") or path == "/api/v1/webhooks" or path.startswith("/api/v1/webhooks/") or path == "/api/v1/product-feedback" or path == "/api/v1/me") and not path.startswith("//") and "#" not in path and "\\" not in path, "Unexpected API path refused.") waited = 0 for attempt in range(self.attempts): delay = attempt + 1 # Unless the server says how long to wait. headers = {"Authorization": "Bearer " + self._key, "Accept": "application/json", "User-Agent": "tacksheet-cli/1"} if idempotency: headers["Idempotency-Key"] = idempotency body = encode(payload) if payload is not None else None if payload is not None: headers["Content-Type"] = "application/json" stream = None try: if part is not None: # One part of a large file: its exact length and the SHA-256 of its bytes. body = part headers.update({"Content-Type": "application/octet-stream", "Content-Length": str(len(part)), PART_DIGEST_HEADER: hashlib.sha256(part).hexdigest()}) if source is not None: require(file is not None, "Upload metadata missing.") stream = source.open("rb") verify_stream(stream, file) stream.seek(0) body = stream headers.update({"Content-Type": file["type"], "Content-Length": str(file["size"])}) request = urllib.request.Request(self.origin + path, data=body, headers=headers, method=method) with self.opener.open(request, timeout=self.timeout) as response: require(200 <= response.status < 300, "Unexpected API response status.") return bounded_read(response, limit, self.timeout) except urllib.error.HTTPError as exc: # Never echo the response body or urllib exception: either may contain credentials or content. try: error_body = bounded_read(exc, 65536, self.timeout) code = decode(error_body).get("error") except (ClientError, OSError, http.client.HTTPException): code = None finally: exc.close() code = code if isinstance(code, str) else None wait = retry_after(exc.headers) retry = exc.code in (408, 425, 429, 500, 502, 503, 504) or (method == "PUT" and exc.code == 409 and code in ("upload_superseded", "upload_starting")) if code in QUOTA_CODES: retry = False if wait is not None: delay = wait # Sleep through a short Retry-After; a long one is reported with when to try again. if delay > RETRY_WAIT_CAP or waited + delay > RETRY_WAIT_TOTAL: retry = False if not retry or attempt + 1 == self.attempts: tips = {401: "Check the developer key.", 403: "Check the key scopes and workspace access.", 402: "Workspace writes are unavailable; check billing and launch readiness.", 409: "The operation conflicts with current state. Resume only with unchanged files.", 410: "This upload plan expired. Start a new upload.", 422: "File bytes did not match the upload plan.", 429: limit_message(code, wait)} later = "Retry later with the saved upload manifest." if wait is None else retry_note(wait, start=True) + " with the saved upload manifest." tip = ROOM_TIPS[code].format(cleanup=self.origin + "/app#cleanup") if code in ROOM_TIPS else tips.get(exc.code, later) raise ClientError(f"API request failed (HTTP {exc.code}). " + tip, exc.code, code, wait) from None except (urllib.error.URLError, TimeoutError, ConnectionError, OSError, http.client.HTTPException): if attempt + 1 == self.attempts: raise ClientError("The API connection failed after bounded retries. Resume the saved upload; no new review is needed.") from None finally: if stream is not None: stream.close() time.sleep(delay) waited += delay raise ClientError("Request retry budget exhausted.") def json(self, method: str, path: str, **kwargs) -> dict: return decode(self.request(method, path, **kwargs)) def download(self, path: str, target: Path, size: int) -> str: """Streams one file to a new private file in bounded reads and returns its SHA-256. Large files never sit in memory; a slow but moving download gets about 256 KiB/s of deadline.""" require(path.startswith("/api/v1/reviews/") and "#" not in path and "\\" not in path, "Unexpected API path refused.") waited = 0 for attempt in range(self.attempts): delay = attempt + 1 headers = {"Authorization": "Bearer " + self._key, "User-Agent": "tacksheet-cli/1"} deadline = time.monotonic() + max(self.timeout, size // (256 * 1024)) try: request = urllib.request.Request(self.origin + path, headers=headers, method="GET") with self.opener.open(request, timeout=self.timeout) as response: require(200 <= response.status < 300, "Unexpected API response status.") digest, received = hashlib.sha256(), 0 descriptor = os.open(target, os.O_CREAT | os.O_TRUNC | os.O_WRONLY, 0o600) with os.fdopen(descriptor, "wb") as output: while True: require(time.monotonic() <= deadline, "Download deadline exceeded.") chunk = response.read1(1024 * 1024) if not chunk: break received += len(chunk) require(received <= size, "A downloaded export file exceeds its listed size.") digest.update(chunk) output.write(chunk) return digest.hexdigest() except urllib.error.HTTPError as exc: exc.close() wait = retry_after(exc.headers) delay = delay if wait is None else wait if (exc.code not in (408, 429, 500, 502, 503, 504) or attempt + 1 == self.attempts or delay > RETRY_WAIT_CAP or waited + delay > RETRY_WAIT_TOTAL): raise ClientError(f"API request failed (HTTP {exc.code})." + ( " " + retry_note(wait, start=True) + "." if wait is not None else ""), exc.code, None, wait) from None except (urllib.error.URLError, TimeoutError, ConnectionError, OSError, http.client.HTTPException): if attempt + 1 == self.attempts: raise ClientError("The API connection failed after bounded retries.") from None time.sleep(delay) waited += delay raise ClientError("Request retry budget exhausted.") def verify_stream(stream, file: dict) -> None: digest = hashlib.sha256() size = 0 limit = SHEET_MAX_FILE if file.get("interactive") is True else MAX_FILE for chunk in iter(lambda: stream.read(65536), b""): size += len(chunk) require(size <= limit, "File exceeds the size limit.") digest.update(chunk) require(size == file["size"] and digest.hexdigest() == file["sha256"], "A source file changed. Keep the original bytes for resume, or start a new upload.") def describe_file(value: str, interactive: bool = False) -> dict: path = Path(value).expanduser().resolve(strict=True) require(path.is_file(), "Every upload source must be a regular file.") media_type = TYPES.get(path.suffix.lower()) require(media_type is not None, "Unsupported file extension.") require(not interactive or media_type in HTML_TYPES, "Only HTML files can be interactive sheets.") limit = SHEET_MAX_FILE if interactive else MAX_FILE size = path.stat().st_size require(1 <= size <= limit, "File size is outside the supported range.") require(1 <= len(path.name) <= 160, "File names must contain 1 to 160 characters.") require("/" not in path.name and "\\" not in path.name and ".." not in path.name and not path.name.startswith("."), "File names cannot contain path syntax.") digest = hashlib.sha256() actual = 0 with path.open("rb") as stream: for chunk in iter(lambda: stream.read(65536), b""): actual += len(chunk) require(actual <= limit, "File exceeds the size limit.") digest.update(chunk) require(size == actual, "A source file changed while preparing the upload.") file = {"path": str(path), "name": path.name, "type": media_type, "size": size, "sha256": digest.hexdigest()} if interactive: file["interactive"] = True return file def safe_state_path(path: Path) -> Path: path = path.expanduser().absolute() require(not path.is_symlink(), "Upload manifest symlinks are refused.") for parent in path.parents: require(not parent.is_symlink(), "Upload manifest directory symlinks are refused.") path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) return path @contextmanager def locked(path: Path): lock = path.with_name(path.name + ".lock") require(not lock.is_symlink(), "Upload manifest lock symlinks are refused.") descriptor = os.open(lock, os.O_CREAT | os.O_RDWR, 0o600) stream = os.fdopen(descriptor, "r+b") try: if os.fstat(stream.fileno()).st_size == 0: stream.write(b"0") stream.flush() stream.seek(0) try: if os.name == "nt": import msvcrt msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1) else: import fcntl fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except OSError as exc: raise ClientError("Another process is using this upload manifest.") from exc yield finally: stream.close() # OS lock also releases automatically after a crash. def save(path: Path, state: dict) -> None: require(not path.is_symlink(), "Upload manifest symlinks are refused.") temporary = path.with_name(path.name + "." + uuid.uuid4().hex + ".tmp") try: descriptor = os.open(temporary, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) with os.fdopen(descriptor, "wb") as stream: stream.write(encode(state)) stream.flush() os.fsync(stream.fileno()) os.replace(temporary, path) finally: temporary.unlink(missing_ok=True) def plan_shape(plan: dict, files: list[dict], expected_review: str | None = None) -> dict: review = plan.get("review", {}).get("id") version = plan.get("version", {}).get("id") require(isinstance(review, str) and ID.fullmatch(review) is not None and isinstance(version, str) and ID.fullmatch(version) is not None, "Invalid review or version identifier in upload plan.") require(expected_review is None or review == expected_review, "The API returned a different review.") prefix = f"/api/v1/reviews/{review}/versions/{version}" require(plan.get("finalizeUrl") == prefix + "/finalize", "Unexpected finalize URL refused.") uploads = plan.get("uploads") require(isinstance(uploads, list) and len(uploads) == len(files), "Upload plan file count mismatch.") remaining = list(files) normalized = [] ids = set() for upload in uploads: require(isinstance(upload, dict), "Invalid upload entry.") asset_id = upload.get("id") require(isinstance(asset_id, str) and ID.fullmatch(asset_id) is not None and asset_id not in ids, "Invalid or duplicate asset identifier.") ids.add(asset_id) require(upload.get("method") == "PUT" and upload.get("url") == prefix + "/assets/" + asset_id, "Unexpected upload URL refused.") matching = next((file for file in remaining if file["size"] == upload.get("size") and file["sha256"] == upload.get("sha256")), None) require(matching is not None, "Upload plan file integrity mismatch.") remaining.remove(matching) entry = {"id": asset_id, "url": upload["url"], "method": "PUT", "size": matching["size"], "sha256": matching["sha256"], "source": matching["path"]} # The workspace already stores this file (shared objects, migration 0049): nothing to send. if upload.get("uploaded") is True: entry["uploaded"] = True # A large file uploads in parts. Part and completion URLs derive from the checked upload URL; # only the layout is read from the plan (a saved plan keeps it as partBytes). layout = upload.get("multipart") if "multipart" in upload else ( {"partBytes": upload["partBytes"], "partCount": -(-matching["size"] // upload["partBytes"])} if type(upload.get("partBytes")) is int else None) if layout is not None: require(isinstance(layout, dict), "Invalid multipart upload entry.") part_bytes = layout.get("partBytes") require(type(part_bytes) is int and MIN_PART <= part_bytes <= MAX_PART and layout.get("partCount") == -(-matching["size"] // part_bytes), "Invalid multipart layout.") entry["partBytes"] = part_bytes normalized.append(entry) return {"review": {"id": review}, "version": {"id": version}, "uploads": normalized, "finalizeUrl": prefix + "/finalize"} def upload_parts(client: Client, path: Path, state: dict, upload: dict, source: dict) -> None: """Sends a large file's parts in order. Verified parts are recorded in the manifest, so a resume sends only the rest; the whole file is hashed on the way and must still match its plan.""" part_bytes = upload["partBytes"] count = -(-source["size"] // part_bytes) done = state.setdefault("parts", {}).setdefault(upload["id"], []) digest = hashlib.sha256() with Path(source["path"]).open("rb") as stream: for number in range(1, count + 1): data = stream.read(part_bytes) require(len(data) == min(part_bytes, source["size"] - (number - 1) * part_bytes), "A source file changed. Keep the original bytes for resume, or start a new upload.") digest.update(data) if number in done: continue receipt = client.json("PUT", f"{upload['url']}/parts/{number}", part=data) require(receipt.get("uploaded") is True and receipt.get("id") == upload["id"] and receipt.get("partNumber") == number and receipt.get("sha256") == hashlib.sha256(data).hexdigest(), "The API returned an invalid part receipt.") done.append(number) save(path, state) require(stream.read(1) == b"" and digest.hexdigest() == source["sha256"], "A source file changed. Keep the original bytes for resume, or start a new upload.") def run_upload(client: Client, path: Path, state: dict) -> dict: require(not state.get("publicationConflict"), PUBLICATION_CONFLICT) files = state["files"] # Validate every source before any network work. The API verifies bytes again during upload. for file in files: require(Path(file["path"]).is_file(), "An upload source is no longer a regular file.") with Path(file["path"]).open("rb") as stream: verify_stream(stream, file) review_id = state.get("reviewId") if "plan" not in state: result = client.json("POST", state["requestPath"], payload=state["payload"], idempotency=state["idempotencyKey"]) state["plan"] = plan_shape(result, files, review_id) save(path, state) else: state["plan"] = plan_shape(state["plan"], files, review_id) plan = state["plan"] # A plan that asks for replies asks again at finalize: a new version of an existing sheet turns # them on as it lands (the plan alone sets them only for a new sheet). finalize_body = {"feedbackEnabled": True} if state["payload"].get("feedbackEnabled") is True else None def finalize() -> dict: try: return client.json("POST", plan["finalizeUrl"], payload=finalize_body) except ClientError as error: if error.status == 409 and error.code == "publication_conflict": state["publicationConflict"] = True save(path, state) raise ClientError(PUBLICATION_CONFLICT, error.status, error.code) from None raise # A previous finalize may have committed despite a lost response. Probe its idempotent endpoint # before PUT: ready versions correctly reject further uploads. try: result = finalize() except ClientError as error: if error.status != 409 or error.code != "uploads_incomplete": raise for upload in plan["uploads"]: if upload.get("uploaded") is True: continue source = next(file for file in files if file["path"] == upload["source"]) if "partBytes" in upload: upload_parts(client, path, state, upload, source) receipt = client.json("POST", upload["url"] + "/complete") else: receipt = client.json("PUT", upload["url"], source=Path(source["path"]), file=source) require(receipt.get("uploaded") is True and receipt.get("id") == upload["id"] and receipt.get("sha256") == source["sha256"], "The API returned an invalid upload receipt.") result = finalize() require(result.get("review", {}).get("id") == plan["review"]["id"] and result.get("version", {}).get("id") == plan["version"]["id"] and result.get("version", {}).get("status") == "ready", "The API did not confirm a ready version.") state["completed"] = True save(path, state) done = {"status": "ready", "reviewId": plan["review"]["id"], "versionId": plan["version"]["id"], "workspaceUrl": client.origin + "/app", "manifest": str(path)} if type(result["review"].get("feedbackEnabled")) is bool: done["feedbackEnabled"] = result["review"]["feedbackEnabled"] if isinstance(result.get("usage"), dict): done["usage"] = result["usage"] # The plan's counted limits after this version (plan-usage.ts). stored = sum(1 for upload in plan["uploads"] if upload.get("uploaded") is True) if stored: done["alreadyStored"] = stored # Files the workspace already stored: not sent again. return done def load_state(path: Path, origin: str) -> dict: require(path.is_file() and not path.is_symlink() and path.stat().st_size <= 65536, "Upload manifest is missing or too large.") state = decode(path.read_bytes()) require(state.get("format") == 1 and state.get("origin") == origin, "Upload manifest origin or format mismatch.") require(isinstance(state.get("idempotencyKey"), str) and re.fullmatch(r"[A-Za-z0-9_.:-]{8,100}", state["idempotencyKey"]) is not None, "Invalid manifest idempotency key.") files = state.get("files") require(isinstance(files, list) and 1 <= len(files) <= 50, "Invalid manifest files.") for file in files: require(isinstance(file, dict) and isinstance(file.get("path"), str) and Path(file["path"]).is_absolute() and isinstance(file.get("name"), str) and 1 <= len(file["name"]) <= 180 and file.get("type") in TYPES.values() and type(file.get("size")) is int and 1 <= file["size"] <= MAX_FILE and isinstance(file.get("sha256"), str) and HASH.fullmatch(file["sha256"]) is not None, "Invalid manifest file metadata.") parts = state.get("parts", {}) require(isinstance(parts, dict) and all(isinstance(key, str) and ID.fullmatch(key) is not None and isinstance(value, list) and all(type(n) is int and 1 <= n <= 10000 for n in value) for key, value in parts.items()), "Invalid manifest part record.") review_id = state.get("reviewId") require(review_id is None or isinstance(review_id, str) and ID.fullmatch(review_id) is not None, "Invalid manifest review ID.") expected_path = f"/api/v1/reviews/{review_id}/versions" if review_id else "/api/v1/reviews" require(state.get("requestPath") == expected_path and isinstance(state.get("payload"), dict), "Invalid manifest request.") expected_files = [{key: file[key] for key in ("name", "type", "size", "sha256")} | ({"interactive": True} if file.get("interactive") is True else {}) for file in files] require(state["payload"].get("files") == expected_files, "Manifest payload differs from source metadata.") return state def read_form(path: Path) -> dict: require(path.is_file() and not path.is_symlink() and path.stat().st_size <= 64 * 1024, "Form schema file is missing or too large.") value = decode(path.read_bytes()) require(isinstance(value.get("schemaId"), str) and isinstance(value.get("schemaVersion"), str) and isinstance(value.get("schema"), dict), "Form schema needs schemaId, schemaVersion and an object schema.") return value def review_revision(client: Client, review_id: str) -> tuple[int, str]: require(ID.fullmatch(review_id) is not None, "Invalid review ID.") detail = client.json("GET", f"/api/v1/reviews/{review_id}", limit=MAX_RESPONSE) review = detail.get("review") require(isinstance(review, dict) and type(review.get("revision")) is int, "The API returned an invalid review.") capability = "shared" if isinstance(review.get("share"), dict) and review["share"].get("on") is True else "private" return review["revision"], capability def run_share(client: Client, review_id: str, days: int | None) -> dict: revision, _ = review_revision(client, review_id) payload: dict = {"expectedRevision": revision} if days is not None: payload["expiresInDays"] = days result = client.json("POST", f"/api/v1/reviews/{review_id}/share", payload=payload, idempotency=uuid.uuid4().hex) url = result.get("shareUrl") require(isinstance(url, str) and url.startswith(client.origin + "/r/"), "The API returned an invalid share link.") share = (result.get("review") or {}).get("share", {}) return {"status": "shared", "shareUrl": url, "shareExpiresAt": share.get("expiresAt")} def run_unshare(client: Client, review_id: str) -> dict: revision, _ = review_revision(client, review_id) result = client.json("DELETE", f"/api/v1/reviews/{review_id}/share", payload={"expectedRevision": revision}, idempotency=uuid.uuid4().hex) require(isinstance(result.get("review"), dict), "The API returned an invalid response.") return {"status": "unshared", "reviewId": review_id} def safe_manifest_item(value: dict) -> bool: # Every kind the account manifest has carried; files are the only ones this client downloads. if value.get("kind") not in ("reviews", "versions", "files", "feedback", "categories", "forms", "workspaces", "styles", "product_feedback"): return False return all(isinstance(key, str) for key in value) def run_export_all(client: Client, directory: Path, manifest_only: bool) -> dict: items: list[dict] = [] cursor = "" for _ in range(1000): # Bounded paging; a pathological cursor cannot loop forever. path = "/api/v1/account/export" + (f"?cursor={urllib.parse.quote(cursor, safe='')}" if cursor else "") page = client.json("GET", path, limit=MAX_RESPONSE) require(page.get("format") == "tacksheet-account-manifest-v1" and isinstance(page.get("items"), list), "The API returned an invalid account manifest.") for item in page["items"]: require(isinstance(item, dict) and safe_manifest_item(item), "The API returned an invalid manifest item.") items.append(item) next_cursor = page.get("nextCursor") if next_cursor is None: break require(isinstance(next_cursor, str) and len(next_cursor) <= 80, "The API returned an invalid manifest cursor.") cursor = next_cursor require(not directory.exists() or (directory.is_dir() and not any(directory.iterdir())), "Export destination exists and is not empty.") directory.mkdir(parents=True, exist_ok=True, mode=0o700) require(not directory.is_symlink(), "Export destination symlinks are refused.") manifest = {"format": "tacksheet-account-manifest-v1", "exportedAt": int(time.time()), "items": items} (directory / "manifest.json").write_bytes(encode(manifest)) downloaded = 0 total = 0 if not manifest_only: for item in items: if item.get("kind") != "files": continue url = item.get("downloadUrl") name = item.get("name") require(isinstance(url, str) and url.startswith(client.origin) and isinstance(name, str) and ID.fullmatch(str(item.get("version_id", ""))) is not None and ID.fullmatch(str(item.get("review_id", ""))) is not None and type(item.get("size")) is int and 1 <= item["size"] <= MAX_FILE and isinstance(item.get("sha256"), str) and HASH.fullmatch(item["sha256"]) is not None, "The API returned an invalid file entry.") path = url[len(client.origin):] require(path.startswith("/api/v1/reviews/") and len(name) <= 160 and "/" not in name and ".." not in name, "The API returned an invalid file entry.") total += item["size"] require(total <= MAX_EXPORT_BYTES and downloaded < MAX_EXPORT_FILES, "Export exceeds the client limits.") target = directory / "files" / str(item["review_id"]) / str(item["version_id"]) / name target.parent.mkdir(parents=True, exist_ok=True) require(not target.exists(), "Export file collision.") # Streamed to a temporary name; only verified bytes take the file's name. partial = target.with_name(target.name + ".partial") try: digest = client.download(path, partial, item["size"]) require(partial.stat().st_size == item["size"] and digest == item["sha256"], "A downloaded export file failed integrity verification.") os.replace(partial, target) finally: partial.unlink(missing_ok=True) downloaded += 1 return {"status": "exported", "directory": str(directory), "items": len(items), "files": downloaded, "bytes": total} def main(argv: list[str] | None = None) -> int: parser = SafeArgumentParser(description=__doc__) parser.add_argument("--origin", default=DEFAULT_ORIGIN) parser.add_argument("--allow-local", action="store_true", help="Permit a localhost development API.") sub = parser.add_subparsers(dest="command", required=True) publish = sub.add_parser("publish", help="Create a private review.") publish.add_argument("files", nargs="+") publish.add_argument("--title", required=True) publish.add_argument("--retention", choices=("kept", "expiring"), default=None, help="Sheet retention; kept is the default and never expires.") publish.add_argument("--expires-in-days", type=int, choices=range(1, 366), metavar="1..365", help="Expire after this many days (implies --retention expiring).") publish.add_argument("--feedback", action="store_true") publish.add_argument("--form", type=Path, help="JSON file with a custom feedback form schema.") publish.add_argument("--interactive", action="store_true", help="Mark HTML files as interactive sheets.") publish.add_argument("--note", help="Owner note stored on the sheet.") publish.add_argument("--source", help="Free-text source label (for example 'postit').") publish.add_argument("--category", help="Category ID for the new sheet.") workspace = publish.add_mutually_exclusive_group() workspace.add_argument("--workspace", metavar="NAME", help="File the new sheet in the workspace with this exact name.") workspace.add_argument("--workspace-id", metavar="ID", help="File the new sheet in the workspace with this ID.") publish.add_argument("--state", type=Path) update = sub.add_parser("update", help="Append a retained version to an existing review.") update.add_argument("review_id") update.add_argument("files", nargs="+") update.add_argument("--interactive", action="store_true", help="Mark HTML files as interactive sheets.") update.add_argument("--form", type=Path, help="JSON file with a custom feedback form schema; turns replies on.") update.add_argument("--feedback", action="store_true", help="Turn free-text replies on with this version.") update.add_argument("--state", type=Path) resume = sub.add_parser("resume", help="Retry the same upload plan after interruption.") resume.add_argument("manifest", type=Path) share = sub.add_parser("share", help="Create or rotate the review's share link.") share.add_argument("review_id") share.add_argument("--expires-in-days", type=int, choices=range(1, 366), metavar="1..365") unshare = sub.add_parser("unshare", help="Revoke the review's share link.") unshare.add_argument("review_id") export = sub.add_parser("export", help="Export a review with metadata, content and feedback.") export.add_argument("review_id") export.add_argument("--format", choices=("json", "md", "html"), default="json") export.add_argument("--version", default="current", help="current, all, or a 1-based version number.") export.add_argument("--content", choices=("describe", "embed"), default="describe") export.add_argument("--output", required=True, type=Path) export_all = sub.add_parser("export-all", help="Export the whole account manifest and every file.") export_all.add_argument("--output", required=True, type=Path) export_all.add_argument("--manifest-only", action="store_true", help="Save the manifest without file bodies.") if tacksheet_sheet is not None: tacksheet_sheet.register(sub) # Adds the `sheet` subcommand; dispatch is below. args = parser.parse_args(argv) try: if args.command == "sheet": require(tacksheet_sheet is not None, "Sheet commands are unavailable in this copy.") return tacksheet_sheet.sheet_main(args) origin = origin_value(args.origin, args.allow_local) client = Client(origin, os.environ.get("TACKSHEET_API_KEY", "")) if args.command == "share": print(json.dumps(run_share(client, args.review_id, args.expires_in_days), sort_keys=True)) return 0 if args.command == "unshare": print(json.dumps(run_unshare(client, args.review_id), sort_keys=True)) return 0 if args.command == "export-all": print(json.dumps(run_export_all(client, args.output, args.manifest_only), sort_keys=True)) return 0 if args.command == "export": require(ID.fullmatch(args.review_id) is not None, "Invalid review ID.") require(not args.output.exists() and not args.output.is_symlink(), "Export destination already exists.") version = args.version require(version in ("current", "all") or (version.isdigit() and 1 <= int(version) <= 250), "Invalid version selection.") query = urllib.parse.urlencode({"format": args.format, "version": version, "content": args.content}) data = client.request("GET", f"/api/v1/reviews/{args.review_id}/export?{query}", limit=MAX_RESPONSE) descriptor = os.open(args.output, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) with os.fdopen(descriptor, "wb") as stream: stream.write(data) print(json.dumps({"status": "exported", "file": str(args.output), "bytes": len(data)})) return 0 path = safe_state_path(args.manifest if args.command == "resume" else args.state or Path.home() / ".tacksheet" / "uploads" / (uuid.uuid4().hex + ".json")) with locked(path): if args.command == "resume": state = load_state(path, origin) if state.get("completed") is True: plan = plan_shape(state.get("plan", {}), state["files"], state.get("reviewId")) print(json.dumps({"status": "already_completed", "reviewId": plan["review"]["id"], "versionId": plan["version"]["id"]})) return 0 else: require(not path.exists(), "Upload manifest already exists. Use resume or choose a different --state.") require(1 <= len(args.files) <= 50, "Upload 1 to 50 files.") # --interactive marks HTML files only; a mixed upload keeps other types plain. files = [describe_file(value, interactive=(args.interactive and TYPES.get(Path(value).suffix.lower()) in HTML_TYPES)) for value in args.files] require(len({file["path"] for file in files}) == len(files), "The same source file was supplied twice.") if args.interactive: require(any(file.get("interactive") is True for file in files), "--interactive needs at least one HTML file.") payload = {"files": [{key: file[key] for key in ("name", "type", "size", "sha256")} | ({"interactive": True} if file.get("interactive") is True else {}) for file in files]} review_id = args.review_id if args.command == "update" else None if review_id: require(ID.fullmatch(review_id) is not None, "Invalid review ID.") else: require(1 <= len(args.title.strip()) <= 160, "Titles must contain 1 to 160 characters.") require(not (args.retention == "kept" and args.expires_in_days is not None), "--retention kept and --expires-in-days conflict.") require(not (args.retention == "expiring" and args.expires_in_days is None), "Expiring retention needs --expires-in-days.") retention = args.retention or ("expiring" if args.expires_in_days is not None else "kept") payload.update(title=args.title.strip(), retention=retention, feedbackEnabled=args.feedback) if args.expires_in_days is not None: payload["expiresInDays"] = args.expires_in_days if args.note is not None: require(len(args.note) <= 2000, "Notes must contain at most 2000 characters.") payload["note"] = args.note if args.source is not None: require(len(args.source) <= 200, "Source labels must contain at most 200 characters.") payload["source"] = args.source if args.category is not None: require(ID.fullmatch(args.category) is not None, "Invalid category ID.") payload["categoryId"] = args.category # Workspaces group sheets by project; they are not the account. Unknown names are refused by the server. if args.workspace is not None: require(1 <= len(args.workspace.strip()) <= 60 and not any(ord(c) < 32 for c in args.workspace), "Workspace names contain 1 to 60 characters.") payload["workspace"] = {"name": args.workspace.strip()} if args.workspace_id is not None: require(ID.fullmatch(args.workspace_id) is not None, "Invalid workspace ID.") payload["workspace"] = {"id": args.workspace_id} if args.form is not None: payload["feedback"] = {"schema": read_form(args.form)} if review_id and (args.form is not None or args.feedback): payload["feedbackEnabled"] = True # Sent again at finalize, where a new version turns replies on. state = {"format": 1, "origin": origin, "idempotencyKey": uuid.uuid4().hex, "files": files, "requestPath": f"/api/v1/reviews/{review_id}/versions" if review_id else "/api/v1/reviews", "payload": payload} if review_id: state["reviewId"] = review_id save(path, state) # Before the first request: an ambiguous plan response cannot create a duplicate. print("Upload manifest: " + str(path), file=sys.stderr) print(json.dumps(run_upload(client, path, state), sort_keys=True)) return 0 except ClientError as exc: print("Tacksheet: " + str(exc), file=sys.stderr) except (OSError, ValueError, KeyError, TypeError, AttributeError): print("Tacksheet: Local file or manifest operation failed. No credentials were printed; inspect the source files and retry with the saved manifest.", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())