"""PyPI and direct package downloading"""
import sys, os.path, re, urlparse, urllib2
from pkg_resources import *
from distutils import log
from distutils.errors import DistutilsError
from md5 import md5
from fnmatch import translate
EGG_FRAGMENT = re.compile(r'^egg=([-A-Za-z0-9_.]+)$')
HREF = re.compile("""href\\s*=\\s*['"]?([^'"> ]+)""", re.I)
# this is here to fix emacs' cruddy broken syntax highlighting
PYPI_MD5 = re.compile(
'([^<]+)\n\s+\\(md5\\)'
)
URL_SCHEME = re.compile('([-+.a-z0-9]{2,}):',re.I).match
EXTENSIONS = ".tar.gz .tar.bz2 .tar .zip .tgz".split()
__all__ = [
'PackageIndex', 'distros_for_url', 'parse_bdist_wininst',
'interpret_distro_name',
]
def parse_bdist_wininst(name):
"""Return (base,pyversion) or (None,None) for possible .exe name"""
lower = name.lower()
base, py_ver = None, None
if lower.endswith('.exe'):
if lower.endswith('.win32.exe'):
base = name[:-10]
elif lower.startswith('.win32-py',-16):
py_ver = name[-7:-4]
base = name[:-16]
return base,py_ver
def distros_for_url(url, metadata=None):
"""Yield egg or source distribution objects that might be found at a URL"""
scheme, server, path, parameters, query, fragment = urlparse.urlparse(url)
base = urllib2.unquote(path.split('/')[-1])
if '#' in base: base, fragment = base.split('#',1)
dists = distros_for_location(url, base, metadata)
if fragment and not dists:
match = EGG_FRAGMENT.match(fragment)
if match:
return interpret_distro_name(
url, match.group(1), metadata, precedence = CHECKOUT_DIST
)
return dists
def distros_for_location(location, basename, metadata=None):
"""Yield egg or source distribution objects based on basename"""
if basename.endswith('.egg.zip'):
basename = basename[:-4] # strip the .zip
if basename.endswith('.egg'): # only one, unambiguous interpretation
return [Distribution.from_location(location, basename, metadata)]
if basename.endswith('.exe'):
win_base, py_ver = parse_bdist_wininst(basename)
if win_base is not None:
return interpret_distro_name(
location, win_base, metadata, py_ver, BINARY_DIST, "win32"
)
# Try source distro extensions (.zip, .tgz, etc.)
#
for ext in EXTENSIONS:
if basename.endswith(ext):
basename = basename[:-len(ext)]
return interpret_distro_name(location, basename, metadata)
return [] # no extension matched
def distros_for_filename(filename, metadata=None):
"""Yield possible egg or source distribution objects based on a filename"""
return distros_for_location(
normalize_path(filename), os.path.basename(filename), metadata
)
def interpret_distro_name(location, basename, metadata,
py_version=None, precedence=SOURCE_DIST, platform=None
):
"""Generate alternative interpretations of a source distro name
Note: if `location` is a filesystem filename, you should call
``pkg_resources.normalize_path()`` on it before passing it to this
routine!
"""
# Generate alternative interpretations of a source distro name
# Because some packages are ambiguous as to name/versions split
# e.g. "adns-python-1.1.0", "egenix-mx-commercial", etc.
# So, we generate each possible interepretation (e.g. "adns, python-1.1.0"
# "adns-python, 1.1.0", and "adns-python-1.1.0, no version"). In practice,
# the spurious interpretations should be ignored, because in the event
# there's also an "adns" package, the spurious "python-1.1.0" version will
# compare lower than any numeric version number, and is therefore unlikely
# to match a request for it. It's still a potential problem, though, and
# in the long run PyPI and the distutils should go for "safe" names and
# versions in distribution archive names (sdist and bdist).
parts = basename.split('-')
for p in range(1,len(parts)+1):
yield Distribution(
location, metadata, '-'.join(parts[:p]), '-'.join(parts[p:]),
py_version=py_version, precedence = precedence,
platform = platform
)
class PackageIndex(Environment):
"""A distribution index that scans web pages for download URLs"""
def __init__(self,index_url="http://www.python.org/pypi",hosts=('*',),*args,**kw):
Environment.__init__(self,*args,**kw)
self.index_url = index_url + "/"[:not index_url.endswith('/')]
self.scanned_urls = {}
self.fetched_urls = {}
self.package_pages = {}
self.allows = re.compile('|'.join(map(translate,hosts))).match
def process_url(self, url, retrieve=False):
"""Evaluate a URL as a possible download, and maybe retrieve it"""
if url in self.scanned_urls and not retrieve:
return
self.scanned_urls[url] = True
if not URL_SCHEME(url):
# process filenames or directories
if os.path.isfile(url):
map(self.add, distros_for_filename(url))
return # no need to retrieve anything
elif os.path.isdir(url):
url = os.path.realpath(url)
for item in os.listdir(url):
self.process_url(os.path.join(url,item))
return
else:
self.warn("Not found: %s", url)
return
else:
dists = list(distros_for_url(url))
if dists:
if not self.url_ok(url):
return
self.debug("Found link: %s", url)
if dists or not retrieve or url in self.fetched_urls:
map(self.add, dists)
return # don't need the actual page
if not self.url_ok(url):
self.fetched_urls[url] = True
return
self.info("Reading %s", url)
f = self.open_url(url)
self.fetched_urls[url] = self.fetched_urls[f.url] = True
if 'html' not in f.headers['content-type'].lower():
f.close() # not html, we can't process it
return
base = f.url # handle redirects
page = f.read()
f.close()
if url.startswith(self.index_url):
page = self.process_index(url, page)
for match in HREF.finditer(page):
link = urlparse.urljoin(base, match.group(1))
self.process_url(link)
def url_ok(self, url, fatal=False):
if self.allows(urlparse.urlparse(url)[1]):
return True
msg = "\nLink to % s ***BLOCKED*** by --allow-hosts\n"
if fatal:
raise DistutilsError(msg % url)
else:
self.warn(msg, url)
def process_index(self,url,page):
"""Process the contents of a PyPI page"""
def scan(link):
# Process a URL to see if it's for a package page
if link.startswith(self.index_url):
parts = map(
urllib2.unquote, link[len(self.index_url):].split('/')
)
if len(parts)==2:
# it's a package page, sanitize and index it
pkg = safe_name(parts[0])
ver = safe_version(parts[1])
self.package_pages.setdefault(pkg.lower(),{})[link] = True
if url==self.index_url or 'Index of Packages' in page:
# process an index page into the package-page index
for match in HREF.finditer(page):
scan( urlparse.urljoin(url, match.group(1)) )
else:
scan(url) # ensure this page is in the page index
# process individual package page
for tag in ("| Home Page", " | Download URL"):
pos = page.find(tag)
if pos!=-1:
match = HREF.search(page,pos)
if match:
# Process the found URL
self.scan_url(urlparse.urljoin(url, match.group(1)))
return PYPI_MD5.sub(
lambda m: '%s' % m.group(1,3,2), page
)
def find_packages(self, requirement):
self.scan_url(self.index_url + requirement.project_name+'/')
if not self.package_pages.get(requirement.key):
# We couldn't find the target package, so search the index page too
self.warn(
"Couldn't find index page for %r (maybe misspelled?)",
requirement.project_name
)
if self.index_url not in self.fetched_urls:
self.warn(
"Scanning index of all packages (this may take a while)"
)
self.scan_url(self.index_url)
for url in self.package_pages.get(requirement.key,()):
# scan each page that might be related to the desired package
self.scan_url(url)
def obtain(self, requirement, installer=None):
self.find_packages(requirement)
for dist in self[requirement.key]:
if dist in requirement:
return dist
self.debug("%s does not match %s", requirement, dist)
return super(PackageIndex, self).obtain(requirement,installer)
def check_md5(self, cs, info, filename, tfp):
if re.match('md5=[0-9a-f]{32}$', info):
self.debug("Validating md5 checksum for %s", filename)
if cs.hexdigest()<>info[4:]:
tfp.close()
os.unlink(filename)
raise DistutilsError(
"MD5 validation failed for "+os.path.basename(filename)+
"; possible download problem?"
)
def download(self, spec, tmpdir):
"""Locate and/or download `spec` to `tmpdir`, returning a local path
`spec` may be a ``Requirement`` object, or a string containing a URL,
an existing local filename, or a project/version requirement spec
(i.e. the string form of a ``Requirement`` object).
If `spec` is a ``Requirement`` object or a string containing a
project/version requirement spec, this method is equivalent to
the ``fetch()`` method. If `spec` is a local, existing file or
directory name, it is simply returned unchanged. If `spec` is a URL,
it is downloaded to a subpath of `tmpdir`, and the local filename is
returned. Various errors may be raised if a problem occurs during
downloading.
"""
if not isinstance(spec,Requirement):
scheme = URL_SCHEME(spec)
if scheme:
# It's a url, download it to tmpdir
return self._download_url(scheme.group(1), spec, tmpdir)
elif os.path.exists(spec):
# Existing file or directory, just return it
return spec
else:
try:
spec = Requirement.parse(spec)
except ValueError:
raise DistutilsError(
"Not a URL, existing file, or requirement spec: %r" %
(spec,)
)
return self.fetch(spec, tmpdir)
def fetch(self, requirement, tmpdir, force_scan=False, source=False):
"""Obtain a file suitable for fulfilling `requirement`
`requirement` must be a ``pkg_resources.Requirement`` instance.
If necessary, or if the `force_scan` flag is set, the requirement is
searched for in the (online) package index as well as the locally
installed packages. If a distribution matching `requirement` is found,
the return value is the same as if you had called the ``download()``
method with the matching distribution's URL. If no matching
distribution is found, returns ``None``.
If the `source` flag is set, only source distributions and source
checkout links will be considered.
"""
# process a Requirement
self.info("Searching for %s", requirement)
def find(req):
for dist in self[req.key]:
if dist in req and (dist.precedence<=SOURCE_DIST or not source):
self.info("Best match: %s", dist)
return self.download(dist.location, tmpdir)
if force_scan:
self.find_packages(requirement)
dist = find(requirement)
else:
dist = find(requirement)
if dist is None:
self.find_packages(requirement)
dist = find(requirement)
if dist is None:
self.warn(
"No local packages or download links found for %s%s",
(source and "a source distribution of " or ""),
requirement,
)
return dist
dl_blocksize = 8192
def _download_to(self, url, filename):
self.url_ok(url,True) # raises error if not allowed
self.info("Downloading %s", url)
# Download the file
fp, tfp, info = None, None, None
try:
if '#' in url:
url, info = url.split('#', 1)
fp = self.open_url(url)
if isinstance(fp, urllib2.HTTPError):
raise DistutilsError(
"Can't download %s: %s %s" % (url, fp.code,fp.msg)
)
cs = md5()
headers = fp.info()
blocknum = 0
bs = self.dl_blocksize
size = -1
if "content-length" in headers:
size = int(headers["Content-Length"])
self.reporthook(url, filename, blocknum, bs, size)
tfp = open(filename,'wb')
while True:
block = fp.read(bs)
if block:
cs.update(block)
tfp.write(block)
blocknum += 1
self.reporthook(url, filename, blocknum, bs, size)
else:
break
if info: self.check_md5(cs, info, filename, tfp)
return headers
finally:
if fp: fp.close()
if tfp: tfp.close()
def reporthook(self, url, filename, blocknum, blksize, size):
pass # no-op
def open_url(self, url):
try:
return urllib2.urlopen(url)
except urllib2.HTTPError, v:
return v
except urllib2.URLError, v:
raise DistutilsError("Download error: %s" % v.reason)
def _download_url(self, scheme, url, tmpdir):
# Determine download filename
#
name = filter(None,urlparse.urlparse(url)[2].split('/'))
if name:
name = name[-1]
while '..' in name:
name = name.replace('..','.').replace('\\','_')
else:
name = "__downloaded__" # default if URL has no path contents
if name.endswith('.egg.zip'):
name = name[:-4] # strip the extra .zip before download
filename = os.path.join(tmpdir,name)
# Download the file
#
if scheme=='svn' or scheme.startswith('svn+'):
return self._download_svn(url, filename)
else:
headers = self._download_to(url, filename)
if 'html' in headers['content-type'].lower():
return self._download_html(url, headers, filename, tmpdir)
else:
return filename
def scan_url(self, url):
self.process_url(url, True)
def _download_html(self, url, headers, filename, tmpdir):
# Check for a sourceforge URL
sf_url = url.startswith('http://prdownloads.')
file = open(filename)
for line in file:
if line.strip():
# Check for a subversion index page
if re.search(r'Revision \d+:', line):
# it's a subversion index page:
file.close()
os.unlink(filename)
return self._download_svn(url, filename)
# Check for a SourceForge header
elif sf_url:
page = ''.join(list(file))
if '?use_mirror=' in page:
file.close()
os.unlink(filename)
return self._download_sourceforge(url, page, tmpdir)
break # not an index page
file.close()
os.unlink(filename)
raise DistutilsError("Unexpected HTML page found at "+url)
def _download_svn(self, url, filename):
url = url.split('#',1)[0] # remove any fragment for svn's sake
self.info("Doing subversion checkout from %s to %s", url, filename)
os.system("svn checkout -q %s %s" % (url, filename))
return filename
def debug(self, msg, *args):
log.debug(msg, *args)
def info(self, msg, *args):
log.info(msg, *args)
def warn(self, msg, *args):
log.warn(msg, *args)
def _download_sourceforge(self, source_url, sf_page, tmpdir):
"""Download package from randomly-selected SourceForge mirror"""
self.debug("Processing SourceForge mirror page")
mirror_regex = re.compile(r'HREF="?(/.*?\?use_mirror=[^">]*)', re.I)
urls = [m.group(1) for m in mirror_regex.finditer(sf_page)]
if not urls:
raise DistutilsError(
"URL looks like a Sourceforge mirror page, but no URLs found"
)
import random
url = urlparse.urljoin(source_url, random.choice(urls))
self.info(
"Requesting redirect to (randomly selected) %r mirror",
url.split('=',1)[-1]
)
f = self.open_url(url)
match = re.search(
r'<META HTTP-EQUIV="refresh" content=".*?URL=(.*?)"',
f.read()
)
f.close()
if match:
download_url = match.group(1)
scheme = URL_SCHEME(download_url)
return self._download_url(scheme.group(1), download_url, tmpdir)
else:
raise DistutilsError(
'No META HTTP-EQUIV="refresh" found in Sourceforge page at %s'
% url
)
|