Hello All, I have this class I like to call pcrunchly. That works to do all my request via libcurl library. What I want to do is add capability for this class to fallback to urllib if pycurl module is not install or is not importable for some reason. I'm posting my whole class. Use it however you want, just please help me to make it better. As you can see from the code I've already started to add this functionality. I found some code on Google that I'm trying to combine with my class, so that it can achieve this task. I want to ask the reader of this thread. Is this a good conservative thing to do. Will this make my class to bulky. Is this something I want? or do I want something else? Is there a better way to achieve this perhaps? I would appreciate any comments on this matter. I would like this class to be able to handle any protocol you throw at it. So as to make it other apps that need to use a different protocol and stuff. Please notice the first huge comment. That is what I'm currently combining into my working pycurl class. Is this good? Oh yeah, please don't mind all those self.soc functions calls. They print things in color to my terminal for me.
#!/usr/bin/env python from ctypes import * import os, sys, types, urllib, urllib2, urlparse import stdout_colours #1. #!/usr/bin/env python #2. # -*- coding: UTF-8 -*- #3. #4. import cookielib #5. import urllib #6. import urllib2 #7. #8. cj = cookielib.CookieJar() #9. opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(c j)) #10. resp = opener.open('http://www.amm.com/login.asp') # save a cookie #11. #12. theurl = 'http://www.amm.com/login.asp' #13. # an example url that sets a cookie, try different urls here and see the cookie collection you can make ! #14. body={'username':'AMMT54590570','password':'AMMT32 564288'} #15. txdata = urllib.urlencode(body) #16. # if we were making a POST type request, we could encode a dictionary of values here - using urllib.urlencode #17. txheaders = {'User-agent' : 'Mozilla/4.0 (compatible; MSIE 5.5; Windows NT)'} #18. # fake a user agent, some websites (like google) don't like automated exploration #19. #20. #21. try: #22. req = urllib2.Request(theurl, txdata, txheaders) # create a request object #23. handle = opener.open(req) # and open it to return a handle on the url #24. HTMLSource = handle.read() #25. f = file('test.html', 'w') #26. f.write(HTMLSource) #27. f.close() #28. #29. except IOError, e: #30. print 'We failed to open "%s".' % theurl #31. if hasattr(e, 'code'): #32. print 'We failed with error code - %s.' % e.code #33. elif hasattr(e, 'reason'): #34. print "The error object has the following 'reason' attribute :", e.reason #35. print "This usually means the server doesn't exist, is down, or we don't have an internet connection." #36. sys.exit() #37. #38. else: #39. print 'Here are the headers of the page :' #40. print handle.info() # handle.read() returns the page, handle.geturl() returns the true url of the page fetched (in case urlopen has followed any redirects, which it sometimes does) class curl(object): "Encapsulate user operations on CGIs through curl." def __init__(self, base_url=""): self.func_me_color="white_on_black" self.soc=stdout_colours.stdout_colors() self.soc.me_him(['ENTER:',__name__],self.func_me_color) # These members might be set. self.base_url = base_url self.verbose = 0 self.response = "" self.PYCURL=None try: import pycurl self.PYCURL = True except ImportError,e: import cookielib import urllib import urllib2 self.PYCURL = False #print "\ntrouble with importing pycurl %s\n" % e #sys.exit(1) # Nothing past here should be modified by the caller. if self.PYCURL: self.curlobj = pycurl.Curl() # Verify that we've got the right site... #self.curlobj.setopt(pycurl.SSL_VERIFYHOST, 2) # Follow redirects in case it wants to take us to a CGI... self.curlobj.setopt(pycurl.FOLLOWLOCATION, 1) #self.curlobj.setopt(pycurl.MAXREDIRS, 15) # Setting this option with even a nonexistent file makes libcurl # handle cookie capture and playback automatically. self.curlobj.setopt(pycurl.COOKIEFILE, "/dev/null") # Set timeouts to avoid hanging too long self.curlobj.setopt(pycurl.CONNECTTIMEOUT, 60) self.curlobj.setopt(pycurl.TIMEOUT, 600) #self.set_verbosity(self.verbose) # Set up a callback to capture else: self.cj=cookielib.CookieJar() print "\nurllib:\n" def response_callback(x): self.soc.me_him(['ENTER:',__name__],self.func_me_color) if self.PYCURL: self.response += x self.curlobj.setopt(pycurl.WRITEFUNCTION, response_callback) self.soc.me_him(['EXIT:',__name__],self.func_me_color) def set_verbosity(self, level): "Set verbosity to 1 to see transactions." self.soc.me_him(['ENTER:',__name__],self.func_me_color) if self.PYCURL: self.verbose=int(level) self.curlobj.setopt(pycurl.VERBOSE, self.verbose) self.soc.me_him(['EXIT:',__name__],self.func_me_color) def get(self, cgi, params=""): "Ship a GET request to a specified CGI, capture the response body." self.soc.me_him(['ENTER:',__name__],self.func_me_color) if self.PYCURL: if params: cgi += "?" + urllib.urlencode(params) self.curlobj.setopt(pycurl.URL, os.path.join(self.base_url, cgi)) self.curlobj.setopt(pycurl.HTTPGET, 1) self.response = "" self.curlobj.perform() # if self.verbose > 0: # print self.response self.soc.me_him(['EXIT:',__name__],self.func_me_color) def post(self, cgi, params): "Ship a POST request to a specified CGI, capture the response body.." self.soc.me_him(['ENTER:',__name__],self.func_me_color) if self.PYCURL: self.curlobj.setopt(pycurl.URL, os.path.join(self.base_url, cgi)) self.curlobj.setopt(pycurl.POST, 1) self.curlobj.setopt(pycurl.POSTFIELDS, urllib.urlencode(params)) self.response = "" self.curlobj.perform() # if self.verbose>0: # print self.response self.soc.me_him(['EXIT:',__name__],self.func_me_color) def upload(self, cgi, file_name, file): "POST file from localhost to location/cgi." self.soc.me_him(['ENTER:',__name__],self.func_me_color) if self.PYCURL: self.curlobj.setopt(pycurl.URL, os.path.join(self.base_url, cgi)) self.curlobj.setopt(pycurl.HTTPPOST,[(file_name, (pycurl.FORM_FILE,file))]) self.response = "" self.curlobj.perform() if self.verbose>0: print self.response self.soc.me_him(['EXIT:',__name__],self.func_me_color) # -------------------------------- # DJANGO RECEIVE TEST APPLICATION # -------------------------------- # --------- urls.py ---------------- #from django.conf.urls.defaults import * #urlpatterns = patterns('', #(r'^receive/$', 'web.views.receive'), #) # --------- web\views.py ---------------- #def receive(request): #assert request.method=="POST" #print "receive.META.SERVER_PORT", request.META["SERVER_PORT"], request.POST #files = [] #for multipart_name in request.FILES.keys(): #multipart_obj = request.FILES[multipart_name] #content_type = multipart_obj['content-type'] #filename = multipart_obj['filename'] #content = multipart_obj['content'] #files.append((filename, content_type, content)) #import datetime # write file to the system - add timestamp in the name #file("c:\\tmp\\%s_%s" % (datetime.datetime.now().isoformat().replace(":", "-"), filename), "wb").write(content) #fnames = ",".join([fname for fname, ct, c in files]) #return HttpResponse("me-%s-RECEIVE-OK[POST=%s,files=%s]" % (request.META["SERVER_PORT"], request.POST.values(), fnames )) def answered(self, check): "Does a given check string occur in the response?" self.soc.me_him(['ENTER:',__name__],self.func_me_color) self.soc.me_him(['RETURN:',__name__],self.func_me_color) if self.PYCURL: return self.response.find(check) >= 0 def close(self): "Close a session, freeing resources." self.soc.me_him(['ENTER:',__name__],self.func_me_color) if self.PYCURL: self.curlobj.close() self.soc.me_him(['EXIT:',__name__],self.func_me_color) class session(curl): def login(self, cgisite,username, password): """login - cgi="login.php",params=(("username",name),("password",pass),("foo","bar")) """ self.soc.me_him(['ENTER:',__name__],self.func_me_color) self.post(cgisite, (("username",username), ("password",password))) self.soc.me_him(['EXIT:',__name__],self.func_me_color) def logout(self, cgisite): """logout - cgi="logout.php" """ self.soc.me_him(['ENTER:',__name__],self.func_me_color) self.get(cgisite) self.soc.me_him(['EXIT:',__name__],self.func_me_color) if __name__ == "__main__": if len(sys.argv) < 3: print "Usage: %s \"schema://site/cgi\" \"username\" \"password\"" % sys.argv[0] site=sys.argv[1] username=sys.argv[2] password=sys.argv[3] sess=session("") sess.set_verbosity(1) sess.login(site,username,password) a="" for i in range(len(password)): a+="*" print "YOU ARE LOGGED IN!",site,username,a sess.logout() sess.close() -- А-Б-В-Г-Д-Е-Ё-Ж-З-И-Й-К-Л-М-Н-О-П-Р-С-Т-У-Ф-Х-Ц-Ч-Ш-Щ-Ъ-Ы-Ь-Э-Ю-Я а-б-в-г-д-е-ё-ж-з-и-й-к-л-м-н-о-п-р-с-т-у-ф-х-ц-ч-ш-щ-ъ-ы-ь-э-ю-я
-- http://mail.python.org/mailman/listinfo/python-list