"""Connect an explicitly paired project's Android runtime. Python 3 + Pillow + adb. No inbound listener, arbitrary shell commands, automatic installs or plugin activation. """ import argparse, base64, getpass, io, json, re, subprocess, time, urllib.request, urllib.error from PIL import Image parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--origin', required=True) parser.add_argument('--package', required=True, help='Game Android package name') parser.add_argument('--adb', default='adb') parser.add_argument('--serial', help='Optional when exactly one device is online') parser.add_argument('--scenes', help='Local JSON mapping scene names to bounded tap/back actions') args = parser.parse_args() if not re.fullmatch(r'[A-Za-z][A-Za-z0-9_]*(\.[A-Za-z0-9_]+)+', args.package): parser.error('Invalid package') origin = args.origin.rstrip('/') if not re.fullmatch(r'https://[A-Za-z0-9.-]+(?::\d+)?', origin): parser.error('Use the platform HTTPS origin without a path') def adb(*parts, timeout=15): return subprocess.run([args.adb, '-s', args.serial, *parts], check=True, capture_output=True, timeout=timeout).stdout if not args.serial: listing = subprocess.run([args.adb, 'devices'], check=True, capture_output=True, timeout=10).stdout.decode() devices = re.findall(r'^([^\s]+)\s+device$', listing, re.M) if len(devices) != 1: parser.error('Connect exactly one online Android device, or use --serial') args.serial = devices[0] scenes = {} if args.scenes: with open(args.scenes, encoding='utf-8') as f: scenes = json.load(f) if not isinstance(scenes, dict) or len(scenes) > 20: parser.error('At most 20 scenes') for name, scene in scenes.items(): if not isinstance(name, str) or not 1 <= len(name) <= 60 or not isinstance(scene, dict): parser.error('Invalid scene') steps = scene.get('steps', []) if not isinstance(steps, list) or not 1 <= len(steps) <= 12: parser.error('Each scene needs 1-12 steps') # Precondition is a visible Android activity, not a model-generated command. if not re.fullmatch(r'[A-Za-z0-9_.$/]+', scene.get('activity', '')): parser.error('Each scene needs an exact activity precondition') for step in steps: if step == {'back': True}: continue if set(step) != {'tap'} or not isinstance(step['tap'], list) or len(step['tap']) != 2 or any(type(x) not in (int, float) or not 0 <= x <= 1 for x in step['tap']): parser.error('Only normalized tap:[x,y] or back:true steps are allowed') def focused(): text = adb('shell', 'dumpsys', 'window', 'windows').decode('utf-8', 'replace') lines = [line.strip() for line in text.splitlines() if 'mCurrentFocus=' in line] value = '\n'.join(lines) if not re.search(r'\b' + re.escape(args.package) + r'/', value): raise RuntimeError('Selected game must be in the foreground; no input or capture performed') return value[:1000] def screenshot(): focused() raw = adb('exec-out', 'screencap', '-p') if len(raw) > 24000000: raise RuntimeError('Screenshot too large') focused() picture = Image.open(io.BytesIO(raw)); picture.thumbnail((800, 800)) # Palette reduction bounds uploads without sending the full resolution screen. out = io.BytesIO(); picture.convert('RGB').quantize(colors=128).save(out, format='PNG', optimize=True) while out.tell() > 60000 and max(picture.size) > 320: picture.thumbnail((max(320, int(picture.width * .8)), max(320, int(picture.height * .8)))) out = io.BytesIO(); picture.convert('RGB').quantize(colors=64).save(out, format='PNG', optimize=True) if out.tell() > 60000: raise RuntimeError('Screenshot exceeds image-context limit') return 'data:image/png;base64,' + base64.b64encode(out.getvalue()).decode(), picture.size class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *unused, **kwargs): raise RuntimeError('Redirect refused') opener = urllib.request.build_opener(NoRedirect) token = None def request(path, body): headers = {'Content-Type': 'application/json'} if token: headers['Authorization'] = 'Bearer ' + token req = urllib.request.Request(origin + '/api/v1/plugins/verification-bridge/' + path, json.dumps(body).encode(), headers) with opener.open(req, timeout=20) as response: return json.load(response) code = getpass.getpass('Project pairing code (not saved): ') reply = request('connect', {'code': code, 'capabilities': {'name': 'Android project connection', 'packageName': args.package, 'scenes': list(scenes), 'screenshots': True, 'logs': True}}) token = reply['token']; del code print('Project connected for up to 24 hours. Keep the game foreground. Ctrl+C disconnects this process.') pending = None try: while True: try: if pending: request('result', pending); pending = None for job in request('poll', {})['jobs']: result = {'ok': False, 'capturedAt': int(time.time()*1000), 'packageName': args.package} try: result['scene'] = focused()[:160] operation = job['operation'] if operation == 'status': result['text'] = 'Game foreground. Plugin version and function outcome require plugin logs.' elif operation == 'logs': pid = adb('shell', 'pidof', args.package).decode().strip().split()[0] if not pid.isdigit(): raise RuntimeError('Invalid process ID') result['text'] = adb('logcat', '-d', '--pid=' + pid, '-t', '150').decode('utf-8', 'replace')[-24000:] elif operation == 'screenshot': result['image'], _ = screenshot() elif operation == 'navigate': scene = scenes.get(job['args'].get('scene')) if not scene: raise RuntimeError('Scene is not locally configured') if scene['activity'] not in focused(): raise RuntimeError('Scene activity precondition failed') raw = adb('exec-out', 'screencap', '-p'); width, height = Image.open(io.BytesIO(raw)).size for step in scene['steps']: if job['expires_at'] <= time.time()*1000: raise RuntimeError('Navigation expired; partial input may have occurred') focused() if 'back' in step: adb('shell', 'input', 'keyevent', '4') else: adb('shell', 'input', 'tap', str(round(step['tap'][0]*(width-1))), str(round(step['tap'][1]*(height-1)))) time.sleep(0.6) result['text'] = 'Navigation inputs sent. Capture a new screenshot to verify destination; scene success is unverified.' else: raise RuntimeError('Unsupported operation') result['ok'] = True except Exception as error: result['error'] = str(error)[:1000] pending = {'commandId': job['id'], 'result': result} request('result', pending); pending = None except urllib.error.HTTPError as error: if error.code == 401: raise RuntimeError('Connection expired or revoked') from None if error.code == 409: pending = None print('Connection HTTP error:', error.code) except (OSError, TimeoutError): print('Connection interrupted; receipts will retry, commands will not replay.') time.sleep(3) except KeyboardInterrupt: print('\nStopped. The platform marks this project offline after 20 seconds; revoke it in Project connection.')