This is my first blog, I hope to be useful to everyone. In order to facilitate the update the Maxthon proxy server, write a Python script. Processing proxy4free gets the proxy server file, verifying the proxy server, sorted by response speed, generating the MaxThon configuration format output. Here is the file: proxy2maxthon.py '' 'gets the proxy information from http://www.proxy4free.com/page1.html, outputs the usage of Maxthon (MYIE2 format) Usage: Paste the proxy information Copy Paste to a text file proxylist.txt, each acts as a proxy message format: 213.199.192.37 3128 transparent Poland 15.10.2004 Whois 156.110.47.251 8080 anonymous United States 16.10.2004 Whois 68.191.111.217 80 anonymous United States 16.10.2004 Whois 24.14.246.55 80 anonymous United STATES 16.10.2004 After performing this procedure, paste the output screen prompt information into the Proxy section in the Maxthon profile (setupCenter.ini in the config directory of the Maxthon directory), pay attention: When modifying the configuration file, you must ensure that Maxthon is not run. Author: invalid email: invalid@21cn.com Date: 2004-10-16 Day '' 'import osimport urllibimport timeimport threadingimport Queuefrom string import strip
Version_info = (1, 2, "alpha", '20041016')
Max_threads = 50max_proxys = 30 # As long as the fastest Max_Proxys a proxy http_timeout = 10.0 # Wait to return to the maximum time, second TestURL = 'http://www.google.com/' # Used to test the website TestResp = 'Google' # 标 成 网 内容 内容 内容 排 排 排 排 排 序 排 排 排 序 类 类 序 类 序 序 序 序::: 序 类:::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::: ( I] for junk, i in aux] if INPLACE: DATA [:] = Result Return RESULT
Def Byitem (Self, Data, ItemIndex = None, Inplay = 1): if ItemIndex is None: if ITEMINDEX IS NONE: IF INPLACE: DATA.SORT () Result = data else: Result = data [:] result.sort () Return Result else: AUX = [(Data [i] [itemindex], i) for i in ing (len (data))] Return Self._helper (data, aux, inplay) # a couple of handy synonyms sort = byitem __call__ = byitem
DEF TESPPROXYDIRECT (Proxy): 'The function of the test proxy server, the return value is a tuple (success flag, the time) disadvantage is uncontrollable, the test will be slower parameter proxy format: IP: port' ' IRET = 0 usedtime = 0; proxies = proxies = {'http': 'http: //' proxy} try: tstart = time.time () filehandle = urllib.urlopen (testURL, proxies = proxies) Data = filehandle. Readlines () FileHandle.Close () TEND = Time.Time () Usedtime = TEND - TSTART for LINE IN DATA: IF line.Find (TestRESP)> 0: IRET = 1 Break Except: Pass Return (IRet, UsedTime);
DEF TestProxy (WorkQueue, Resultqueue):
'' 'Working thread, whether the test agent works, and the agent response speed acquires the proxy parameters from WorkQueue, the test successfully saves to ResultQueue.' '' DEF SubthreadProc (URL, Result):
'' 'Working process, can test if the test agent works, and return response speed' '' '' '' '' '' '' '= proxies = {' http ':' http: // ' proxy} try: tstart = time.time FileHandle = urllib.urlopen (TestURL, Proxies = proxies) Data = fileHandle.Readlines () filehandle.close () TEND = time.time () USEDTIME = (TEND - TSTART) * 1000 for line in data: if line.find (TESTRESP)> 0: result.append (usedtime) except: result.append (-1) return while 1: # contine pulling data from the work queue until it's empty try: proxy, local = workQueue.get (0) except Queue .Empty: # work queue is empty - exit the thread proc return # Create a single subthread to do the actual work result = [] subThread = threading.Thread (target = SubthreadProc, args = (proxy, result)).
# Daemonize the Subthread So That Even IFE WILL EXIT. Subthread.SetDaemon (TRUE)
# Run the Subthread and Wait for it to finish, or time out subthread.start () Subthread.join (http_timeout)
IF [] == Result: # Subthread Hasn't Give A Result Yet. Consider IT Timed Out. # print proxy, "timeout" pass elif -1 == result [0]: # Subthread Returned An Error from getURL. # print Proxy, "failed" Pass Else: # Subthread Returned A Time. Store It. # Print Proxy, Result [0] ResultQueue.put ((Proxy, Local, Result [0])))))
Def genmaxthoncfg (proxys): '' Generate the Maxthon Profile Description Format '' '#index indicates that the start configuration number index = 4 #Timeout indicates that the agent gets the maximum time of Google, more than this time agent is discarded, the unit: millisecond Timeout = 10000 # Sort by acquisition time, time short ranks in front sort = sORTER () sort (proxys, 2) print "********************** ****************************** "for item in proxys: print item [0], item [1], int (Item [2]) Print "************************************************** ********* "Print" Paste the following to paste to Maxthon's setupCenter.ini configuration file's proxy section "count = 0 for item in proxys: #maxthon agent configures: # p1 = http = 210.230.192.39:3128 # ps1 = 3 # pn1 = japan if Int (item [2])> Timeout: Continue P = "p" str (index) "= http =" item [0] ps = "PS " strick " = 3 "pn =" pn " str (index) " = " item [1] str (index) Print P Print PS Print PN INDEX = 1 count = 1 if Count> max_proxys: Break Return Def Main (): # Open the proxy file, start processing try: file = open ("proxylist.txt", "R") Except: Print "Open proxylist file error" sys.exit (1) data = file.readline () file.close () Print "Proxylist File Have Proxy:", Len (DATA) URLS = [] # Record The Start Time, So We Can Print A Nice Message At The End ProcessStartTime = Time.Time ()
Numthreads = min (max_threads, len (data))
# Validated agent record, unit format is a proxy address port, name, get google time proxys = [] WorkQueue = queue.queue () for line in data: if line.find ("China"> 0: # Proxy server contract = line.split ("") if len (sects)> 6: ip = strip (sects [0]) port = strip (sects [1]) local = SECTS [-4] if urls.count (ip)> 0: # Proxy server repeated Continue Urls.Append (IP) Workqueue.put ((ip ": port, local) # 工作 工作 线 w = [] resultqueue = queue.queue ()
# Create worker threads to load-balance the retrieval print "create threads ..." for threadNum in range (0, numThreads): workers.append (threading.Thread (target = TestProxy, args = (workQueue, resultQueue))) workers [-1] .start ()
# Wait for all the workers to finish print "Waiting Threads Finish ..." for W in workers: w.join () print "threads done." While NOT RESUEUE.EMPTY (): proxy, local, result = resultQueue.get (0) Proxys = proxys [(Proxy, local, int (result))]
GenMaxthoncfg (Proxys)
IF __NAME__ == '__main__': main ()
