73 lines
5.0 KiB
Python
73 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""One-off native plan setup and read-only diagnostics; never prints credentials."""
|
|
from pathlib import Path
|
|
import datetime,json,os,re,subprocess,time
|
|
from zoneinfo import ZoneInfo
|
|
BASE=Path(__file__).resolve().parent
|
|
ROOT=BASE.parents[3]
|
|
CONF=Path('/home/wow_admin/wow_cata_server/server/debug/etc/worldserver.conf')
|
|
TZ=ZoneInfo('Europe/Berlin')
|
|
DATE=20261102
|
|
FORMATION=6
|
|
ACTIVITY=7
|
|
RAID=[159,160,161,162,163,164,165,166,167,176]
|
|
MEET={'alliance':'the_gilded_rose','horde':'poi_npc_287214'}
|
|
def now():return datetime.datetime.now(TZ)
|
|
def query(kind,sql):
|
|
if not re.match(r'^\s*(SELECT|SHOW)\b',sql,re.I):raise ValueError('Read-only database helper')
|
|
conf=CONF.read_text();value=re.search(r'^'+kind+r'DatabaseInfo\s*=\s*"([^"]+)"',conf,re.M).group(1)
|
|
h,port,u,password,db=value.split(';');env=os.environ.copy();env['MYSQL_PWD']=password
|
|
r=subprocess.run(['mysql','--protocol=TCP','--host='+h,'--port='+port,'--user='+u,'--database='+db,'--batch','--default-character-set=utf8mb4','--connect-timeout=10'],input=sql,text=True,capture_output=True,env=env,timeout=25)
|
|
if r.returncode:raise RuntimeError(r.stderr.replace(password,'[redacted]'))
|
|
lines=r.stdout.splitlines()
|
|
return [dict(zip(lines[0].split('\t'),line.split('\t'))) for line in lines[1:]] if lines else []
|
|
def command(text):
|
|
subprocess.run(['tmux','send-keys','-t','vcore-worldserver','-l',text],check=True,timeout=5)
|
|
subprocess.run(['tmux','send-keys','-t','vcore-worldserver','Enter'],check=True,timeout=5)
|
|
def console(lines=2000):
|
|
return subprocess.check_output(['tmux','capture-pane','-p','-J','-t','vcore-worldserver','-S',str(-lines)],text=True,timeout=10)
|
|
def roster():
|
|
rows=query('Character',"SELECT c.guid,c.name,c.race,c.level,c.online,c.map,c.instance_id,c.position_x,c.position_y,c.position_z,HEX(COALESCE(g.profile_lua,'')) profile FROM characters c LEFT JOIN vcore_adventure_generated_bot g ON g.character_guid=c.guid WHERE c.online=1 AND (c.guid=26 OR g.state='ready') ORDER BY c.guid;")
|
|
for r in rows:
|
|
for k in ['guid','race','level','online','map','instance_id']:r[k]=int(r[k])
|
|
r['faction']='alliance' if r['race'] in [1,3,4,7,11,22] else 'horde'
|
|
profile=bytes.fromhex(r.pop('profile')).decode(errors='replace')
|
|
role=re.search(r'\brole\s*=\s*["\'](tank|healer|damage)',profile)
|
|
r['role']=role[1] if role else ('tank' if r['guid']==26 else 'unknown')
|
|
return rows
|
|
def plans():return query('World',f'SELECT character_guid,hour_slot,intent,location_key FROM vcore_adventure_day WHERE local_date={DATE} AND hour_slot IN({FORMATION},{ACTIVITY}) ORDER BY character_guid,hour_slot;')
|
|
def apply():
|
|
rows=roster();p=BASE/'before-plans.json'
|
|
if not p.exists():p.write_text(json.dumps(plans(),indent=2))
|
|
current={(int(r['character_guid']),int(r['hour_slot'])):r for r in plans()}
|
|
pending=[]
|
|
for r in rows:
|
|
g=r['guid']
|
|
if r['level']<15:continue
|
|
a=current.get((g,ACTIVITY),{});f=current.get((g,FORMATION),{})
|
|
valid=a.get('intent') in ['raid','instance'] and f.get('intent')=='groupformation' and f.get('location_key')==MEET[r['faction']]
|
|
if g in RAID:valid=valid and a.get('location_key')=='instance_visit_at_6354_d0' and a.get('intent')=='raid'
|
|
if valid:continue
|
|
pending.append(g)
|
|
target='raid instance_visit_at_6354_d0' if g in RAID else 'instance-or-raid auto'
|
|
command(f'clientless adventure editplan {g} {ACTIVITY} {target} both 15 85 apply')
|
|
command(f'clientless adventure editplan {g} {FORMATION} groupformation {MEET[r["faction"]]} {r["faction"]} 15 85 apply')
|
|
time.sleep(2 if pending else .1)
|
|
current=plans();lookup={(int(r['character_guid']),int(r['hour_slot'])):r for r in current}
|
|
eligible=[];missing=[]
|
|
for r in rows:
|
|
if r['level']<15:continue
|
|
g=r['guid'];a=lookup.get((g,ACTIVITY),{});f=lookup.get((g,FORMATION),{})
|
|
good=f.get('intent')=='groupformation' and f.get('location_key')==MEET[r['faction']] and a.get('intent') in ['raid','instance']
|
|
if g in RAID:good=good and a.get('location_key')=='instance_visit_at_6354_d0' and a.get('intent')=='raid'
|
|
(eligible if good else missing).append(g)
|
|
manifest={'updated_at':now().isoformat(),'date':DATE,'formation_hour':FORMATION,'activity_hour':ACTIVITY,'evaluate_at':'2026-11-02T08:00:00+01:00','raid_guids':RAID,'meeting_locations':MEET,'roster':rows,'eligible_guids':eligible,'missing_plan_guids':missing,'excluded_under15':[r['guid'] for r in rows if r['level']<15],'plans':current}
|
|
(BASE/'manifest.json').write_text(json.dumps(manifest,ensure_ascii=False,indent=2))
|
|
(BASE/('plan-apply-'+now().strftime('%H%M%S')+'.log')).write_text(console())
|
|
print(json.dumps({'online':len(rows),'configured':len(eligible),'missing':missing,'newly_applied':pending,'excluded_under15':manifest['excluded_under15'],'raid_online':sorted(set(RAID)&{r['guid'] for r in rows})}))
|
|
return manifest
|
|
if __name__=='__main__':
|
|
import sys
|
|
if sys.argv[1:] == ['apply']:apply()
|
|
elif sys.argv[1:]==['roster']:print(json.dumps(roster(),ensure_ascii=False,indent=2))
|