"""Download files with progress indicators.
|
"""
|
import cgi
|
import logging
|
import mimetypes
|
import os
|
|
from pip._vendor.requests.models import CONTENT_CHUNK_SIZE
|
|
from pip._internal.cli.progress_bars import DownloadProgressProvider
|
from pip._internal.exceptions import NetworkConnectionError
|
from pip._internal.models.index import PyPI
|
from pip._internal.network.cache import is_from_cache
|
from pip._internal.network.utils import (
|
HEADERS,
|
raise_for_status,
|
response_chunks,
|
)
|
from pip._internal.utils.misc import (
|
format_size,
|
redact_auth_from_url,
|
splitext,
|
)
|
from pip._internal.utils.typing import MYPY_CHECK_RUNNING
|
|
if MYPY_CHECK_RUNNING:
|
from typing import Iterable, Optional
|
|
from pip._vendor.requests.models import Response
|
|
from pip._internal.models.link import Link
|
from pip._internal.network.session import PipSession
|
|
logger = logging.getLogger(__name__)
|
|
|
def _get_http_response_size(resp):
|
# type: (Response) -> Optional[int]
|
try:
|
return int(resp.headers['content-length'])
|
except (ValueError, KeyError, TypeError):
|
return None
|
|
|
def _prepare_download(
|
resp, # type: Response
|
link, # type: Link
|
progress_bar # type: str
|
):
|
# type: (...) -> Iterable[bytes]
|
total_length = _get_http_response_size(resp)
|
|
if link.netloc == PyPI.file_storage_domain:
|
url = link.show_url
|
else:
|
url = link.url_without_fragment
|
|
logged_url = redact_auth_from_url(url)
|
|
if total_length:
|
logged_url = '{} ({})'.format(logged_url, format_size(total_length))
|
|
if is_from_cache(resp):
|
logger.info("Using cached %s", logged_url)
|
else:
|
logger.info("Downloading %s", logged_url)
|
|
if logger.getEffectiveLevel() > logging.INFO:
|
show_progress = False
|
elif is_from_cache(resp):
|
show_progress = False
|
elif not total_length:
|
show_progress = True
|
elif total_length > (40 * 1000):
|
show_progress = True
|
else:
|
show_progress = False
|
|
chunks = response_chunks(resp, CONTENT_CHUNK_SIZE)
|
|
if not show_progress:
|
return chunks
|
|
return DownloadProgressProvider(
|
progress_bar, max=total_length
|
)(chunks)
|
|
|
def sanitize_content_filename(filename):
|
# type: (str) -> str
|
"""
|
Sanitize the "filename" value from a Content-Disposition header.
|
"""
|
return os.path.basename(filename)
|
|
|
def parse_content_disposition(content_disposition, default_filename):
|
# type: (str, str) -> str
|
"""
|
Parse the "filename" value from a Content-Disposition header, and
|
return the default filename if the result is empty.
|
"""
|
_type, params = cgi.parse_header(content_disposition)
|
filename = params.get('filename')
|
if filename:
|
# We need to sanitize the filename to prevent directory traversal
|
# in case the filename contains ".." path parts.
|
filename = sanitize_content_filename(filename)
|
return filename or default_filename
|
|
|
def _get_http_response_filename(resp, link):
|
# type: (Response, Link) -> str
|
"""Get an ideal filename from the given HTTP response, falling back to
|
the link filename if not provided.
|
"""
|
filename = link.filename # fallback
|
# Have a look at the Content-Disposition header for a better guess
|
content_disposition = resp.headers.get('content-disposition')
|
if content_disposition:
|
filename = parse_content_disposition(content_disposition, filename)
|
ext = splitext(filename)[1] # type: Optional[str]
|
if not ext:
|
ext = mimetypes.guess_extension(
|
resp.headers.get('content-type', '')
|
)
|
if ext:
|
filename += ext
|
if not ext and link.url != resp.url:
|
ext = os.path.splitext(resp.url)[1]
|
if ext:
|
filename += ext
|
return filename
|
|
|
def _http_get_download(session, link):
|
# type: (PipSession, Link) -> Response
|
target_url = link.url.split('#', 1)[0]
|
resp = session.get(target_url, headers=HEADERS, stream=True)
|
raise_for_status(resp)
|
return resp
|
|
|
class Download(object):
|
def __init__(
|
self,
|
response, # type: Response
|
filename, # type: str
|
chunks, # type: Iterable[bytes]
|
):
|
# type: (...) -> None
|
self.response = response
|
self.filename = filename
|
self.chunks = chunks
|
|
|
class Downloader(object):
|
def __init__(
|
self,
|
session, # type: PipSession
|
progress_bar, # type: str
|
):
|
# type: (...) -> None
|
self._session = session
|
self._progress_bar = progress_bar
|
|
def __call__(self, link):
|
# type: (Link) -> Download
|
try:
|
resp = _http_get_download(self._session, link)
|
except NetworkConnectionError as e:
|
assert e.response is not None
|
logger.critical(
|
"HTTP error %s while getting %s", e.response.status_code, link
|
)
|
raise
|
|
return Download(
|
resp,
|
_get_http_response_filename(resp, link),
|
_prepare_download(resp, link, self._progress_bar),
|
)
|
