Source code for

"""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 import is_from_cache
from import (
from pip._internal.utils.misc import (
from pip._internal.utils.typing import MYPY_CHECK_RUNNING

    from typing import Iterable, Optional

    from pip._vendor.requests.models import Response

    from import Link
    from import PipSession

logger = logging.getLogger(__name__)

def _get_http_response_size(resp):
    # type: (Response) -> Optional[int]
        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
        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):"Using cached %s", logged_url)
    else:"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
        show_progress = False

    chunks = response_chunks(resp, CONTENT_CHUNK_SIZE)

    if not show_progress:
        return chunks

    return DownloadProgressProvider(
        progress_bar, max=total_length

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)
    return resp

class Download(object):
    def __init__(
        response,  # type: Response
        filename,  # type: str
        chunks,  # type: Iterable[bytes]
        # type: (...) -> None
        self.response = response
        self.filename = filename
        self.chunks = chunks

[docs]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), )