#!/usr/bin/env python3 """Relay typed Classic commands and current native observations; no Lua execution.""" import argparse import json import os import pathlib import re import subprocess import time import urllib.request RACES={1:'Human',2:'Orc',3:'Dwarf',4:'Night Elf',5:'Undead',6:'Tauren',7:'Gnome',8:'Troll'} CLASSES={1:'Warrior',2:'Paladin',3:'Hunter',4:'Rogue',5:'Priest',7:'Shaman',8:'Mage',9:'Warlock',11:'Druid'} def validate(command): if not isinstance(command,dict) or command.get('version')!=1 or command.get('backend')!='cmangos-classic': raise ValueError('Not a typed Classic command') if command.get('kind') not in {'party','leave','recruit','intent','say'}:raise ValueError('Unsupported command kind') for key in ['guid','accountId','peerGuid','peerAccountId','expiresAt']: if type(command.get(key)) is not int or command[key]<0:raise ValueError('Invalid '+key) if not command['guid'] or not command['accountId'] or not re.fullmatch('[A-Za-z]{2,12}',command.get('name','')):raise ValueError('Invalid actor identity') if command['peerGuid'] and (not command['peerAccountId'] or not re.fullmatch('[A-Za-z]{2,12}',command.get('peerName',''))):raise ValueError('Invalid peer identity') if command['kind'] in {'party','recruit','say'} and not command['peerGuid']:raise ValueError('Peer identity required') if not isinstance(command.get('argument'),str) or len(command['argument'].encode())>1000:raise ValueError('Invalid argument') if command['expiresAt']time.time()+120:raise ValueError('Expired or invalid deadline') if command['kind']=='intent' and command['argument'] not in {'quest','explore','rest','trade','socialize'}:raise ValueError('Unsupported intention') return command def main(root,base,env_file): credentials=json.loads((root/'database-credentials.json').read_text()) values={} for line in env_file.read_text().splitlines(): if '=' in line and not line.lstrip().startswith('#'): key,value=line.split('=',1);values[key.strip()]=value.strip().strip('\"\'') token=values['ALE_BRIDGE_TOKEN'] env=dict(os.environ,MYSQL_PWD=credentials['root']) def mysql(sql): result=subprocess.run(['docker','exec','-i','-e','MYSQL_PWD','azerothai-classic-db','mysql','-uroot','-N','-B','--raw','classiccharacters'],input=sql,text=True,capture_output=True,env=env,timeout=10) if result.returncode:raise RuntimeError('Classic database query failed') return [line.split('\t') for line in result.stdout.strip().splitlines()] def quote(text):return "CONVERT(UNHEX('"+str(text).encode().hex()+"') USING utf8mb4)" def request(path,payload=None): body=json.dumps(payload or {}).encode() req=urllib.request.Request(base.rstrip('/')+path,data=body,headers={'Authorization':'Bearer '+token,'Content-Type':'application/json'}) with urllib.request.urlopen(req,timeout=10) as response:return response.headers,response.read().decode() def complete(id,ok,result):request('/api/bridge/ale/result',{'id':id,'ok':ok,'result':result}) last_snapshot=0;last_cleanup=0 while True: try: now=int(time.time()) if now-last_snapshot>=5: rows=mysql('SELECT HEX(payload) FROM azerothai_snapshots WHERE at>=UNIX_TIMESTAMP()-15 ORDER BY guid LIMIT 51') if len(rows)>50:raise RuntimeError('Classic snapshot population exceeds 50') payloads=[] for row in rows: item=json.loads(bytes.fromhex(row[0])) item['raceName']=RACES.get(item.get('race'),'Unknown') item['className']=CLASSES.get(item.get('class'),'Unknown') payloads.append(item) roster={'roster':True,'observedAt':now,'characterGuids':[p['characterGuid'] for p in payloads]} # Rotate eligible nearby pairs; normal app cooldowns bound conversations. pairs=[] available=lambda p:not any(p.get(k) for k in ['humanControlled','isDead','isGhost','inCombat','taxi','casting']) for p in payloads: if not available(p):continue for q in payloads: if p['characterGuid']>=q['characterGuid'] or not available(q):continue if p['team']==q['team'] and p['mapId']==q['mapId'] and (p['x']-q['x'])**2+(p['y']-q['y'])**2<=625:pairs.append((p,q)) if pairs: a,b=pairs[(now//10)%len(pairs)];roster.update(socialA=a['name'],socialB=b['name'],zoneName=a.get('zoneName','')) request('/api/bridge/ale/snapshots',{'payloads':payloads+[roster]});last_snapshot=now headers,body=request('/api/bridge/ale/poll') id=headers.get('X-AzerothAI-Command') if id: if not re.fullmatch('[0-9a-fA-F-]{36}',id):raise ValueError('Invalid command ID') try:c=validate(json.loads(body)) except (ValueError,TypeError) as error:complete(id,False,str(error)) else: mysql(f"INSERT IGNORE INTO azerothai_commands(id,guid,account_id,name,kind,argument,peer_guid,peer_account_id,peer_name,created_at,expires_at) VALUES ({quote(id)},{c['guid']},{c['accountId']},{quote(c['name'])},{quote(c['kind'])},{quote(c['argument'])},{c['peerGuid']},{c['peerAccountId']},{quote(c['peerName'])},{c['expiresAt']-90},{c['expiresAt']})") for id,status,result in mysql("SELECT id,status,HEX(COALESCE(result,'')) FROM azerothai_commands WHERE delivered=0 AND status IN ('completed','failed') LIMIT 16"): complete(id,status=='completed',bytes.fromhex(result).decode());mysql('UPDATE azerothai_commands SET delivered=1 WHERE id='+quote(id)) mysql("UPDATE azerothai_commands SET status='failed',result='Expired before native execution' WHERE status='pending' AND expires_at3600: mysql("DELETE FROM azerothai_commands WHERE delivered=1 AND created_at