// PORT: AVA2-style channel switch (ENABLE_CHANNEL_CHANGE), client-only. Instead of a server-side warp it runs // the original paths in order: /logout (the server's 3 s / 10 s countdown, which also refuses while fighting), // the disconnect brings back introLogin.LoginWindow, which then connects to the chosen channel of serverInfo // with the account kept by networkModule.MainStream, and stream.isAutoSelect makes introSelect start the same // character. See docs/AVA2-CHANNEL.md. static const char c_szChannelScript[] = R"PY(# -*- coding: utf-8 -*- import app import chat import game import grp import introLogin import net import ui import weakref from mt_biolog import Action, Text, L from mt_wiki import B try: import serverInfo except ImportError: serverInfo=None g_window=None g_entry=None CHANNEL_FILE='channel.inf' # the file introLogin's server board reads LOGOUT_TIMEOUT=20. # countdown is 3 s, 10 s while fighting; nothing back means the server refused LOGIN_TIMEOUT=30. def Locate(addr,port): # (region, server, channel) of serverInfo whose game address is addr:port; channel 0 when none matches. regions=getattr(serverInfo,'REGION_DICT',{}) for region in sorted(regions): for server in sorted(regions[region]): for channel,data in sorted(regions[region][server].get('channel',{}).items()): if data.get('ip')==addr and data.get('tcp_port')==port:return region,server,channel try: f=open(CHANNEL_FILE,'r');words=f.read().split();f.close() server=int(words[0]);region=int(words[2]) if len(words)>2 else 0 if server in regions.get(region,{}):return region,server,0 except (IOError,OSError,ValueError,IndexError):pass for region in sorted(regions): for server in sorted(regions[region]):return region,server,0 return None def Channels(place): if not place:return [] region,server,current=place data=serverInfo.REGION_DICT[region][server] return [(channel,info) for channel,info in sorted(data.get('channel',{}).items())] class Switcher(object): # stage: None, 'logout' (waiting for the server to close the session), 'login' (connecting to the target). def __init__(self,send,now,listener=None): self.send=send;self.now=now;self.listener=listener self.stage=None;self.deadline=0;self.target=None;self.account=None;self.error=None;self.stream=None def Busy(self):return self.stage is not None def Start(self,stream,place,channel): if self.Busy():return False region,server,current=place info=serverInfo.REGION_DICT[region][server]['channel'].get(channel) if not info or channel==current:self.error='channel';self.Notify('error');return False if not getattr(stream,'id','') or not getattr(stream,'pwd',''):self.error='account';self.Notify('error');return False auth=getattr(serverInfo,'REGION_AUTH_SERVER_DICT',{}).get(region,{}).get(server) account=(auth['ip'],auth['port']) if auth else (stream.account_addr,stream.account_port) self.target=dict(region=region,server=server,channel=channel,ip=info['ip'],port=info['tcp_port'], account=account,label='%s, %s'%(serverInfo.REGION_DICT[region][server].get('name',''),info.get('name',''))) self.account=(stream.id,stream.pwd,stream.GetCharacterSlot());self.stream=stream self.error=None;self.stage='logout';self.deadline=self.now()+LOGOUT_TIMEOUT self.send('/logout');self.Notify('start') return True def Tick(self): if self.stage and self.now()>=self.deadline: self.error='timeout' if self.stage=='logout' else 'login';self.Clear();self.Notify('error') def Clear(self): if self.stream:self.stream.isAutoSelect=0 # never leave the original auto-select on for a manual login self.stage=None;self.target=None;self.account=None;self.stream=None def Notify(self,kind): if self.listener:self.listener(kind) def OnLoginWindow(self,window): # LoginWindow.Open has just reset the connect info from loginInfo.xml; point it at the target channel. if self.stage!='logout':return False t=self.target;user,pwd,slot=self.account;stream=window.stream stream.SetConnectInfo(t['ip'],t['port'],t['account'][0],t['account'][1]) stream.SetCharacterSlot(slot);stream.isAutoSelect=1 net.SetServerInfo(t['label'].strip()) try: f=open(CHANNEL_FILE,'w');f.write('%d %d %d'%(t['server'],t['channel'],t['region']));f.close() except (IOError,OSError):pass self.stage='login';self.deadline=self.now()+LOGIN_TIMEOUT window.Connect(user,pwd) return True def OnLoginFailure(self,error): if self.stage=='login':self.error='login';self.Clear();self.Notify('error') def OnGame(self,stream): if self.stage=='login': channel=self.target['channel'];self.Clear();self.Notify('done') return channel return 0 ERRORS={'channel':'无法换到该频道','account':'没有保存的登录信息，请重新登录后再换线', 'timeout':'服务器没有退出（战斗中或不允许退出），换线已取消','login':'登录目标频道失败'} class ChannelWindow(ui.Window): def __init__(self,switcher,stream): ui.Window.__init__(self);self.SetWindowName('ChannelWindow');self.AddFlag('float');self.AddFlag('movable') self.switcher=switcher;self.stream=stream;self.place=None;self.channelButtons=[] self.SetSize(332,260) self.heading=Text(self,18,18,'换线',0xffffd68b) self.closeButton=Action(self,256,6,60,'关闭',self.Hide) self.current=Text(self,18,52,'',0xffc3b7a1) self.status=Text(self,18,232,'',0xffc3b7a1) self.Hide() def Buttons(self):return [self.closeButton]+self.channelButtons def Build(self): for b in self.channelButtons:b.Hide();b.event=None self.place=Locate(self.stream.addr,self.stream.port) channels=Channels(self.place) self.channelButtons=[Action(self,18+(i%2)*152,82+(i//2)*52,144,'',lambda channel=channel:self.Switch(channel)) for i,(channel,info) in enumerate(channels)] self.SetSize(332,max(160,82+(len(channels)+1)//2*52+40));self.status.SetPosition(18,self.GetHeight()-28) def Open(self):self.Build();self.Refresh();self.SetCenterPosition();self.Show();self.SetTop() def Switch(self,channel): if self.place:self.switcher.Start(self.stream,self.place,channel) self.Refresh() def Refresh(self): current=self.place[2] if self.place else 0 channels=Channels(self.place);busy=self.switcher.Busy() name=dict(channels).get(current,{}).get('name') self.current.SetText(L('当前频道：'+(name if name else '未知'))) for button,(channel,info) in zip(self.channelButtons,channels): button.SetText(L(B(u'【%s】'%info.get('name','') if channel==current else info.get('name','')))) button.SetEnabled(not busy and channel!=current) self.status.SetText(L(self.StatusText())) def StatusText(self): s=self.switcher if s.stage=='logout':return '正在退出，稍后自动登录 %s'%s.target['label'].split(', ')[-1] if s.stage=='login':return '正在登录 %s'%s.target['label'].split(', ')[-1] if s.error:return '！'+ERRORS[s.error] if not self.place:return '！没有可用的频道列表' return '退出后自动登录所选频道，并进入同一角色' def OnRender(self): x,y=self.GetGlobalPosition();grp.SetColor(0xf51b1916);grp.RenderBar(x,y,self.GetWidth(),self.GetHeight());grp.SetColor(0xff9b7b48);grp.RenderBox(x,y,self.GetWidth(),self.GetHeight()) def OnPressEscapeKey(self):self.Hide();return True def Destroy(self): self.Hide() for b in self.Buttons():b.event=None class Entry(Action): def __init__(self,window): Action.__init__(self,None,0,0,64,'换线',Toggle) self.minimap=weakref.proxy(window.interface.wndMiniMap);self.SetWindowName('ChannelEntry');self.OnUpdate() def OnUpdate(self): x,y=self.minimap.GetGlobalPosition();self.SetPosition(max(4,x-68),y+304) def OnSwitch(kind): if kind=='error':chat.AppendChat(chat.CHAT_TYPE_INFO,L(ERRORS[g_switcher.error])) if g_window:g_window.Refresh() g_switcher=Switcher(net.SendChatPacket,app.GetTime,OnSwitch) def Toggle(): if g_window: if g_window.IsShow():g_window.Hide() else:g_window.Open() _open=game.GameWindow.Open _close=game.GameWindow.Close _update=game.GameWindow.OnUpdate _login_open=introLogin.LoginWindow.Open _login_update=introLogin.LoginWindow.OnUpdate _login_failure=introLogin.LoginWindow.OnLoginFailure _connect_failure=introLogin.LoginWindow.OnConnectFailure def Open(self): global g_window,g_entry _open(self) channel=g_switcher.OnGame(self.stream) g_window=ChannelWindow(g_switcher,self.stream);g_entry=Entry(self) if channel: name=dict(Channels(Locate(self.stream.addr,self.stream.port))).get(channel,{}).get('name','') chat.AppendChat(chat.CHAT_TYPE_INFO,L('已换到 %s'%name)) def Close(self): global g_window,g_entry if g_entry:g_entry.Hide();g_entry.event=None;g_entry=None if g_window:g_window.Destroy();g_window=None _close(self) def OnUpdate(self): _update(self) if g_switcher.stage=='logout':g_switcher.Tick() def LoginOpen(self): _login_open(self);g_switcher.OnLoginWindow(self) def LoginUpdate(self): _login_update(self);g_switcher.Tick() def LoginFailure(self,error): g_switcher.OnLoginFailure(error);_login_failure(self,error) def ConnectFailure(self): g_switcher.OnLoginFailure('connect');_connect_failure(self) game.GameWindow.Open=Open game.GameWindow.Close=Close game.GameWindow.OnUpdate=OnUpdate introLogin.LoginWindow.Open=LoginOpen introLogin.LoginWindow.OnUpdate=LoginUpdate introLogin.LoginWindow.OnLoginFailure=LoginFailure introLogin.LoginWindow.OnConnectFailure=ConnectFailure )PY";