#!/usr/bin/env python3 """RCT2 arcade client. Python standard library only. Gameplay credentials never printed.""" import json, os, ssl, sys, time, uuid, urllib.request, urllib.error, urllib.parse, http.cookiejar from pathlib import Path FILE = Path(os.environ.get('ARCADE_SESSION_FILE', '.arcade-agent.json')) class ApiError(RuntimeError): def __init__(self, status, message): super().__init__(message); self.status = status class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): return None def save(state): temporary = FILE.with_name(FILE.name + '.tmp') fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) os.fchmod(fd, 0o600) with os.fdopen(fd, 'w') as f: json.dump(state, f) os.replace(temporary, FILE) def request(base, path, body=None, token=None, opener=None): headers = {'Content-Type': 'application/json'} if token: headers['Authorization'] = 'Bearer ' + token opener = opener or urllib.request.build_opener(NoRedirect()) for attempt in range(6): try: req = urllib.request.Request(base + path, data=None if body is None else json.dumps(body).encode(), headers=headers) with opener.open(req, timeout=40) as response: return json.load(response) except urllib.error.HTTPError as error: if error.code not in (429, 500, 502, 503, 504): raise ApiError(error.code, 'HTTP %s: %s' % (error.code, error.read().decode()[:500])) from None delay = error.headers.get('Retry-After', '') time.sleep(min(60, int(delay) if delay.isdigit() else 2 ** attempt)) except (urllib.error.URLError, TimeoutError, ssl.SSLError) as error: reason = getattr(error, 'reason', error) if isinstance(reason, ssl.SSLCertVerificationError): raise RuntimeError('TLS certificate verification failed (SSLCertVerificationError). Configure a trusted CA bundle for Python, e.g. SSL_CERT_FILE=/etc/ssl/cert.pem on macOS if that bundle exists. Do not disable certificate verification. Pending requests remain saved; use resume after fixing trust.') from None if attempt == 5: raise RuntimeError('Service unavailable (%s). Pending request is saved; use resume.' % type(reason).__name__) from None time.sleep(min(30, 2 ** attempt)) raise RuntimeError('Service unavailable. Pending request is saved; use resume.') def submission(state, name=None, publish=False): base, path, token = state['base'], '/runs/' + state['id'], state['token'] info = request(base, path + '?view=submission', token=token) if not isinstance(info, dict) or not info.get('submitPath') or 'eligible' not in info: raise RuntimeError('Leaderboard submission is unavailable from this host. The finished replay is preserved.') result = {'submitUrl': base + info['submitPath'], 'eligible': info['eligible']} prefs = state.get('submission', {}) if publish: chosen = name or prefs.get('name') if not chosen: raise RuntimeError('Provide a leaderboard name: python agent.py submit --name "Your Name"') result.update(request(base, path + '/publish', {'name': chosen, 'mode': prefs.get('mode', 'agent')}, token=token)) result['submitted'] = True return result def main(): if len(sys.argv) < 2: raise RuntimeError('Commands: init URL | attach (JSON stdin) | introduce JSON (or stdin) | designCoaster JSON (or stdin) | draft JSON (or stdin) | preview JSON (or stdin) | act JSON (or stdin) | resume | finish | submit [--name NAME] | state | inspect VIEW [ID or JSON bounds]') command = sys.argv[1] if command == 'designCoaster': command = 'jevDesign' if command == 'attach': if FILE.exists(): raise RuntimeError('Session file exists. Use state/resume, or select another ARCADE_SESSION_FILE.') state = json.loads(sys.stdin.read()) if not all(isinstance(state.get(k), str) and state[k] for k in ('base', 'id', 'token')): raise RuntimeError('Provide base, id and token as JSON on stdin.') state = {k: state[k] for k in ('base', 'id', 'token', 'submission') if k in state} prefs = state.get('submission', {}) if not isinstance(prefs, dict) or not isinstance(prefs.get('autoSubmit', False), bool) or (prefs.get('autoSubmit') and not isinstance(prefs.get('name'), str)): raise RuntimeError('Invalid submission preferences.') state['base'] = state['base'].rstrip('/') parsed = urllib.parse.urlsplit(state['base']) if parsed.scheme != 'https' and not (parsed.scheme == 'http' and parsed.hostname in ('localhost', '127.0.0.1')): raise RuntimeError('Use HTTPS, or localhost for development.') if parsed.path or parsed.query or parsed.fragment or parsed.username: raise RuntimeError('Use the API origin URL only.') result = request(state['base'], '/runs/' + state['id'] + '?view=brief', token=state['token']) state['revision'] = result['revision']; save(state) elif command == 'init': if FILE.exists(): raise RuntimeError('Session file exists. Use state/resume, or select another ARCADE_SESSION_FILE.') base = sys.argv[2].rstrip('/') parsed = urllib.parse.urlsplit(base) if parsed.scheme != 'https' and not (parsed.scheme == 'http' and parsed.hostname in ('localhost', '127.0.0.1')): raise RuntimeError('Use HTTPS, or localhost for development.') if parsed.path or parsed.query or parsed.fragment or parsed.username: raise RuntimeError('Use the API origin URL only.') jar = http.cookiejar.CookieJar(); opener = urllib.request.build_opener(NoRedirect(), urllib.request.HTTPCookieProcessor(jar)) request(base, '/arcade/session', {}, opener=opener) result = request(base, '/arcade/runs', {'requestId': str(uuid.uuid4())}, opener=opener) state = {'base': base, 'id': result['id'], 'token': result['token'], 'revision': result['revision']} save(state) else: state = json.loads(FILE.read_text()) base, path, token = state['base'], '/runs/' + state['id'], state['token'] if command in ('init', 'attach', 'state'): result = request(base, path + '?view=brief', token=token) if not state.get('pending'): state['revision'] = result['revision']; save(state) elif command == 'submit': name = None if len(sys.argv) > 2: if len(sys.argv) != 4 or sys.argv[2] != '--name': raise RuntimeError('Usage: submit [--name NAME]') name = sys.argv[3] result = submission(state, name, publish=True) elif command == 'introduce': identity = json.loads(sys.argv[2] if len(sys.argv) > 2 else sys.stdin.read()) result = request(base, path + '/introduce', identity, token=token) elif command in ('preview', 'draft'): if state.get('pending'): raise RuntimeError('Pending request exists. Run resume before previewing.') action = json.loads(sys.argv[2] if len(sys.argv) > 2 else sys.stdin.read()) result = request(base, path + '/preview', {'revision': state['revision'], 'action': action, **({'draft': True} if command == 'draft' else {})}, token=token) elif command == 'inspect': view = sys.argv[2] if len(sys.argv) > 2 else 'summary'; query = {'view': view} if len(sys.argv) > 3: query.update({'id': sys.argv[3]} if view == 'ride' else json.loads(sys.argv[3])) result = request(base, path + '/inspect?' + urllib.parse.urlencode(query), token=token) elif command in ('act', 'resume', 'finish', 'jevDesign'): if command in ('act', 'finish', 'jevDesign'): if state.get('pending'): raise RuntimeError('Pending request exists. Run resume before submitting another action.') state['pending'] = {'requestId': str(uuid.uuid4()), 'revision': state['revision']} state['pendingPath'] = command if command in ('finish', 'jevDesign') else 'step' if command == 'jevDesign': design = json.loads(sys.argv[2] if len(sys.argv) > 2 else sys.stdin.read()) state['pending']['design'] = {'points': design} if isinstance(design, list) else design if command == 'act': actions = json.loads(sys.argv[2] if len(sys.argv) > 2 else sys.stdin.read()) if isinstance(actions, dict): actions = [actions] if not isinstance(actions, list) or not 1 <= len(actions) <= 32: raise RuntimeError('Submit 1 to 32 actions per batch.') state['pending']['actions'] = actions save(state) if not state.get('pending'): raise RuntimeError('No pending request.') try: result = request(base, path + '/' + state.get('pendingPath', 'step') + '?wait=25&view=brief', state['pending'], token) except ApiError as error: if error.status == 400: state.pop('pending'); state.pop('pendingPath', None); save(state) raise if error.status != 409: raise current = request(base, path + '?view=brief', token=token) state.pop('pending'); state.pop('pendingPath', None); state['revision'] = current['revision']; save(state) print(json.dumps({'error': str(error), 'state': current, 'hint': 'Inspect the current park before choosing a new action.'}, separators=(',', ':'))); return while result['status'] not in ('done', 'failed'): result = request(base, path + '/operations/' + result['id'] + '?wait=25&view=brief', token=token) if result['status'] == 'done': finished = state.get('pendingPath') == 'finish' or bool(result['result'].get('replayPath')) state['revision'] = result['result']['revision']; state.pop('pending'); state.pop('pendingPath', None); save(state) if finished: try: info = submission(state) result['submission'] = info if info['eligible'] and state.get('submission', {}).get('autoSubmit'): result['submission'] = submission(state, publish=True) except (RuntimeError, ValueError, OSError) as error: result['submissionError'] = str(error) result['submissionHint'] = 'Run is finished. Retry publication with python agent.py submit --name "Your Name".' else: state.pop('pending'); state.pop('pendingPath', None); save(state) result['hint'] = 'No actions from this operation committed. State remains available; inspect before retrying.' else: raise RuntimeError('Unknown command') watch_path = result.get('watchPath') or (result.get('result') or {}).get('watchPath') if watch_path: result['watchUrl'] = base + watch_path print(json.dumps(result, separators=(',', ':'))) if __name__ == '__main__': try: main() except (RuntimeError, ValueError, KeyError, IndexError, OSError) as error: print(json.dumps({'error': str(error)})); sys.exit(1)