import logging
from email.message import Message
from email.parser import Parser
from typing import Tuple
from zipfile import BadZipFile, ZipFile
from pip._vendor.packaging.utils import canonicalize_name
from pip._internal.exceptions import UnsupportedWheel
VERSION_COMPATIBLE = (1, 0)
logger = logging.getLogger(__name__)
def parse_wheel(wheel_zip: ZipFile, name: str) -> Tuple[str, Message]:
try:
info_dir = wheel_dist_info_dir(wheel_zip, name)
metadata = wheel_metadata(wheel_zip, info_dir)
version = wheel_version(metadata)
except UnsupportedWheel as e:
raise UnsupportedWheel(f"{name} has an invalid wheel, {str(e)}")
check_compatibility(version, name)
return info_dir, metadata
def wheel_dist_info_dir(source: ZipFile, name: str) -> str:
subdirs = {p.split("/", 1)[0] for p in source.namelist()}
info_dirs = [s for s in subdirs if s.endswith(".dist-info")]
if not info_dirs:
raise UnsupportedWheel(".dist-info directory not found")
if len(info_dirs) > 1:
raise UnsupportedWheel(
"multiple .dist-info directories found: {}".format(", ".join(info_dirs))
)
info_dir = info_dirs[0]
info_dir_name = canonicalize_name(info_dir)
canonical_name = canonicalize_name(name)
if not info_dir_name.startswith(canonical_name):
raise UnsupportedWheel(
f".dist-info directory {info_dir!r} does not start with {canonical_name!r}"
)
return info_dir
def read_wheel_metadata_file(source: ZipFile, path: str) -> bytes:
try:
return source.read(path)
except (BadZipFile, KeyError, RuntimeError) as e:
raise UnsupportedWheel(f"could not read {path!r} file: {e!r}")
def wheel_metadata(source: ZipFile, dist_info_dir: str) -> Message:
path = f"{dist_info_dir}/WHEEL"
wheel_contents = read_wheel_metadata_file(source, path)
try:
wheel_text = wheel_contents.decode()
except UnicodeDecodeError as e:
raise UnsupportedWheel(f"error decoding {path!r}: {e!r}")
return Parser().parsestr(wheel_text)
def wheel_version(wheel_data: Message) -> Tuple[int, ...]:
version_text = wheel_data["Wheel-Version"]
if version_text is None:
raise UnsupportedWheel("WHEEL is missing Wheel-Version")
version = version_text.strip()
try:
return tuple(map(int, version.split(".")))
except ValueError:
raise UnsupportedWheel(f"invalid Wheel-Version: {version!r}")
def check_compatibility(version: Tuple[int, ...], name: str) -> None:
if version[0] > VERSION_COMPATIBLE[0]:
raise UnsupportedWheel(
"{}'s Wheel-Version ({}) is not compatible with this version "
"of pip".format(name, ".".join(map(str, version)))
)
elif version > VERSION_COMPATIBLE:
logger.warning(
"Installing from a newer Wheel-Version (%s)",
".".join(map(str, version)),
)