#!/usr/bin/env python ## Python Grub Client ## http://www.grub.org ## ## Copyright (c) 2008 Giorgos Logiotatidis (seadog@sealabs.net) ## ## Grub.py is free software; you can redistribute it and/or modify ## it under the terms of the GNU General Public License as published by ## the Free Software Foundation; either version 3 of the License, or ## (at your option) any later version. ## ## Grub.py is distributed in the hope that it will be useful, ## but WITHOUT ANY WARRANTY; without even the implied warranty of ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the ## GNU General Public License for more details. ## ## You should have received a copy of the GNU General Public License ## along with Grub.py; if not, write to the Free Software ## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ## import urllib2 import httplib import time import gzip import socket from threading import Thread import thread import sys import os import getopt from cStringIO import StringIO import zlib from shutil import copy NUM_CLIENTS = 5 NUM_UPLOADERS = 1 USERNAME = None PASSWORD = None VERBOSE = True DEBUG = False DISPATCHER = 'http://soap.grub.org/cgi-bin/dispatch.cgi' VERSION = '0.4.3' # set socket timeout socket.setdefaulttimeout(30) workerlist = [] threadlock = thread.allocate_lock() threadlock_fetch = thread.allocate_lock() class Uploader(Thread): def __init__(self, name="Uploader"): Thread.__init__(self) self.setName(name) self._sleeptime = 10 self.LOOP = True def sleep(self, error, seconds): printer("%s Retrying in %s seconds" % (error, seconds), error=1) time.sleep(seconds) def run(self): while self.LOOP: if os.path.exists("workunits.dat") == False: # nothing ready yet. Get some sleep and try again later time.sleep(self._sleeptime) continue threadlock.acquire() try: fp = open("workunits.dat", "rb") workUnit = fp.readlines() fp.close() fp = open("workunits.dat", "wb") fp.writelines(workUnit[1:]) fp.close() except IOError, e: printer("Workunit.dat open failed in %s: %s" % (self.getName(), e), debug=1) threadlock.release() # keep only the first line of the file try: workUnit = workUnit[0].strip() except IndexError: # seems there is no line # sleep some time and then retry time.sleep(self._sleeptime) continue if len(workUnit) > 0: try: host, url, filename = workUnit.split() except ValueError: # oops maybe wrong entry. lets ignore continue # compress the workunit try: self.compressWorkUnit(filename) except IOError, e: # hmm arc file maybe deleted. just move on printer("CompressWorkUnit error %s:%s" % (self.getName(), e), debug=1) continue i=0 while self.LOOP: i+=1 if i==10: break try: result = self.uploadWorkUnit(host, url, filename) except (httplib.HTTPException, socket.error): result = -500 if result == 1: # everything went OK break elif result == -500: # let's retry self.sleep("Thread %s failed to upload arc file because of a server or network error." % self.getName(), seconds=self._sleeptime*i) elif result == -401: # skip cleaning by changing the filename if DEBUG: copy(filename, "%s.401" % filename) copy("%s.wu" % filename, "%s.wu.401" % filename) break self.cleanup(filename) def compressWorkUnit(self, filename): # gzip the data printer("Thread %s is packing" % self.getName()) farc = open(filename, 'rb') f = gzip.open("%s.gz" % filename, 'wb') data = farc.read(500) while len(data) > 0: f.write(data) data = farc.read(500) f.close() farc.close() printer("Thread %s finished packing" % self.getName(), debug=1) def cleanup(self, filename): try: os.remove("%s" % filename) os.remove("%s.gz" % filename) if DEBUG: # remove workunit os.remove("%s.wu" % filename) except (OSError, TypeError), e: printer("Cleanup error %s:%s" % (self.getName(), e), debug=1) def join(self): self.LOOP=False def uploadWorkUnit(self, host, url, filename): # upload data filename = "%s.gz" % filename start_stamp = int(time.time()) printer("Thread %s is uploading" % self.getName()) try: fsize = os.path.getsize(filename) f = open("%s" % filename, "rb") except IOError, e: # the file does not exist, return a 401 to ignore that printer("uploadWorkUnit error %s: %s" % (self.getName(), e), debug=1) return -401 try: conn = httplib.HTTP(host) conn.putrequest('PUT', url) conn.putheader('Content-Length', str(fsize)) conn.endheaders() conn.send(f.read()) errcode, errmsg, headers = conn.getreply() except (socket.error), e: # ok socket timedout, let's retry printer("uploadWorkUnit error %s: %s" % (self.getName(), e), debug=1) f.close() conn.close() raise socket.error f.close() conn.close() if errcode == 200: stop_stamp = int(time.time() - start_stamp) printer("Thread %s upload OK at %s Kb/s" % (self.getName(), str(int((fsize/1024)/stop_stamp)) ) ) return 1 else: printer("Thread %s upload FAILED, Code: %s Message: %s" % (self.getName(), errcode, errmsg), error=1) return -errcode return 0 class Crawler(Thread): def __init__(self, name): Thread.__init__(self) self.setName(name) self.wu = None self.filename = None self.arc = None self._sleeptime = 10 self.LOOP = True def sleep(self, error): printer("%s Retrying in %s seconds" % (error, self._sleeptime), error=1) time.sleep(self._sleeptime) self._sleeptime *= 2 def fetchWorkUnit(self): # make this function atomic so we don't hammer the dispatch server threadlock_fetch.acquire() # create an OpenerDirector with suppoer for Basic HTTP Authentication printer("Thread %s is fetching Work Unit" % self.getName()) auth_handler = urllib2.HTTPBasicAuthHandler() auth_handler.add_password('test', 'soap.grub.org', USERNAME, PASSWORD) opener = urllib2.build_opener(auth_handler) try: response = opener.open(DISPATCHER) self.wu = response.readlines() response.close() opener.close() #reset sleep time self._sleeptime = 10 except (urllib2.URLError, socket.error), e: threadlock_fetch.release() self.sleep("Thread %s failed to fetch workunit. Maybe your username or password is incorrect." % self.getName()) return False threadlock_fetch.release() return True def checkWorkUnit(self): """ Set self.filename and Return 1 if OK Return 0 otherwise """ printer("Thread %s is checking Work Unit" % self.getName()) for line in self.wu: if line[0:3] == "PUT": try: line = line.split()[1].split('/') self.filename = line[len(line)-1][:-3] if os.path.exists("%s.gz" % self.filename): # oops file already exists! # The server feed us previous stuff # we should fetch another wu printer("Oops, same file!") return False return True except: return False return False def run(self): while self.LOOP: if self.fetchWorkUnit() == False or self.checkWorkUnit() == False: # oops, bad workunit, let's move to a new one continue # if debug save workunit if DEBUG: f = open("%s.wu" % self.filename, 'wb') f.writelines(self.wu) f.close() self.arc = open("%s" % self.filename, 'wb') self.arc.write("filedesc://dummy.arc.gz 0.0.0.0 %s text/plain 69\n" % time.strftime("%Y%m%d%H%M%S", time.localtime())) self.arc.write('1 0 grub.org\n') self.arc.write('URL IP-address Archive-date Content-type Archive-length\n\n') printer("Thread %s is crawling" % self.getName()) url = '' host = '' wuIterator = iter(self.wu) while self.LOOP: line = wuIterator.next() while line.strip() == '': try: line = wuIterator.next() except StopIteration: return 0 if line[0:3] == "PUT": url = line.split()[1] host = wuIterator.next()[6:].strip() break data = '' body = '' host = '' url = '' try: url = line.split()[1] host = wuIterator.next()[6:].strip() except StopIteration: return 0 except IndexError: # lets hope it's just a wrong entry and continue continue #python2.4 compatibility with try/finally try: try: headers = { # 'Accept':'*/*', # 'Accept-encoding':'gzip,deflate', 'Connection':'close', } while True: header = wuIterator.next().strip() if header == '': break headerType, headerValue = header.split(':', 1) headerType = headerType.strip() headerValue = headerValue.strip() headers[headerType] = headerValue ip = socket.gethostbyname(host) conn = httplib.HTTPConnection(host) conn.request("GET", url, headers=headers) response = conn.getresponse() if response.version == "10": data += "HTTP/1.0 " else: data += "HTTP/1.1 " data += "%s %s\r\n" % (response.status, response.reason) for header in response.getheaders(): data += ("%s: %s\r\n" % (header[0], header[1])) data += "\r\n" body = self.readResponse(response) if len(body) == 0: # we got only headers # printer("Headers only: --%s%s-- Status %s"% (host,url, response.status), error=1) raise httplib.HTTPException # check for compressed content if response.getheader('content-encoding') in ('gzip', 'deflate'): body = self.decompressData(body, response.getheader('content-encoding')) responseType = self.getResponseType(response.msg.type) # close connection conn.close() except (httplib.HTTPException, ValueError, socket.error), e: data = 'HTTP/1.0 500 Invalid URL\r\n\r\nInvalid URL' body = '' ip = "0.0.0.0" responseType = 'application/x-grub-error' #printer("Invalid URL %s%s %s" % (host,url,e), error=1) finally: self.arc.writelines("http://%s%s %s %s %s %s\n" % (host, url, ip, time.strftime("%Y%m%d%H%M%S", time.localtime()), responseType, len(data) + len(body))) self.arc.writelines(data) self.arc.writelines(body) self.arc.writelines("\n") # ok we finished the workunit self.arc.close() # maybe we got a signal to quit before we finish # so check if we reached the last line before writing # to datafiel if line[0:3] == "PUT": try: self.writeDataFile(host, url) except IOError: # we have to do something with "too many open files errors" printer("Thread %s exception on writeDataFile %s"% (self.getName(), e), debug=1) self.cleanup() self.LOOP = False self.cleanup(real=True) def readResponse(self, r): data = '' while True: tmp = r.read(500) if len(tmp) == 0: break data += tmp return data def getResponseType(self, responseType): try: if not responseType: # broken http server. let's guess it's text/html return "text/html" # deal with other broken http servers. Some even send 'text/html text/html' return responseType.split()[0] except: raise httplib.HTTPException def decompressData(self, compressed_data, method): try: if method == 'gzip': compressed_data = StringIO(compressed_data) vfile = gzip.GzipFile(fileobj=compressed_data) data = vfile.read() vfile.close() compressed_data.close() elif method == 'deflate': try: data = zlib.decompress(compressed_data) except zlib.error: # server does not respect RFC 1950 # check http://www.velocityreviews.com/forums/t634601-deflate-with-urllib2.html data = zlib.decompress(compressed_data, -zlib.MAX_WBITS) except Exception, e: printer("Decompress %s: %s" % (self.getName(), e), debug=1) vfile.close() compressed_data.close() raise ValueError return data def writeDataFile(self, host, url): if (len(host) == 0 or len(url) == 0): return threadlock.acquire() try: fp = open("workunits.dat", "ab") fp.write("%s %s %s\n" % (host, url, self.filename)) fp.close() except IOError, e: printer("WriteDataFile error %s: %s" % (self.getName(), e), debug=1) threadlock.release() def cleanup(self, real=False): if real: try: os.remove(self.filename) except (OSError, TypeError), e: printer("Cleanup error %s:%s" % (self.getName(), e), debug=1) self.filename = None self.wu = None def join(self): self.LOOP=False def printer(msg, error=0, debug=0): p = False if (debug==1): if DEBUG: p = True elif (error==1): p = True elif (VERBOSE == True): p = True if p: print "[%s] %s" % (time.strftime("%a %b %d %H:%M:%S %Y"), msg) sys.stdout.flush() def printUsage(): global VERSION print "Grub Python Client version %s" % VERSION print "http://www.grub.org/\n" print "Usage:" \ "\tgrub.py [-v] [-t NUMBER] -u USERNAME -p PASSWORD\n" \ "\n" \ "\t -v : Verbose (Default. Use quiet (-q) to turn off messages\n" \ "\t -q : Quiet mode. Print only error messages\n" \ "\t -t, --threads : Number of threads. Default 5\n" \ "\t -u, --username : Grub.org username\n" \ "\t -i, --info : Advanced, prints client and OS imformation\n" \ "\t -p, --password : Grub.org password\n" \ "\t --debug : Set debug mode. Failed arcs are not removed\n" \ "\t --version : Print version\n" \ "\n" def parseConfig(): global USERNAME, PASSWORD if USERNAME == None or PASSWORD == None: # try to read username and password from config file passwdfile = "" if os.path.exists(os.path.expanduser("~") + "/.grub/account"): passwdfile = os.path.expanduser("~") + "/.grub/account" elif os.path.exists("./account"): passwdfile = "./account" else: # no username/password specified! printer("No username and/or password specified!", error=1) sys.exit() try: USERNAME, PASSWORD = open(passwdfile).readline().split(':') USERNAME = USERNAME.strip() PASSWORD = PASSWORD.strip() except: printer("Error reading username and password from %s" % passwdfile) sys.exit(1) def printInfo(): print "Grub Python Client version: %s" % VERSION print "Username: %s" % USERNAME print "Password set: %s" % bool(PASSWORD) print "Number of clients: %s" % NUM_CLIENTS print "Operating system: %s" % sys.platform print "Python version: %s" % sys.version def main(): global USERNAME, PASSWORD, VERBOSE, NUM_CLIENTS, DEBUG try: opts, args = getopt.getopt(sys.argv[1:], 'ihvqt:u:p:', ["username=", "password=", "threads=", "version", "info", "debug", "help"]) except getopt.GetoptError, err: print str(err) sys.exit(2) for o, a in opts: if o == "-v": printer("Verbose mode (-v) is now default. Use quiet (-q) to turn off messages") elif o == "-q": VERBOSE=False elif o in ("-t", "--threads"): try: NUM_CLIENTS = int(a) except ValueError: printer("Please give a valid thread number", error=1) sys.exit(1) elif o in ("-u", "--username"): USERNAME = a elif o in ("-p", "--password"): PASSWORD = a elif o in ("-h", "--help"): printUsage() sys.exit(0) elif o in ("-i", "--info"): parseConfig() printInfo() sys.exit(0) elif o in ("--debug"): DEBUG = True elif o in ("--version"): print "Grub Python Client, version %s" % VERSION sys.exit(0) else: assert False, "unhandled option" parseConfig() for i in range(NUM_CLIENTS): w = Crawler(i) workerlist.append(w) w.start() for i in range(NUM_UPLOADERS): w = Uploader("Uploader-%s" % i) workerlist.append(w) w.start() try: sys.stdin.read() except (KeyboardInterrupt, SystemExit): print "Closing threads, please wait..." sys.stdout.flush() for worker in workerlist: worker.join() sys.exit() if __name__ == "__main__": main()