#!/usr/bin/env python3 """Reference client for the Ponyglot Cloud API v1. Stands in for the connectors until they exist: pushes JSON fixtures shaped like django CMS pages, parler models and Wagtail pages, and shows status. Standard library only, so it runs anywhere with Python 3.10+. export PONYGLOT_API_KEY=pg_… export PONYGLOT_API_URL=http://api.localhost:8000/v1 # default python tools/refclient/refclient.py handshake --languages en de fr it python tools/refclient/refclient.py push tools/refclient/fixtures/djangocms_pricing.json python tools/refclient/refclient.py status djangocms:page:42 python tools/refclient/refclient.py translate --wait # delta job for stale + missing python tools/refclient/refclient.py pull --ack --approve # drafts in, editor approves """ import argparse import json import os import sys import time import urllib.error import urllib.request SYMBOLS = {"ok": "✓", "stale": "stale", "review": "review", "missing": "·", "attention": "!"} class Client: def __init__(self, base_url, key): if not base_url.startswith(("http://", "https://")): sys.exit(f"Not an http(s) URL: {base_url}") self.base_url = base_url.rstrip("/") self.key = key def request(self, method, path, payload=None): data = json.dumps(payload).encode() if payload is not None else None request = urllib.request.Request( # noqa: S310 (scheme checked in __init__) self.base_url + path, data=data, method=method, headers={"Authorization": f"Bearer {self.key}", "Content-Type": "application/json"}, ) try: with urllib.request.urlopen(request, timeout=60) as response: # noqa: S310 body = response.read() except urllib.error.HTTPError as error: sys.exit(f"HTTP {error.code} {method} {path}: {error.read().decode()[:2000]}") return json.loads(body) if body else None def print_json(data): print(json.dumps(data, indent=2, ensure_ascii=False)) def print_matrix(status): languages = [language["code"] for language in status["languages"]] print(f"{status['label'] or status['external_key']} {status['path']}") print(f"{status['segments_needing_work']} of {status['segments_total']} segments need work\n") width = max([len(s["key"]) for s in status["segments"]] + [7]) print("segment".ljust(width), *(code.upper().ljust(7) for code in languages)) for segment in status["segments"]: cells = (SYMBOLS[segment["states"][code]].ljust(7) for code in languages) print(segment["key"].ljust(width), *cells) def print_job(job): print( f"job {job['id']} {job['type']} [{', '.join(job['languages'])}]: {job['status']} · " f"{job['done_segments']}/{job['estimated_segments']} segments · " f"{job['characters']} characters · {job['failed_segments']} failed" ) for error in job["errors"]: print(f" ! {error['language'] or '*'}: {error['message']}") def wait_for(client, job_id, timeout=600): deadline = time.monotonic() + timeout while True: job = client.request("GET", f"/jobs/{job_id}") if job["status"] not in ("queued", "running") or time.monotonic() > deadline: return job time.sleep(1) def pull(client, ack, approve): """Fetch all ready results (like a connector writing drafts), optionally ack/approve.""" results, cursor = [], None while True: page = client.request("GET", "/results" + (f"?cursor={cursor}" if cursor else "")) results += page["results"] cursor = page["next_cursor"] if not cursor: break for result in results: print(f"{result['id']:>6} {result['language']} {result['unit']} {result['key']}") print(f" {result['text'][:100]}") ids = [result["id"] for result in results] if ids and ack: print(client.request("POST", "/results/ack", {"ids": ids})) if ids and approve: reviews = [{"id": i, "outcome": "approved", "reviewer": "refclient"} for i in ids] print(client.request("POST", "/reviews", {"reviews": reviews})) print(f"{len(ids)} result(s)") def main(argv=None): parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) parser.add_argument( "--url", default=os.environ.get("PONYGLOT_API_URL", "http://api.localhost:8000/v1") ) parser.add_argument("--key", default=os.environ.get("PONYGLOT_API_KEY")) commands = parser.add_subparsers(dest="command", required=True) commands.add_parser("site", help="Show the site this key belongs to.") handshake = commands.add_parser("handshake", help="Report connector version and languages.") handshake.add_argument("--version", default="refclient") handshake.add_argument("--source", default="en") handshake.add_argument("--languages", nargs="*", default=["en", "de", "fr", "it"]) handshake.add_argument("--adapters", nargs="*", default=["djangocms", "parler", "wagtail"]) push = commands.add_parser("push", help="Push unit snapshot(s) from JSON files.") push.add_argument("files", nargs="+") delete = commands.add_parser("delete", help="Delete a unit.") delete.add_argument("external_key") status = commands.add_parser("status", help="Site status, or the matrix of one unit.") status.add_argument("external_key", nargs="?") status.add_argument("--json", action="store_true") translate = commands.add_parser("translate", help="Request translation (a job).") translate.add_argument("--units", nargs="*", help="translate these units (default: delta)") translate.add_argument("--languages", nargs="*") translate.add_argument("--force", action="store_true") translate.add_argument("--wait", action="store_true", help="wait until the job is done") job = commands.add_parser("job", help="Show a job (or the recent jobs).") job.add_argument("job_id", nargs="?", type=int) job.add_argument("--wait", action="store_true") pull_parser = commands.add_parser("pull", help="Fetch ready translations.") pull_parser.add_argument("--ack", action="store_true", help="acknowledge (drafts written)") pull_parser.add_argument("--approve", action="store_true", help="report them approved") review = commands.add_parser("review", help="Report review outcomes for result ids.") review.add_argument("outcome", choices=["approved", "rejected"]) review.add_argument("ids", nargs="+", help="opaque result IDs, e.g. r_5012") args = parser.parse_args(argv) if not args.key: parser.error("set PONYGLOT_API_KEY or pass --key") client = Client(args.url, args.key) if args.command == "site": print_json(client.request("GET", "/site")) elif args.command == "handshake": print_json( client.request( "POST", "/handshake", { "connector_version": args.version, "adapters": args.adapters, "source_language": args.source, "languages": args.languages, }, ) ) elif args.command == "push": for path in args.files: with open(path, encoding="utf-8") as file: data = json.load(file) if "units" in data: print_json(client.request("PUT", "/units/batch", data)) else: print_json(client.request("PUT", "/units", data)) elif args.command == "delete": client.request("DELETE", f"/units/{args.external_key}") print(f"deleted {args.external_key}") elif args.command == "translate": payload = {"type": "translate" if args.units else "delta", "force": args.force} if args.units: payload["units"] = args.units if args.languages: payload["languages"] = args.languages created = client.request("POST", "/jobs", payload) print_job(wait_for(client, created["id"]) if args.wait else created) elif args.command == "job": if args.job_id is None: for recent in client.request("GET", "/jobs"): print_job(recent) else: job_data = ( wait_for(client, args.job_id) if args.wait else client.request("GET", f"/jobs/{args.job_id}") ) print_job(job_data) elif args.command == "pull": pull(client, args.ack, args.approve) elif args.command == "review": reviews = [{"id": i, "outcome": args.outcome, "reviewer": "refclient"} for i in args.ids] print(client.request("POST", "/reviews", {"reviews": reviews})) elif args.command == "status": if args.external_key is None: print_json(client.request("GET", "/status")) else: data = client.request("GET", f"/units/{args.external_key}/status") print_json(data) if args.json else print_matrix(data) if __name__ == "__main__": main()