home *** CD-ROM | disk | FTP | other *** search
- # Open an arbitrary URL
- #
- # See the following document for more info on URLs:
- # "Names and Addresses, URIs, URLs, URNs, URCs", at
- # http://www.w3.org/pub/WWW/Addressing/Overview.html
- #
- # See also the HTTP spec (from which the error codes are derived):
- # "HTTP - Hypertext Transfer Protocol", at
- # http://www.w3.org/pub/WWW/Protocols/
- #
- # Related standards and specs:
- # - RFC1808: the "relative URL" spec. (authoritative status)
- # - RFC1738 - the "URL standard". (authoritative status)
- # - RFC1630 - the "URI spec". (informational status)
- #
- # The object returned by URLopener().open(file) will differ per
- # protocol. All you know is that is has methods read(), readline(),
- # readlines(), fileno(), close() and info(). The read*(), fileno()
- # and close() methods work like those of open files.
- # The info() method returns a mimetools.Message object which can be
- # used to query various info about the object, if available.
- # (mimetools.Message objects are queried with the getheader() method.)
-
- import string
- import socket
- import os
- import sys
-
-
- __version__ = '1.10'
-
- MAXFTPCACHE = 10 # Trim the ftp cache beyond this size
-
- # Helper for non-unix systems
- if os.name == 'mac':
- from macurl2path import url2pathname, pathname2url
- elif os.name == 'nt':
- from nturl2path import url2pathname, pathname2url
- else:
- def url2pathname(pathname):
- return unquote(pathname)
- def pathname2url(pathname):
- return quote(pathname)
-
- # This really consists of two pieces:
- # (1) a class which handles opening of all sorts of URLs
- # (plus assorted utilities etc.)
- # (2) a set of functions for parsing URLs
- # XXX Should these be separated out into different modules?
-
-
- # Shortcut for basic usage
- _urlopener = None
- def urlopen(url, data=None):
- global _urlopener
- if not _urlopener:
- _urlopener = FancyURLopener()
- if data is None:
- return _urlopener.open(url)
- else:
- return _urlopener.open(url, data)
- def urlretrieve(url, filename=None, reporthook=None):
- global _urlopener
- if not _urlopener:
- _urlopener = FancyURLopener()
- return _urlopener.retrieve(url, filename, reporthook)
- def urlcleanup():
- if _urlopener:
- _urlopener.cleanup()
-
-
- # Class to open URLs.
- # This is a class rather than just a subroutine because we may need
- # more than one set of global protocol-specific options.
- # Note -- this is a base class for those who don't want the
- # automatic handling of errors type 302 (relocated) and 401
- # (authorization needed).
- ftpcache = {}
- class URLopener:
-
- __tempfiles = None
-
- # Constructor
- def __init__(self, proxies=None):
- if proxies is None:
- proxies = getproxies()
- assert hasattr(proxies, 'has_key'), "proxies must be a mapping"
- self.proxies = proxies
- server_version = "Python-urllib/%s" % __version__
- self.addheaders = [('User-agent', server_version)]
- self.__tempfiles = []
- self.__unlink = os.unlink # See cleanup()
- self.tempcache = None
- # Undocumented feature: if you assign {} to tempcache,
- # it is used to cache files retrieved with
- # self.retrieve(). This is not enabled by default
- # since it does not work for changing documents (and I
- # haven't got the logic to check expiration headers
- # yet).
- self.ftpcache = ftpcache
- # Undocumented feature: you can use a different
- # ftp cache by assigning to the .ftpcache member;
- # in case you want logically independent URL openers
- # XXX This is not threadsafe. Bah.
-
- def __del__(self):
- self.close()
-
- def close(self):
- self.cleanup()
-
- def cleanup(self):
- # This code sometimes runs when the rest of this module
- # has already been deleted, so it can't use any globals
- # or import anything.
- if self.__tempfiles:
- for file in self.__tempfiles:
- try:
- self.__unlink(file)
- except:
- pass
- del self.__tempfiles[:]
- if self.tempcache:
- self.tempcache.clear()
-
- # Add a header to be used by the HTTP interface only
- # e.g. u.addheader('Accept', 'sound/basic')
- def addheader(self, *args):
- self.addheaders.append(args)
-
- # External interface
- # Use URLopener().open(file) instead of open(file, 'r')
- def open(self, fullurl, data=None):
- fullurl = unwrap(fullurl)
- if self.tempcache and self.tempcache.has_key(fullurl):
- filename, headers = self.tempcache[fullurl]
- fp = open(filename, 'rb')
- return addinfourl(fp, headers, fullurl)
- type, url = splittype(fullurl)
- if not type: type = 'file'
- if self.proxies.has_key(type):
- proxy = self.proxies[type]
- type, proxy = splittype(proxy)
- host, selector = splithost(proxy)
- url = (host, fullurl) # Signal special case to open_*()
- name = 'open_' + type
- if '-' in name:
- # replace - with _
- name = string.join(string.split(name, '-'), '_')
- if not hasattr(self, name):
- if data is None:
- return self.open_unknown(fullurl)
- else:
- return self.open_unknown(fullurl, data)
- try:
- if data is None:
- return getattr(self, name)(url)
- else:
- return getattr(self, name)(url, data)
- except socket.error, msg:
- raise IOError, ('socket error', msg), sys.exc_info()[2]
-
- # Overridable interface to open unknown URL type
- def open_unknown(self, fullurl, data=None):
- type, url = splittype(fullurl)
- raise IOError, ('url error', 'unknown url type', type)
-
- # External interface
- # retrieve(url) returns (filename, None) for a local object
- # or (tempfilename, headers) for a remote object
- def retrieve(self, url, filename=None, reporthook=None):
- url = unwrap(url)
- if self.tempcache and self.tempcache.has_key(url):
- return self.tempcache[url]
- type, url1 = splittype(url)
- if not filename and (not type or type == 'file'):
- try:
- fp = self.open_local_file(url1)
- hdrs = fp.info()
- del fp
- return url2pathname(splithost(url1)[1]), hdrs
- except IOError, msg:
- pass
- fp = self.open(url)
- headers = fp.info()
- if not filename:
- import tempfile
- garbage, path = splittype(url)
- garbage, path = splithost(path or "")
- path, garbage = splitquery(path or "")
- path, garbage = splitattr(path or "")
- suffix = os.path.splitext(path)[1]
- filename = tempfile.mktemp(suffix)
- self.__tempfiles.append(filename)
- result = filename, headers
- if self.tempcache is not None:
- self.tempcache[url] = result
- tfp = open(filename, 'wb')
- bs = 1024*8
- size = -1
- blocknum = 1
- if reporthook:
- if headers.has_key("content-length"):
- size = int(headers["Content-Length"])
- reporthook(0, bs, size)
- block = fp.read(bs)
- if reporthook:
- reporthook(1, bs, size)
- while block:
- tfp.write(block)
- block = fp.read(bs)
- blocknum = blocknum + 1
- if reporthook:
- reporthook(blocknum, bs, size)
- fp.close()
- tfp.close()
- del fp
- del tfp
- return result
-
- # Each method named open_<type> knows how to open that type of URL
-
- # Use HTTP protocol
- def open_http(self, url, data=None):
- import httplib
- user_passwd = None
- if type(url) is type(""):
- host, selector = splithost(url)
- if host:
- user_passwd, host = splituser(host)
- host = unquote(host)
- realhost = host
- else:
- host, selector = url
- urltype, rest = splittype(selector)
- url = rest
- user_passwd = None
- if string.lower(urltype) != 'http':
- realhost = None
- else:
- realhost, rest = splithost(rest)
- if realhost:
- user_passwd, realhost = splituser(realhost)
- if user_passwd:
- selector = "%s://%s%s" % (urltype, realhost, rest)
- #print "proxy via http:", host, selector
- if not host: raise IOError, ('http error', 'no host given')
- if user_passwd:
- import base64
- auth = string.strip(base64.encodestring(user_passwd))
- else:
- auth = None
- h = httplib.HTTP(host)
- if data is not None:
- h.putrequest('POST', selector)
- h.putheader('Content-type', 'application/x-www-form-urlencoded')
- h.putheader('Content-length', '%d' % len(data))
- else:
- h.putrequest('GET', selector)
- if auth: h.putheader('Authorization', 'Basic %s' % auth)
- if realhost: h.putheader('Host', realhost)
- for args in self.addheaders: apply(h.putheader, args)
- h.endheaders()
- if data is not None:
- h.send(data + '\r\n')
- errcode, errmsg, headers = h.getreply()
- fp = h.getfile()
- if errcode == 200:
- return addinfourl(fp, headers, "http:" + url)
- else:
- if data is None:
- return self.http_error(url, fp, errcode, errmsg, headers)
- else:
- return self.http_error(url, fp, errcode, errmsg, headers, data)
-
- # Handle http errors.
- # Derived class can override this, or provide specific handlers
- # named http_error_DDD where DDD is the 3-digit error code
- def http_error(self, url, fp, errcode, errmsg, headers, data=None):
- # First check if there's a specific handler for this error
- name = 'http_error_%d' % errcode
- if hasattr(self, name):
- method = getattr(self, name)
- if data is None:
- result = method(url, fp, errcode, errmsg, headers)
- else:
- result = method(url, fp, errcode, errmsg, headers, data)
- if result: return result
- return self.http_error_default(url, fp, errcode, errmsg, headers)
-
- # Default http error handler: close the connection and raises IOError
- def http_error_default(self, url, fp, errcode, errmsg, headers):
- void = fp.read()
- fp.close()
- raise IOError, ('http error', errcode, errmsg, headers)
-
- # Use Gopher protocol
- def open_gopher(self, url):
- import gopherlib
- host, selector = splithost(url)
- if not host: raise IOError, ('gopher error', 'no host given')
- host = unquote(host)
- type, selector = splitgophertype(selector)
- selector, query = splitquery(selector)
- selector = unquote(selector)
- if query:
- query = unquote(query)
- fp = gopherlib.send_query(selector, query, host)
- else:
- fp = gopherlib.send_selector(selector, host)
- return addinfourl(fp, noheaders(), "gopher:" + url)
-
- # Use local file or FTP depending on form of URL
- def open_file(self, url):
- if url[:2] == '//' and url[2:3] != '/':
- return self.open_ftp(url)
- else:
- return self.open_local_file(url)
-
- # Use local file
- def open_local_file(self, url):
- import mimetypes, mimetools, StringIO
- mtype = mimetypes.guess_type(url)[0]
- headers = mimetools.Message(StringIO.StringIO(
- 'Content-Type: %s\n' % (mtype or 'text/plain')))
- host, file = splithost(url)
- if not host:
- return addinfourl(open(url2pathname(file), 'rb'),
- headers, 'file:'+file)
- host, port = splitport(host)
- if not port \
- and socket.gethostbyname(host) in (localhost(), thishost()):
- return addinfourl(open(url2pathname(file), 'rb'),
- headers, 'file:'+file)
- raise IOError, ('local file error', 'not on local host')
-
- # Use FTP protocol
- def open_ftp(self, url):
- host, path = splithost(url)
- if not host: raise IOError, ('ftp error', 'no host given')
- host, port = splitport(host)
- user, host = splituser(host)
- if user: user, passwd = splitpasswd(user)
- else: passwd = None
- host = unquote(host)
- user = unquote(user or '')
- passwd = unquote(passwd or '')
- host = socket.gethostbyname(host)
- if not port:
- import ftplib
- port = ftplib.FTP_PORT
- else:
- port = int(port)
- path, attrs = splitattr(path)
- path = unquote(path)
- dirs = string.splitfields(path, '/')
- dirs, file = dirs[:-1], dirs[-1]
- if dirs and not dirs[0]: dirs = dirs[1:]
- key = (user, host, port, string.joinfields(dirs, '/'))
- # XXX thread unsafe!
- if len(self.ftpcache) > MAXFTPCACHE:
- # Prune the cache, rather arbitrarily
- for k in self.ftpcache.keys():
- if k != key:
- v = self.ftpcache[k]
- del self.ftpcache[k]
- v.close()
- try:
- if not self.ftpcache.has_key(key):
- self.ftpcache[key] = \
- ftpwrapper(user, passwd, host, port, dirs)
- if not file: type = 'D'
- else: type = 'I'
- for attr in attrs:
- attr, value = splitvalue(attr)
- if string.lower(attr) == 'type' and \
- value in ('a', 'A', 'i', 'I', 'd', 'D'):
- type = string.upper(value)
- (fp, retrlen) = self.ftpcache[key].retrfile(file, type)
- if retrlen is not None and retrlen >= 0:
- import mimetools, StringIO
- headers = mimetools.Message(StringIO.StringIO(
- 'Content-Length: %d\n' % retrlen))
- else:
- headers = noheaders()
- return addinfourl(fp, headers, "ftp:" + url)
- except ftperrors(), msg:
- raise IOError, ('ftp error', msg), sys.exc_info()[2]
-
- # Use "data" URL
- def open_data(self, url, data=None):
- # ignore POSTed data
- #
- # syntax of data URLs:
- # dataurl := "data:" [ mediatype ] [ ";base64" ] "," data
- # mediatype := [ type "/" subtype ] *( ";" parameter )
- # data := *urlchar
- # parameter := attribute "=" value
- import StringIO, mimetools, time
- try:
- [type, data] = string.split(url, ',', 1)
- except ValueError:
- raise IOError, ('data error', 'bad data URL')
- if not type:
- type = 'text/plain;charset=US-ASCII'
- semi = string.rfind(type, ';')
- if semi >= 0 and '=' not in type[semi:]:
- encoding = type[semi+1:]
- type = type[:semi]
- else:
- encoding = ''
- msg = []
- msg.append('Date: %s'%time.strftime('%a, %d %b %Y %T GMT',
- time.gmtime(time.time())))
- msg.append('Content-type: %s' % type)
- if encoding == 'base64':
- import base64
- data = base64.decodestring(data)
- else:
- data = unquote(data)
- msg.append('Content-length: %d' % len(data))
- msg.append('')
- msg.append(data)
- msg = string.join(msg, '\n')
- f = StringIO.StringIO(msg)
- headers = mimetools.Message(f, 0)
- f.fileno = None # needed for addinfourl
- return addinfourl(f, headers, url)
-
-
- # Derived class with handlers for errors we can handle (perhaps)
- class FancyURLopener(URLopener):
-
- def __init__(self, *args):
- apply(URLopener.__init__, (self,) + args)
- self.auth_cache = {}
-
- # Default error handling -- don't raise an exception
- def http_error_default(self, url, fp, errcode, errmsg, headers):
- return addinfourl(fp, headers, "http:" + url)
-
- # Error 302 -- relocated (temporarily)
- def http_error_302(self, url, fp, errcode, errmsg, headers,
- data=None):
- # XXX The server can force infinite recursion here!
- if headers.has_key('location'):
- newurl = headers['location']
- elif headers.has_key('uri'):
- newurl = headers['uri']
- else:
- return
- void = fp.read()
- fp.close()
- # In case the server sent a relative URL, join with original:
- newurl = basejoin("http:" + url, newurl)
- return self.open(newurl, data)
-
- # Error 301 -- also relocated (permanently)
- http_error_301 = http_error_302
-
- # Error 401 -- authentication required
- # See this URL for a description of the basic authentication scheme:
- # http://www.ics.uci.edu/pub/ietf/http/draft-ietf-http-v10-spec-00.txt
- def http_error_401(self, url, fp, errcode, errmsg, headers,
- data=None):
- if headers.has_key('www-authenticate'):
- stuff = headers['www-authenticate']
- import re
- match = re.match('[ \t]*([^ \t]+)[ \t]+realm="([^"]*)"', stuff)
- if match:
- scheme, realm = match.groups()
- if string.lower(scheme) == 'basic':
- return self.retry_http_basic_auth(url, realm, data)
-
- def retry_http_basic_auth(self, url, realm, data):
- host, selector = splithost(url)
- i = string.find(host, '@') + 1
- host = host[i:]
- user, passwd = self.get_user_passwd(host, realm, i)
- if not (user or passwd): return None
- host = user + ':' + passwd + '@' + host
- newurl = 'http://' + host + selector
- return self.open(newurl, data)
-
- def get_user_passwd(self, host, realm, clear_cache = 0):
- key = realm + '@' + string.lower(host)
- if self.auth_cache.has_key(key):
- if clear_cache:
- del self.auth_cache[key]
- else:
- return self.auth_cache[key]
- user, passwd = self.prompt_user_passwd(host, realm)
- if user or passwd: self.auth_cache[key] = (user, passwd)
- return user, passwd
-
- def prompt_user_passwd(self, host, realm):
- # Override this in a GUI environment!
- import getpass
- try:
- user = raw_input("Enter username for %s at %s: " % (realm,
- host))
- passwd = getpass.getpass("Enter password for %s in %s at %s: " %
- (user, realm, host))
- return user, passwd
- except KeyboardInterrupt:
- print
- return None, None
-
-
- # Utility functions
-
- # Return the IP address of the magic hostname 'localhost'
- _localhost = None
- def localhost():
- global _localhost
- if not _localhost:
- _localhost = socket.gethostbyname('localhost')
- return _localhost
-
- # Return the IP address of the current host
- _thishost = None
- def thishost():
- global _thishost
- if not _thishost:
- _thishost = socket.gethostbyname(socket.gethostname())
- return _thishost
-
- # Return the set of errors raised by the FTP class
- _ftperrors = None
- def ftperrors():
- global _ftperrors
- if not _ftperrors:
- import ftplib
- _ftperrors = ftplib.all_errors
- return _ftperrors
-
- # Return an empty mimetools.Message object
- _noheaders = None
- def noheaders():
- global _noheaders
- if not _noheaders:
- import mimetools
- import StringIO
- _noheaders = mimetools.Message(StringIO.StringIO(), 0)
- _noheaders.fp.close() # Recycle file descriptor
- return _noheaders
-
-
- # Utility classes
-
- # Class used by open_ftp() for cache of open FTP connections
- class ftpwrapper:
- def __init__(self, user, passwd, host, port, dirs):
- self.user = user
- self.passwd = passwd
- self.host = host
- self.port = port
- self.dirs = dirs
- self.init()
- def init(self):
- import ftplib
- self.busy = 0
- self.ftp = ftplib.FTP()
- self.ftp.connect(self.host, self.port)
- self.ftp.login(self.user, self.passwd)
- for dir in self.dirs:
- self.ftp.cwd(dir)
- def retrfile(self, file, type):
- import ftplib
- self.endtransfer()
- if type in ('d', 'D'): cmd = 'TYPE A'; isdir = 1
- else: cmd = 'TYPE ' + type; isdir = 0
- try:
- self.ftp.voidcmd(cmd)
- except ftplib.all_errors:
- self.init()
- self.ftp.voidcmd(cmd)
- conn = None
- if file and not isdir:
- # Use nlst to see if the file exists at all
- try:
- self.ftp.nlst(file)
- except ftplib.error_perm, reason:
- raise IOError, ('ftp error', reason), sys.exc_info()[2]
- # Restore the transfer mode!
- self.ftp.voidcmd(cmd)
- # Try to retrieve as a file
- try:
- cmd = 'RETR ' + file
- conn = self.ftp.ntransfercmd(cmd)
- except ftplib.error_perm, reason:
- if reason[:3] != '550':
- raise IOError, ('ftp error', reason), sys.exc_info()[2]
- if not conn:
- # Set transfer mode to ASCII!
- self.ftp.voidcmd('TYPE A')
- # Try a directory listing
- if file: cmd = 'LIST ' + file
- else: cmd = 'LIST'
- conn = self.ftp.ntransfercmd(cmd)
- self.busy = 1
- # Pass back both a suitably decorated object and a retrieval length
- return (addclosehook(conn[0].makefile('rb'),
- self.endtransfer), conn[1])
- def endtransfer(self):
- if not self.busy:
- return
- self.busy = 0
- try:
- self.ftp.voidresp()
- except ftperrors():
- pass
- def close(self):
- self.endtransfer()
- try:
- self.ftp.close()
- except ftperrors():
- pass
-
- # Base class for addinfo and addclosehook
- class addbase:
- def __init__(self, fp):
- self.fp = fp
- self.read = self.fp.read
- self.readline = self.fp.readline
- self.readlines = self.fp.readlines
- self.fileno = self.fp.fileno
- def __repr__(self):
- return '<%s at %s whose fp = %s>' % (self.__class__.__name__,
- `id(self)`, `self.fp`)
- def close(self):
- self.read = None
- self.readline = None
- self.readlines = None
- self.fileno = None
- if self.fp: self.fp.close()
- self.fp = None
-
- # Class to add a close hook to an open file
- class addclosehook(addbase):
- def __init__(self, fp, closehook, *hookargs):
- addbase.__init__(self, fp)
- self.closehook = closehook
- self.hookargs = hookargs
- def close(self):
- if self.closehook:
- apply(self.closehook, self.hookargs)
- self.closehook = None
- self.hookargs = None
- addbase.close(self)
-
- # class to add an info() method to an open file
- class addinfo(addbase):
- def __init__(self, fp, headers):
- addbase.__init__(self, fp)
- self.headers = headers
- def info(self):
- return self.headers
-
- # class to add info() and geturl() methods to an open file
- class addinfourl(addbase):
- def __init__(self, fp, headers, url):
- addbase.__init__(self, fp)
- self.headers = headers
- self.url = url
- def info(self):
- return self.headers
- def geturl(self):
- return self.url
-
-
- # Utility to combine a URL with a base URL to form a new URL
-
- def basejoin(base, url):
- type, path = splittype(url)
- if type:
- # if url is complete (i.e., it contains a type), return it
- return url
- host, path = splithost(path)
- type, basepath = splittype(base) # inherit type from base
- if host:
- # if url contains host, just inherit type
- if type: return type + '://' + host + path
- else:
- # no type inherited, so url must have started with //
- # just return it
- return url
- host, basepath = splithost(basepath) # inherit host
- basepath, basetag = splittag(basepath) # remove extraneuous cruft
- basepath, basequery = splitquery(basepath) # idem
- if path[:1] != '/':
- # non-absolute path name
- if path[:1] in ('#', '?'):
- # path is just a tag or query, attach to basepath
- i = len(basepath)
- else:
- # else replace last component
- i = string.rfind(basepath, '/')
- if i < 0:
- # basepath not absolute
- if host:
- # host present, make absolute
- basepath = '/'
- else:
- # else keep non-absolute
- basepath = ''
- else:
- # remove last file component
- basepath = basepath[:i+1]
- # Interpret ../ (important because of symlinks)
- while basepath and path[:3] == '../':
- path = path[3:]
- i = string.rfind(basepath[:-1], '/')
- if i > 0:
- basepath = basepath[:i+1]
- elif i == 0:
- basepath = '/'
- break
- else:
- basepath = ''
-
- path = basepath + path
- if type and host: return type + '://' + host + path
- elif type: return type + ':' + path
- elif host: return '//' + host + path # don't know what this means
- else: return path
-
-
- # Utilities to parse URLs (most of these return None for missing parts):
- # unwrap('<URL:type://host/path>') --> 'type://host/path'
- # splittype('type:opaquestring') --> 'type', 'opaquestring'
- # splithost('//host[:port]/path') --> 'host[:port]', '/path'
- # splituser('user[:passwd]@host[:port]') --> 'user[:passwd]', 'host[:port]'
- # splitpasswd('user:passwd') -> 'user', 'passwd'
- # splitport('host:port') --> 'host', 'port'
- # splitquery('/path?query') --> '/path', 'query'
- # splittag('/path#tag') --> '/path', 'tag'
- # splitattr('/path;attr1=value1;attr2=value2;...') ->
- # '/path', ['attr1=value1', 'attr2=value2', ...]
- # splitvalue('attr=value') --> 'attr', 'value'
- # splitgophertype('/Xselector') --> 'X', 'selector'
- # unquote('abc%20def') -> 'abc def'
- # quote('abc def') -> 'abc%20def')
-
- def unwrap(url):
- url = string.strip(url)
- if url[:1] == '<' and url[-1:] == '>':
- url = string.strip(url[1:-1])
- if url[:4] == 'URL:': url = string.strip(url[4:])
- return url
-
- _typeprog = None
- def splittype(url):
- global _typeprog
- if _typeprog is None:
- import re
- _typeprog = re.compile('^([^/:]+):')
-
- match = _typeprog.match(url)
- if match:
- scheme = match.group(1)
- return scheme, url[len(scheme) + 1:]
- return None, url
-
- _hostprog = None
- def splithost(url):
- global _hostprog
- if _hostprog is None:
- import re
- _hostprog = re.compile('^//([^/]+)(.*)$')
-
- match = _hostprog.match(url)
- if match: return match.group(1, 2)
- return None, url
-
- _userprog = None
- def splituser(host):
- global _userprog
- if _userprog is None:
- import re
- _userprog = re.compile('^([^@]*)@(.*)$')
-
- match = _userprog.match(host)
- if match: return match.group(1, 2)
- return None, host
-
- _passwdprog = None
- def splitpasswd(user):
- global _passwdprog
- if _passwdprog is None:
- import re
- _passwdprog = re.compile('^([^:]*):(.*)$')
-
- match = _passwdprog.match(user)
- if match: return match.group(1, 2)
- return user, None
-
- _portprog = None
- def splitport(host):
- global _portprog
- if _portprog is None:
- import re
- _portprog = re.compile('^(.*):([0-9]+)$')
-
- match = _portprog.match(host)
- if match: return match.group(1, 2)
- return host, None
-
- # Split host and port, returning numeric port.
- # Return given default port if no ':' found; defaults to -1.
- # Return numerical port if a valid number are found after ':'.
- # Return None if ':' but not a valid number.
- _nportprog = None
- def splitnport(host, defport=-1):
- global _nportprog
- if _nportprog is None:
- import re
- _nportprog = re.compile('^(.*):(.*)$')
-
- match = _nportprog.match(host)
- if match:
- host, port = match.group(1, 2)
- try:
- if not port: raise string.atoi_error, "no digits"
- nport = string.atoi(port)
- except string.atoi_error:
- nport = None
- return host, nport
- return host, defport
-
- _queryprog = None
- def splitquery(url):
- global _queryprog
- if _queryprog is None:
- import re
- _queryprog = re.compile('^(.*)\?([^?]*)$')
-
- match = _queryprog.match(url)
- if match: return match.group(1, 2)
- return url, None
-
- _tagprog = None
- def splittag(url):
- global _tagprog
- if _tagprog is None:
- import re
- _tagprog = re.compile('^(.*)#([^#]*)$')
-
- match = _tagprog.match(url)
- if match: return match.group(1, 2)
- return url, None
-
- def splitattr(url):
- words = string.splitfields(url, ';')
- return words[0], words[1:]
-
- _valueprog = None
- def splitvalue(attr):
- global _valueprog
- if _valueprog is None:
- import re
- _valueprog = re.compile('^([^=]*)=(.*)$')
-
- match = _valueprog.match(attr)
- if match: return match.group(1, 2)
- return attr, None
-
- def splitgophertype(selector):
- if selector[:1] == '/' and selector[1:2]:
- return selector[1], selector[2:]
- return None, selector
-
- def unquote(s):
- mychr = chr
- myatoi = string.atoi
- list = string.split(s, '%')
- res = [list[0]]
- myappend = res.append
- del list[0]
- for item in list:
- if item[1:2]:
- try:
- myappend(mychr(myatoi(item[:2], 16))
- + item[2:])
- except:
- myappend('%' + item)
- else:
- myappend('%' + item)
- return string.join(res, "")
-
- def unquote_plus(s):
- if '+' in s:
- # replace '+' with ' '
- s = string.join(string.split(s, '+'), ' ')
- return unquote(s)
-
- always_safe = string.letters + string.digits + '_,.-'
- def quote(s, safe = '/'):
- safe = always_safe + safe
- res = list(s)
- for i in range(len(res)):
- c = res[i]
- if c not in safe:
- res[i] = '%%%02x' % ord(c)
- return string.joinfields(res, '')
-
- def quote_plus(s, safe = '/'):
- if ' ' in s:
- # replace ' ' with '+'
- l = string.split(s, ' ')
- for i in range(len(l)):
- l[i] = quote(l[i], safe)
- return string.join(l, '+')
- else:
- return quote(s, safe)
-
- def urlencode(dict):
- l = []
- for k, v in dict.items():
- k = quote_plus(str(k))
- v = quote_plus(str(v))
- l.append(k + '=' + v)
- return string.join(l, '&')
-
-
- # Proxy handling
- if os.name == 'mac':
- def getproxies():
- """Return a dictionary of scheme -> proxy server URL mappings.
-
- By convention the mac uses Internet Config to store
- proxies. An HTTP proxy, for instance, is stored under
- the HttpProxy key.
-
- """
- try:
- import ic
- except ImportError:
- return {}
-
- try:
- config = ic.IC()
- except ic.error:
- return {}
- proxies = {}
- # HTTP:
- if config.has_key('UseHTTPProxy') and config['UseHTTPProxy']:
- try:
- value = config['HTTPProxyHost']
- except ic.error:
- pass
- else:
- proxies['http'] = 'http://%s' % value
- # FTP: XXXX To be done.
- # Gopher: XXXX To be done.
- return proxies
-
- else:
- def getproxies():
- """Return a dictionary of scheme -> proxy server URL mappings.
-
- Scan the environment for variables named <scheme>_proxy;
- this seems to be the standard convention. If you need a
- different way, you can pass a proxies dictionary to the
- [Fancy]URLopener constructor.
-
- """
- proxies = {}
- for name, value in os.environ.items():
- name = string.lower(name)
- if value and name[-6:] == '_proxy':
- proxies[name[:-6]] = value
- return proxies
-
-
- # Test and time quote() and unquote()
- def test1():
- import time
- s = ''
- for i in range(256): s = s + chr(i)
- s = s*4
- t0 = time.time()
- qs = quote(s)
- uqs = unquote(qs)
- t1 = time.time()
- if uqs != s:
- print 'Wrong!'
- print `s`
- print `qs`
- print `uqs`
- print round(t1 - t0, 3), 'sec'
-
-
- def reporthook(blocknum, blocksize, totalsize):
- # Report during remote transfers
- print "Block number: %d, Block size: %d, Total size: %d" % (blocknum, blocksize, totalsize)
-
- # Test program
- def test(args=[]):
- if not args:
- args = [
- '/etc/passwd',
- 'file:/etc/passwd',
- 'file://localhost/etc/passwd',
- 'ftp://ftp.python.org/etc/passwd',
- ## 'gopher://gopher.micro.umn.edu/1/',
- 'http://www.python.org/index.html',
- ]
- try:
- for url in args:
- print '-'*10, url, '-'*10
- fn, h = urlretrieve(url, None, reporthook)
- print fn, h
- if h:
- print '======'
- for k in h.keys(): print k + ':', h[k]
- print '======'
- fp = open(fn, 'rb')
- data = fp.read()
- del fp
- if '\r' in data:
- table = string.maketrans("", "")
- data = string.translate(data, table, "\r")
- print data
- fn, h = None, None
- print '-'*40
- finally:
- urlcleanup()
-
- def main():
- import getopt, sys
- try:
- opts, args = getopt.getopt(sys.argv[1:], "th")
- except getopt.error, msg:
- print msg
- print "Use -h for help"
- return
- t = 0
- for o, a in opts:
- if o == '-t':
- t = t + 1
- if o == '-h':
- print "Usage: python urllib.py [-t] [url ...]"
- print "-t runs self-test;",
- print "otherwise, contents of urls are printed"
- return
- if t:
- if t > 1:
- test1()
- test(args)
- else:
- if not args:
- print "Use -h for help"
- for url in args:
- print urlopen(url).read(),
-
- # Run test program when run as a script
- if __name__ == '__main__':
- main()
-