#------------------# import os,requests,json,time,sys,psutil,string,platform,glob,getpass,shutil,hashlib,random,subprocess,math,zipfile,threading #------------------# class MultiThreadDownload(threading.Thread): '''The Class for Multi-Thread Download''' '''Get from Internet and improved by Ghink Network Studio''' def __init__(self,url,startpos,endpos,f,UA): super(MultiThreadDownload,self).__init__() self.url=url self.startpos=startpos self.endpos=endpos self.fd=f self.UA=UA def download(self): headers=self.UA.update({"Range":"bytes=%s-%s"%(self.startpos,self.endpos)}) res=requests.get(self.url,headers=headers) self.fd.seek(self.startpos) self.fd.write(res.content) def run(self): self.download() class GMCLCore(object): '''The Main Class of the Launcher Core''' ''' self.__GamePath=GamePath self.__Version=Version ''' def __init__(self,LauncherName="GMCLCore",LauncherVersion="A0.2.0",LogOutput=False): '''The global variable set function''' ''':LauncherName The name of your launcher,the default value is "GMCLCore"''' ''':LauncherVersion The version of your launcher,the default value is the version of core''' ''':LogOutput The switch of the log output function,the default value is False''' self.__LauncherName=LauncherName self.__LauncherVersion=LauncherVersion self.__LogOutput=LogOutput self.__UserAgent={'User-Agent':LauncherName+'/'+LauncherVersion+' ((GMCL Core Alpha 0.2.0;Alpha))'} self.__Log=[] def Log(self,Type,Text,Function="Anonymous Function",ErrorType="TypeError"): '''The function which was used to manager the log output system''' ''':Type Type of the log info''' ''':Text The main text of log info''' ''':Function The function that cause the log,the default value is "Anonymous Function"''' ''':Type The type of the error,the default value is "TypeError",Support "TypeError","ValueError","UserWarning"''' if(Text!="" or None): if(Type=="info"): self.__Log.append(("INFO",time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),Function,Text)) return ("INFO",Text) elif(Type=="warn"): self.__Log.append(("WARN",time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),Function,Text)) return ("WARN",Text) elif(Type=="error"): self.__Log.append(("ERROR",time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),Function,Text,ErrorType)) if(ErrorType=="TypeError"): raise TypeError(Text) elif(ErrorType=="ValueError"): raise ValueError(Text) else: raise UserWarning(Text) else: self.__Log.append(("OTHER",time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),Function,Text)) return ("OTHER",Text) def MakeDir(self,Path): '''The function which was used to create a dir with determination''' ''':Path The path of the dir you want to create''' if(os.path.isdir(Path)==False): os.mkdir(Path) return self.Log("info","Successful.","MakeDir") else: return self.Log("warn","Exist.","MakeDir") def SearchFile(self,Path,Tag,Num=1): '''The function which was used to search files''' ''':Path The root path you want to search''' ''':Tag The name keyword of the file you want to search''' ''':Num How much result you want to get,you should set as "all" if you want to return all result,the default value is 1''' List=[] i=0 for root,dirs,files in os.walk(Path,topdown=True): if Tag in files: i+=1 List.append(root+'\\'+Tag) if(i==Num): break else: pass return List def GetUUID(self,Account="Steve"): '''The function which was used to generate UUID''' ''':Account Your account,the default value is "Steve"''' if(self.__AuthWay=="offline"): result="OfflinePlayer:"+Account md5=hashlib.md5() md5.update(result.encode(encoding='utf-8')) return md5.hexdigest() else: return "Error:Wrong Auth Way." def GetJavaPath(self,Num=1): '''The function which was used to search java''' ''':Num How much result you want to get,you should set as "all" if you want to return all result,the default value is 1''' if(platform.system()=="Windows"): return self.SearchFile('C:\\','javaw.exe',Num) elif(platform.system()=="Linux"): return self.SearchFile('\\','java',Num) else: return self.SearchFile('\\','java',Num) def SetAccount(self,Account="Steve",Password=""): '''The function which was used to set game account data''' ''':Account Your account,the default value is "Steve"''' ''':Password Your password''' if(self.__AuthWay=="online"): if(Account.count("@")==1): if(json.loads(requests.get("https://authserver.mojang.com/",headers=self.__UserAgent).text)['status']=="OK"): result=json.loads(requests.post("https://authserver.mojang.com/authenticate",'{"agent":{"name":"Minecraft","Version":1},"username":"'+Account+'","password":"'+Password+'"}',headers=self.__UserAgent).text) self.__Token=result['accessToken'] self.__UUID=result['selectedProfile']['id'] self.__ID=result['selectedProfile']['name'] else: return "Error:Fail to request Mojang\'s Auth Server." else: return "Error:Wrong Account." elif(self.__AuthWay=="offline"): if(Account.count("@")==1): return "Error:Wrong Account." else: self.__ID=Account self.__UUID=self.GetUUID(Account) self.__Token="" else: pass def SetAuthWay(self,AuthWay): '''The auth way set function''' ''':AuthWay Your auth way,must be "offline" or "online"''' if(AuthWay=="offline"): self.__AuthWay="offline" elif(AuthWay=="online"): self.__AuthWay="online" else: return "Error:Wrong Auth Way." def SetRecomMem(self): '''The recommendation memory value set function''' self.__Memory = float(psutil.virtual_memory().free / 1024 ** 3) * 0.8 def SetJavaPath(self,path): if(path==None): return "Error:Wrong local address for java." else: if(os.path.exists(path)): self.__JAVA=path else: return "Error:Wrong local address for java." def SetGameMem(self,Memory): try: self.__Memory=float(Memory) except: return "Error:Wrong parameter for game memory." def Download(self,DownloadFrom,DownloadTo,ThreadNum=3): '''Multi-Thread download function''' if(DownloadFrom=="" or DownloadTo==""): return "Error:Wrong online address or local address for download." else: url = DownloadFrom filename = DownloadTo filesize = int(requests.head(url,headers=self.__UserAgent).headers['Content-Length']) threadnum = ThreadNum threading.BoundedSemaphore(threadnum) step = filesize // threadnum mtd_list = [] start = 0 end = -1 tempf = open(filename,'w') tempf.close() with open(filename,'rb+') as f: fileno = f.fileno() while end < filesize -1: start = end +1 end = start + step -1 if end > filesize: end = filesize dup = os.dup(fileno) fd = os.fdopen(dup,'rb+',-1) t = MultiThreadDownload(url,start,end,fd,self.__UserAgent) t.start() mtd_list.append(t) for i in mtd_list: i.join()