# SPDX-FileCopyrightText: 2015 Eric Larson
#
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations

import logging
from argparse import ArgumentParser
from typing import TYPE_CHECKING

from pip._vendor import requests

from pip._vendor.cachecontrol.adapter import CacheControlAdapter
from pip._vendor.cachecontrol.cache import DictCache
from pip._vendor.cachecontrol.controller import logger

if TYPE_CHECKING:
    from argparse import Namespace

    from pip._vendor.cachecontrol.controller import CacheController


def setup_logging() -> None:
    logger.setLevel(logging.DEBUG)
    handler = logging.StreamHandler()
    logger.addHandler(handler)


def get_session() -> requests.Session:
    adapter = CacheControlAdapter(
        DictCache(), cache_etags=True, serializer=None, heuristic=None
    )
    sess = requests.Session()
    sess.mount("http://", adapter)
    sess.mount("https://", adapter)

    sess.cache_controller = adapter.controller  # type: ignore[attr-defined]
    return sess


def get_args() -> Namespace:
    parser = ArgumentParser()
    parser.add_argument("url", help="The URL to try and cache")
    return parser.parse_args()


def main() -> None:
    args = get_args()
    sess = get_session()

    # Make a request to get a response
    resp = sess.get(args.url)

    # Turn on logging
    setup_logging()

    # try setting the cache
    cache_controller: CacheController = (
        sess.cache_controller  # type: ignore[attr-defined]
    )
    cache_controller.cache_response(resp.request, resp.raw)

    # Now try to get it
    if cache_controller.cached_request(resp.request):
        print("Cached!")
    else:
        print("Not cached :(")


if __name__ == "__main__":
    main()

from typing import List, TypeVar

T = TypeVar("T")


class Stack(List[T]):
    """A small shim over builtin list."""

    @property
    def top(self) -> T:
        """Get top of stack."""
        return self[-1]

    def push(self, item: T) -> None:
        """Push an item on to the stack (append in stack nomenclature)."""
        self.append(item)

"""
    pygments.plugin
    ~~~~~~~~~~~~~~~

    Pygments plugin interface.

    lexer plugins::

        [pygments.lexers]
        yourlexer = yourmodule:YourLexer

    formatter plugins::

        [pygments.formatters]
        yourformatter = yourformatter:YourFormatter
        /.ext = yourformatter:YourFormatter

    As you can see, you can define extensions for the formatter
    with a leading slash.

    syntax plugins::

        [pygments.styles]
        yourstyle = yourstyle:YourStyle

    filter plugin::

        [pygments.filter]
        yourfilter = yourfilter:YourFilter


    :copyright: Copyright 2006-2024 by the Pygments team, see AUTHORS.
    :license: BSD, see LICENSE for details.
"""
from importlib.metadata import entry_points

LEXER_ENTRY_POINT = 'pygments.lexers'
FORMATTER_ENTRY_POINT = 'pygments.formatters'
STYLE_ENTRY_POINT = 'pygments.styles'
FILTER_ENTRY_POINT = 'pygments.filters'


def iter_entry_points(group_name):
    groups = entry_points()
    if hasattr(groups, 'select'):
        # New interface in Python 3.10 and newer versions of the
        # importlib_metadata backport.
        return groups.select(group=group_name)
    else:
        # Older interface, deprecated in Python 3.10 and recent
        # importlib_metadata, but we need it in Python 3.8 and 3.9.
        return groups.get(group_name, [])


def find_plugin_lexers():
    for entrypoint in iter_entry_points(LEXER_ENTRY_POINT):
        yield entrypoint.load()


def find_plugin_formatters():
    for entrypoint in iter_entry_points(FORMATTER_ENTRY_POINT):
        yield entrypoint.name, entrypoint.load()


def find_plugin_styles():
    for entrypoint in iter_entry_points(STYLE_ENTRY_POINT):
        yield entrypoint.name, entrypoint.load()


def find_plugin_filters():
    for entrypoint in iter_entry_points(FILTER_ENTRY_POINT):
        yield entrypoint.name, entrypoint.load()

"""Handles all VCS (version control) support"""

import logging
import os
import shutil
import sys
import urllib.parse
from dataclasses import dataclass, field
from typing import (
    Any,
    Dict,
    Iterable,
    Iterator,
    List,
    Literal,
    Mapping,
    Optional,
    Tuple,
    Type,
    Union,
)

from pip._internal.cli.spinners import SpinnerInterface
from pip._internal.exceptions import BadCommand, InstallationError
from pip._internal.utils.misc import (
    HiddenText,
    ask_path_exists,
    backup_dir,
    display_path,
    hide_url,
    hide_value,
    is_installable_dir,
    rmtree,
)
from pip._internal.utils.subprocess import (
    CommandArgs,
    call_subprocess,
    format_command_args,
    make_command,
)

__all__ = ["vcs"]


logger = logging.getLogger(__name__)

AuthInfo = Tuple[Optional[str], Optional[str]]


def is_url(name: str) -> bool:
    """
    Return true if the name looks like a URL.
    """
    scheme = urllib.parse.urlsplit(name).scheme
    if not scheme:
        return False
    return scheme in ["http", "https", "file", "ftp"] + vcs.all_schemes


def make_vcs_requirement_url(
    repo_url: str, rev: str, project_name: str, subdir: Optional[str] = None
) -> str:
    """
    Return the URL for a VCS requirement.

    Args:
      repo_url: the remote VCS url, with any needed VCS prefix (e.g. "git+").
      project_name: the (unescaped) project name.
    """
    egg_project_name = project_name.replace("-", "_")
    req = f"{repo_url}@{rev}#egg={egg_project_name}"
    if subdir:
        req += f"&subdirectory={subdir}"

    return req


def find_path_to_project_root_from_repo_root(
    location: str, repo_root: str
) -> Optional[str]:
    """
    Find the the Python project's root by searching up the filesystem from
    `location`. Return the path to project root relative to `repo_root`.
    Return None if the project root is `repo_root`, or cannot be found.
    """
    # find project root.
    orig_location = location
    while not is_installable_dir(location):
        last_location = location
        location = os.path.dirname(location)
        if location == last_location:
            # We've traversed up to the root of the filesystem without
            # finding a Python project.
            logger.warning(
                "Could not find a Python project for directory %s (tried all "
                "parent directories)",
                orig_location,
            )
            return None

    if os.path.samefile(repo_root, location):
        return None

    return os.path.relpath(location, repo_root)


class RemoteNotFoundError(Exception):
    pass


class RemoteNotValidError(Exception):
    def __init__(self, url: str):
        super().__init__(url)
        self.url = url


@dataclass(frozen=True)
class RevOptions:
    """
    Encapsulates a VCS-specific revision to install, along with any VCS
    install options.

    Args:
        vc_class: a VersionControl subclass.
        rev: the name of the revision to install.
        extra_args: a list of extra options.
    """

    vc_class: Type["VersionControl"]
    rev: Optional[str] = None
    extra_args: CommandArgs = field(default_factory=list)
    branch_name: Optional[str] = None

    def __repr__(self) -> str:
        return f"<RevOptions {self.vc_class.name}: rev={self.rev!r}>"

    @property
    def arg_rev(self) -> Optional[str]:
        if self.rev is None:
            return self.vc_class.default_arg_rev

        return self.rev

    def to_args(self) -> CommandArgs:
        """
        Return the VCS-specific command arguments.
        """
        args: CommandArgs = []
        rev = self.arg_rev
        if rev is not None:
            args += self.vc_class.get_base_rev_args(rev)
        args += self.extra_args

        return args

    def to_display(self) -> str:
        if not self.rev:
            return ""

        return f" (to revision {self.rev})"

    def make_new(self, rev: str) -> "RevOptions":
        """
        Make a copy of the current instance, but with a new rev.

        Args:
          rev: the name of the revision for the new object.
        """
        return self.vc_class.make_rev_options(rev, extra_args=self.extra_args)


class VcsSupport:
    _registry: Dict[str, "VersionControl"] = {}
    schemes = ["ssh", "git", "hg", "bzr", "sftp", "svn"]

    def __init__(self) -> None:
        # Register more schemes with urlparse for various version control
        # systems
        urllib.parse.uses_netloc.extend(self.schemes)
        super().__init__()

    def __iter__(self) -> Iterator[str]:
        return self._registry.__iter__()

    @property
    def backends(self) -> List["VersionControl"]:
        return list(self._registry.values())

    @property
    def dirnames(self) -> List[str]:
        return [backend.dirname for backend in self.backends]

    @property
    def all_schemes(self) -> List[str]:
        schemes: List[str] = []
        for backend in self.backends:
            schemes.extend(backend.schemes)
        return schemes

    def register(self, cls: Type["VersionControl"]) -> None:
        if not hasattr(cls, "name"):
            logger.warning("Cannot register VCS %s", cls.__name__)
            return
        if cls.name not in self._registry:
            self._registry[cls.name] = cls()
            logger.debug("Registered VCS backend: %s", cls.name)

    def unregister(self, name: str) -> None:
        if name in self._registry:
            del self._registry[name]

    def get_backend_for_dir(self, location: str) -> Optional["VersionControl"]:
        """
        Return a VersionControl object if a repository of that type is found
        at the given directory.
        """
        vcs_backends = {}
        for vcs_backend in self._registry.values():
            repo_path = vcs_backend.get_repository_root(location)
            if not repo_path:
                continue
            logger.debug("Determine that %s uses VCS: %s", location, vcs_backend.name)
            vcs_backends[repo_path] = vcs_backend

        if not vcs_backends:
            return None

        # Choose the VCS in the inner-most directory. Since all repository
        # roots found here would be either `location` or one of its
        # parents, the longest path should have the most path components,
        # i.e. the backend representing the inner-most repository.
        inner_most_repo_path = max(vcs_backends, key=len)
        return vcs_backends[inner_most_repo_path]

    def get_backend_for_scheme(self, scheme: str) -> Optional["VersionControl"]:
        """
        Return a VersionControl object or None.
        """
        for vcs_backend in self._registry.values():
            if scheme in vcs_backend.schemes:
                return vcs_backend
        return None

    def get_backend(self, name: str) -> Optional["VersionControl"]:
        """
        Return a VersionControl object or None.
        """
        name = name.lower()
        return self._registry.get(name)


vcs = VcsSupport()


class VersionControl:
    name = ""
    dirname = ""
    repo_name = ""
    # List of supported schemes for this Version Control
    schemes: Tuple[str, ...] = ()
    # Iterable of environment variable names to pass to call_subprocess().
    unset_environ: Tuple[str, ...] = ()
    default_arg_rev: Optional[str] = None

    @classmethod
    def should_add_vcs_url_prefix(cls, remote_url: str) -> bool:
        """
        Return whether the vcs prefix (e.g. "git+") should be added to a
        repository's remote url when used in a requirement.
        """
        return not remote_url.lower().startswith(f"{cls.name}:")

    @classmethod
    def get_subdirectory(cls, location: str) -> Optional[str]:
        """
        Return the path to Python project root, relative to the repo root.
        Return None if the project root is in the repo root.
        """
        return None

    @classmethod
    def get_requirement_revision(cls, repo_dir: str) -> str:
        """
        Return the revision string that should be used in a requirement.
        """
        return cls.get_revision(repo_dir)

    @classmethod
    def get_src_requirement(cls, repo_dir: str, project_name: str) -> str:
        """
        Return the requirement string to use to redownload the files
        currently at the given repository directory.

        Args:
          project_name: the (unescaped) project name.

        The return value has a form similar to the following:

            {repository_url}@{revision}#egg={project_name}
        """
        repo_url = cls.get_remote_url(repo_dir)

        if cls.should_add_vcs_url_prefix(repo_url):
            repo_url = f"{cls.name}+{repo_url}"

        revision = cls.get_requirement_revision(repo_dir)
        subdir = cls.get_subdirectory(repo_dir)
        req = make_vcs_requirement_url(repo_url, revision, project_name, subdir=subdir)

        return req

    @staticmethod
    def get_base_rev_args(rev: str) -> List[str]:
        """
        Return the base revision arguments for a vcs command.

        Args:
          rev: the name of a revision to install.  Cannot be None.
        """
        raise NotImplementedError

    def is_immutable_rev_checkout(self, url: str, dest: str) -> bool:
        """
        Return true if the commit hash checked out at dest matches
        the revision in url.

        Always return False, if the VCS does not support immutable commit
        hashes.

        This method does not check if there are local uncommitted changes
        in dest after checkout, as pip currently has no use case for that.
        """
        return False

    @classmethod
    def make_rev_options(
        cls, rev: Optional[str] = None, extra_args: Optional[CommandArgs] = None
    ) -> RevOptions:
        """
        Return a RevOptions object.

        Args:
          rev: the name of a revision to install.
          extra_args: a list of extra options.
        """
        return RevOptions(cls, rev, extra_args=extra_args or [])

    @classmethod
    def _is_local_repository(cls, repo: str) -> bool:
        """
        posix absolute paths start with os.path.sep,
        win32 ones start with drive (like c:\\folder)
        """
        drive, tail = os.path.splitdrive(repo)
        return repo.startswith(os.path.sep) or bool(drive)

    @classmethod
    def get_netloc_and_auth(
        cls, netloc: str, scheme: str
    ) -> Tuple[str, Tuple[Optional[str], Optional[str]]]:
        """
        Parse the repository URL's netloc, and return the new netloc to use
        along with auth information.

        Args:
          netloc: the original repository URL netloc.
          scheme: the repository URL's scheme without the vcs prefix.

        This is mainly for the Subversion class to override, so that auth
        information can be provided via the --username and --password options
        instead of through the URL.  For other subclasses like Git without
        such an option, auth information must stay in the URL.

        Returns: (netloc, (username, password)).
        """
        return netloc, (None, None)

    @classmethod
    def get_url_rev_and_auth(cls, url: str) -> Tuple[str, Optional[str], AuthInfo]:
        """
        Parse the repository URL to use, and return the URL, revision,
        and auth info to use.

        Returns: (url, rev, (username, password)).
        """
        scheme, netloc, path, query, frag = urllib.parse.urlsplit(url)
        if "+" not in scheme:
            raise ValueError(
                f"Sorry, {url!r} is a malformed VCS url. "
                "The format is <vcs>+<protocol>://<url>, "
                "e.g. svn+http://myrepo/svn/MyApp#egg=MyApp"
            )
        # Remove the vcs prefix.
        scheme = scheme.split("+", 1)[1]
        netloc, user_pass = cls.get_netloc_and_auth(netloc, scheme)
        rev = None
        if "@" in path:
            path, rev = path.rsplit("@", 1)
            if not rev:
                raise InstallationError(
                    f"The URL {url!r} has an empty revision (after @) "
                    "which is not supported. Include a revision after @ "
                    "or remove @ from the URL."
                )
        url = urllib.parse.urlunsplit((scheme, netloc, path, query, ""))
        return url, rev, user_pass

    @staticmethod
    def make_rev_args(
        username: Optional[str], password: Optional[HiddenText]
    ) -> CommandArgs:
        """
        Return the RevOptions "extra arguments" to use in obtain().
        """
        return []

    def get_url_rev_options(self, url: HiddenText) -> Tuple[HiddenText, RevOptions]:
        """
        Return the URL and RevOptions object to use in obtain(),
        as a tuple (url, rev_options).
        """
        secret_url, rev, user_pass = self.get_url_rev_and_auth(url.secret)
        username, secret_password = user_pass
        password: Optional[HiddenText] = None
        if secret_password is not None:
            password = hide_value(secret_password)
        extra_args = self.make_rev_args(username, password)
        rev_options = self.make_rev_options(rev, extra_args=extra_args)

        return hide_url(secret_url), rev_options

    @staticmethod
    def normalize_url(url: str) -> str:
        """
        Normalize a URL for comparison by unquoting it and removing any
        trailing slash.
        """
        return urllib.parse.unquote(url).rstrip("/")

    @classmethod
    def compare_urls(cls, url1: str, url2: str) -> bool:
        """
        Compare two repo URLs for identity, ignoring incidental differences.
        """
        return cls.normalize_url(url1) == cls.normalize_url(url2)

    def fetch_new(
        self, dest: str, url: HiddenText, rev_options: RevOptions, verbosity: int
    ) -> None:
        """
        Fetch a revision from a repository, in the case that this is the
        first fetch from the repository.

        Args:
          dest: the directory to fetch the repository to.
          rev_options: a RevOptions object.
          verbosity: verbosity level.
        """
        raise NotImplementedError

    def switch(self, dest: str, url: HiddenText, rev_options: RevOptions) -> None:
        """
        Switch the repo at ``dest`` to point to ``URL``.

        Args:
          rev_options: a RevOptions object.
        """
        raise NotImplementedError

    def update(self, dest: str, url: HiddenText, rev_options: RevOptions) -> None:
        """
        Update an already-existing repo to the given ``rev_options``.

        Args:
          rev_options: a RevOptions object.
        """
        raise NotImplementedError

    @classmethod
    def is_commit_id_equal(cls, dest: str, name: Optional[str]) -> bool:
        """
        Return whether the id of the current commit equals the given name.

        Args:
          dest: the repository directory.
          name: a string name.
        """
        raise NotImplementedError

    def obtain(self, dest: str, url: HiddenText, verbosity: int) -> None:
        """
        Install or update in editable mode the package represented by this
        VersionControl object.

        :param dest: the repository directory in which to install or update.
        :param url: the repository URL starting with a vcs prefix.
        :param verbosity: verbosity level.
        """
        url, rev_options = self.get_url_rev_options(url)

        if not os.path.exists(dest):
            self.fetch_new(dest, url, rev_options, verbosity=verbosity)
            return

        rev_display = rev_options.to_display()
        if self.is_repository_directory(dest):
            existing_url = self.get_remote_url(dest)
            if self.compare_urls(existing_url, url.secret):
                logger.debug(
                    "%s in %s exists, and has correct URL (%s)",
                    self.repo_name.title(),
                    display_path(dest),
                    url,
                )
                if not self.is_commit_id_equal(dest, rev_options.rev):
                    logger.info(
                        "Updating %s %s%s",
                        display_path(dest),
                        self.repo_name,
                        rev_display,
                    )
                    self.update(dest, url, rev_options)
                else:
                    logger.info("Skipping because already up-to-date.")
                return

            logger.warning(
                "%s %s in %s exists with URL %s",
                self.name,
                self.repo_name,
                display_path(dest),
                existing_url,
            )
            prompt = ("(s)witch, (i)gnore, (w)ipe, (b)ackup ", ("s", "i", "w", "b"))
        else:
            logger.warning(
                "Directory %s already exists, and is not a %s %s.",
                dest,
                self.name,
                self.repo_name,
            )
            # https://github.com/python/mypy/issues/1174
            prompt = ("(i)gnore, (w)ipe, (b)ackup ", ("i", "w", "b"))  # type: ignore

        logger.warning(
            "The plan is to install the %s repository %s",
            self.name,
            url,
        )
        response = ask_path_exists(f"What to do?  {prompt[0]}", prompt[1])

        if response == "a":
            sys.exit(-1)

        if response == "w":
            logger.warning("Deleting %s", display_path(dest))
            rmtree(dest)
            self.fetch_new(dest, url, rev_options, verbosity=verbosity)
            return

        if response == "b":
            dest_dir = backup_dir(dest)
            logger.warning("Backing up %s to %s", display_path(dest), dest_dir)
            shutil.move(dest, dest_dir)
            self.fetch_new(dest, url, rev_options, verbosity=verbosity)
            return

        # Do nothing if the response is "i".
        if response == "s":
            logger.info(
                "Switching %s %s to %s%s",
                self.repo_name,
                display_path(dest),
                url,
                rev_display,
            )
            self.switch(dest, url, rev_options)

    def unpack(self, location: str, url: HiddenText, verbosity: int) -> None:
        """
        Clean up current location and download the url repository
        (and vcs infos) into location

        :param url: the repository URL starting with a vcs prefix.
        :param verbosity: verbosity level.
        """
        if os.path.exists(location):
            rmtree(location)
        self.obtain(location, url=url, verbosity=verbosity)

    @classmethod
    def get_remote_url(cls, location: str) -> str:
        """
        Return the url used at location

        Raises RemoteNotFoundError if the repository does not have a remote
        url configured.
        """
        raise NotImplementedError

    @classmethod
    def get_revision(cls, location: str) -> str:
        """
        Return the current commit id of the files at the given location.
        """
        raise NotImplementedError

    @classmethod
    def run_command(
        cls,
        cmd: Union[List[str], CommandArgs],
        show_stdout: bool = True,
        cwd: Optional[str] = None,
        on_returncode: 'Literal["raise", "warn", "ignore"]' = "raise",
        extra_ok_returncodes: Optional[Iterable[int]] = None,
        command_desc: Optional[str] = None,
        extra_environ: Optional[Mapping[str, Any]] = None,
        spinner: Optional[SpinnerInterface] = None,
        log_failed_cmd: bool = True,
        stdout_only: bool = False,
    ) -> str:
        """
        Run a VCS subcommand
        This is simply a wrapper around call_subprocess that adds the VCS
        command name, and checks that the VCS is available
        """
        cmd = make_command(cls.name, *cmd)
        if command_desc is None:
            command_desc = format_command_args(cmd)
        try:
            return call_subprocess(
                cmd,
                show_stdout,
                cwd,
                on_returncode=on_returncode,
                extra_ok_returncodes=extra_ok_returncodes,
                command_desc=command_desc,
                extra_environ=extra_environ,
                unset_environ=cls.unset_environ,
                spinner=spinner,
                log_failed_cmd=log_failed_cmd,
                stdout_only=stdout_only,
            )
        except NotADirectoryError:
            raise BadCommand(f"Cannot find command {cls.name!r} - invalid PATH")
        except FileNotFoundError:
            # errno.ENOENT = no such file or directory
            # In other words, the VCS executable isn't available
            raise BadCommand(
                f"Cannot find command {cls.name!r} - do you have "
                f"{cls.name!r} installed and in your PATH?"
            )
        except PermissionError:
            # errno.EACCES = Permission denied
            # This error occurs, for instance, when the command is installed
            # only for another user. So, the current user don't have
            # permission to call the other user command.
            raise BadCommand(
                f"No permission to execute {cls.name!r} - install it "
                f"locally, globally (ask admin), or check your PATH. "
                f"See possible solutions at "
                f"https://pip.pypa.io/en/latest/reference/pip_freeze/"
                f"#fixing-permission-denied."
            )

    @classmethod
    def is_repository_directory(cls, path: str) -> bool:
        """
        Return whether a directory path is a repository directory.
        """
        logger.debug("Checking in %s for %s (%s)...", path, cls.dirname, cls.name)
        return os.path.exists(os.path.join(path, cls.dirname))

    @classmethod
    def get_repository_root(cls, location: str) -> Optional[str]:
        """
        Return the "root" (top-level) directory controlled by the vcs,
        or `None` if the directory is not in any.

        It is meant to be overridden to implement smarter detection
        mechanisms for specific vcs.

        This can do more than is_repository_directory() alone. For
        example, the Git override checks that Git is actually available.
        """
        if cls.is_repository_directory(location):
            return location
        return None

from collections import namedtuple
import datetime
import struct


class ExtType(namedtuple("ExtType", "code data")):
    """ExtType represents ext type in msgpack."""

    def __new__(cls, code, data):
        if not isinstance(code, int):
            raise TypeError("code must be int")
        if not isinstance(data, bytes):
            raise TypeError("data must be bytes")
        if not 0 <= code <= 127:
            raise ValueError("code must be 0~127")
        return super().__new__(cls, code, data)


class Timestamp:
    """Timestamp represents the Timestamp extension type in msgpack.

    When built with Cython, msgpack uses C methods to pack and unpack `Timestamp`.
    When using pure-Python msgpack, :func:`to_bytes` and :func:`from_bytes` are used to pack and
    unpack `Timestamp`.

    This class is immutable: Do not override seconds and nanoseconds.
    """

    __slots__ = ["seconds", "nanoseconds"]

    def __init__(self, seconds, nanoseconds=0):
        """Initialize a Timestamp object.

        :param int seconds:
            Number of seconds since the UNIX epoch (00:00:00 UTC Jan 1 1970, minus leap seconds).
            May be negative.

        :param int nanoseconds:
            Number of nanoseconds to add to `seconds` to get fractional time.
            Maximum is 999_999_999.  Default is 0.

        Note: Negative times (before the UNIX epoch) are represented as neg. seconds + pos. ns.
        """
        if not isinstance(seconds, int):
            raise TypeError("seconds must be an integer")
        if not isinstance(nanoseconds, int):
            raise TypeError("nanoseconds must be an integer")
        if not (0 <= nanoseconds < 10**9):
            raise ValueError("nanoseconds must be a non-negative integer less than 999999999.")
        self.seconds = seconds
        self.nanoseconds = nanoseconds

    def __repr__(self):
        """String representation of Timestamp."""
        return f"Timestamp(seconds={self.seconds}, nanoseconds={self.nanoseconds})"

    def __eq__(self, other):
        """Check for equality with another Timestamp object"""
        if type(other) is self.__class__:
            return self.seconds == other.seconds and self.nanoseconds == other.nanoseconds
        return False

    def __ne__(self, other):
        """not-equals method (see :func:`__eq__()`)"""
        return not self.__eq__(other)

    def __hash__(self):
        return hash((self.seconds, self.nanoseconds))

    @staticmethod
    def from_bytes(b):
        """Unpack bytes into a `Timestamp` object.

        Used for pure-Python msgpack unpacking.

        :param b: Payload from msgpack ext message with code -1
        :type b: bytes

        :returns: Timestamp object unpacked from msgpack ext payload
        :rtype: Timestamp
        """
        if len(b) == 4:
            seconds = struct.unpack("!L", b)[0]
            nanoseconds = 0
        elif len(b) == 8:
            data64 = struct.unpack("!Q", b)[0]
            seconds = data64 & 0x00000003FFFFFFFF
            nanoseconds = data64 >> 34
        elif len(b) == 12:
            nanoseconds, seconds = struct.unpack("!Iq", b)
        else:
            raise ValueError(
                "Timestamp type can only be created from 32, 64, or 96-bit byte objects"
            )
        return Timestamp(seconds, nanoseconds)

    def to_bytes(self):
        """Pack this Timestamp object into bytes.

        Used for pure-Python msgpack packing.

        :returns data: Payload for EXT message with code -1 (timestamp type)
        :rtype: bytes
        """
        if (self.seconds >> 34) == 0:  # seconds is non-negative and fits in 34 bits
            data64 = self.nanoseconds << 34 | self.seconds
            if data64 & 0xFFFFFFFF00000000 == 0:
                # nanoseconds is zero and seconds < 2**32, so timestamp 32
                data = struct.pack("!L", data64)
            else:
                # timestamp 64
                data = struct.pack("!Q", data64)
        else:
            # timestamp 96
            data = struct.pack("!Iq", self.nanoseconds, self.seconds)
        return data

    @staticmethod
    def from_unix(unix_sec):
        """Create a Timestamp from posix timestamp in seconds.

        :param unix_float: Posix timestamp in seconds.
        :type unix_float: int or float
        """
        seconds = int(unix_sec // 1)
        nanoseconds = int((unix_sec % 1) * 10**9)
        return Timestamp(seconds, nanoseconds)

    def to_unix(self):
        """Get the timestamp as a floating-point value.

        :returns: posix timestamp
        :rtype: float
        """
        return self.seconds + self.nanoseconds / 1e9

    @staticmethod
    def from_unix_nano(unix_ns):
        """Create a Timestamp from posix timestamp in nanoseconds.

        :param int unix_ns: Posix timestamp in nanoseconds.
        :rtype: Timestamp
        """
        return Timestamp(*divmod(unix_ns, 10**9))

    def to_unix_nano(self):
        """Get the timestamp as a unixtime in nanoseconds.

        :returns: posix timestamp in nanoseconds
        :rtype: int
        """
        return self.seconds * 10**9 + self.nanoseconds

    def to_datetime(self):
        """Get the timestamp as a UTC datetime.

        :rtype: `datetime.datetime`
        """
        utc = datetime.timezone.utc
        return datetime.datetime.fromtimestamp(0, utc) + datetime.timedelta(seconds=self.to_unix())

    @staticmethod
    def from_datetime(dt):
        """Create a Timestamp from datetime with tzinfo.

        :rtype: Timestamp
        """
        return Timestamp.from_unix(dt.timestamp())

import logging
from optparse import Values
from typing import List

from pip._vendor.packaging.utils import canonicalize_name

from pip._internal.cli import cmdoptions
from pip._internal.cli.base_command import Command
from pip._internal.cli.index_command import SessionCommandMixin
from pip._internal.cli.status_codes import SUCCESS
from pip._internal.exceptions import InstallationError
from pip._internal.req import parse_requirements
from pip._internal.req.constructors import (
    install_req_from_line,
    install_req_from_parsed_requirement,
)
from pip._internal.utils.misc import (
    check_externally_managed,
    protect_pip_from_modification_on_windows,
    warn_if_run_as_root,
)

logger = logging.getLogger(__name__)


class UninstallCommand(Command, SessionCommandMixin):
    """
    Uninstall packages.

    pip is able to uninstall most installed packages. Known exceptions are:

    - Pure distutils packages installed with ``python setup.py install``, which
      leave behind no metadata to determine what files were installed.
    - Script wrappers installed by ``python setup.py develop``.
    """

    usage = """
      %prog [options] <package> ...
      %prog [options] -r <requirements file> ..."""

    def add_options(self) -> None:
        self.cmd_opts.add_option(
            "-r",
            "--requirement",
            dest="requirements",
            action="append",
            default=[],
            metavar="file",
            help=(
                "Uninstall all the packages listed in the given requirements "
                "file.  This option can be used multiple times."
            ),
        )
        self.cmd_opts.add_option(
            "-y",
            "--yes",
            dest="yes",
            action="store_true",
            help="Don't ask for confirmation of uninstall deletions.",
        )
        self.cmd_opts.add_option(cmdoptions.root_user_action())
        self.cmd_opts.add_option(cmdoptions.override_externally_managed())
        self.parser.insert_option_group(0, self.cmd_opts)

    def run(self, options: Values, args: List[str]) -> int:
        session = self.get_default_session(options)

        reqs_to_uninstall = {}
        for name in args:
            req = install_req_from_line(
                name,
                isolated=options.isolated_mode,
            )
            if req.name:
                reqs_to_uninstall[canonicalize_name(req.name)] = req
            else:
                logger.warning(
                    "Invalid requirement: %r ignored -"
                    " the uninstall command expects named"
                    " requirements.",
                    name,
                )
        for filename in options.requirements:
            for parsed_req in parse_requirements(
                filename, options=options, session=session
            ):
                req = install_req_from_parsed_requirement(
                    parsed_req, isolated=options.isolated_mode
                )
                if req.name:
                    reqs_to_uninstall[canonicalize_name(req.name)] = req
        if not reqs_to_uninstall:
            raise InstallationError(
                f"You must give at least one requirement to {self.name} (see "
                f'"pip help {self.name}")'
            )

        if not options.override_externally_managed:
            check_externally_managed()

        protect_pip_from_modification_on_windows(
            modifying_pip="pip" in reqs_to_uninstall
        )

        for req in reqs_to_uninstall.values():
            uninstall_pathset = req.uninstall(
                auto_confirm=options.yes,
                verbose=self.verbosity > 0,
            )
            if uninstall_pathset:
                uninstall_pathset.commit()
        if options.root_user_action == "warn":
            warn_if_run_as_root()
        return SUCCESS

"""
    pygments.sphinxext
    ~~~~~~~~~~~~~~~~~~

    Sphinx extension to generate automatic documentation of lexers,
    formatters and filters.

    :copyright: Copyright 2006-2024 by the Pygments team, see AUTHORS.
    :license: BSD, see LICENSE for details.
"""

import sys

from docutils import nodes
from docutils.statemachine import ViewList
from docutils.parsers.rst import Directive
from sphinx.util.nodes import nested_parse_with_titles


MODULEDOC = '''
.. module:: %s

%s
%s
'''

LEXERDOC = '''
.. class:: %s

    :Short names: %s
    :Filenames:   %s
    :MIME types:  %s

    %s

    %s

'''

FMTERDOC = '''
.. class:: %s

    :Short names: %s
    :Filenames: %s

    %s

'''

FILTERDOC = '''
.. class:: %s

    :Name: %s

    %s

'''


class PygmentsDoc(Directive):
    """
    A directive to collect all lexers/formatters/filters and generate
    autoclass directives for them.
    """
    has_content = False
    required_arguments = 1
    optional_arguments = 0
    final_argument_whitespace = False
    option_spec = {}

    def run(self):
        self.filenames = set()
        if self.arguments[0] == 'lexers':
            out = self.document_lexers()
        elif self.arguments[0] == 'formatters':
            out = self.document_formatters()
        elif self.arguments[0] == 'filters':
            out = self.document_filters()
        elif self.arguments[0] == 'lexers_overview':
            out = self.document_lexers_overview()
        else:
            raise Exception('invalid argument for "pygmentsdoc" directive')
        node = nodes.compound()
        vl = ViewList(out.split('\n'), source='')
        nested_parse_with_titles(self.state, vl, node)
        for fn in self.filenames:
            self.state.document.settings.record_dependencies.add(fn)
        return node.children

    def document_lexers_overview(self):
        """Generate a tabular overview of all lexers.

        The columns are the lexer name, the extensions handled by this lexer
        (or "None"), the aliases and a link to the lexer class."""
        from pip._vendor.pygments.lexers._mapping import LEXERS
        from pip._vendor.pygments.lexers import find_lexer_class
        out = []

        table = []

        def format_link(name, url):
            if url:
                return f'`{name} <{url}>`_'
            return name

        for classname, data in sorted(LEXERS.items(), key=lambda x: x[1][1].lower()):
            lexer_cls = find_lexer_class(data[1])
            extensions = lexer_cls.filenames + lexer_cls.alias_filenames

            table.append({
                'name': format_link(data[1], lexer_cls.url),
                'extensions': ', '.join(extensions).replace('*', '\\*').replace('_', '\\') or 'None',
                'aliases': ', '.join(data[2]),
                'class': f'{data[0]}.{classname}'
            })

        column_names = ['name', 'extensions', 'aliases', 'class']
        column_lengths = [max([len(row[column]) for row in table if row[column]])
                          for column in column_names]

        def write_row(*columns):
            """Format a table row"""
            out = []
            for length, col in zip(column_lengths, columns):
                if col:
                    out.append(col.ljust(length))
                else:
                    out.append(' '*length)

            return ' '.join(out)

        def write_seperator():
            """Write a table separator row"""
            sep = ['='*c for c in column_lengths]
            return write_row(*sep)

        out.append(write_seperator())
        out.append(write_row('Name', 'Extension(s)', 'Short name(s)', 'Lexer class'))
        out.append(write_seperator())
        for row in table:
            out.append(write_row(
                row['name'],
                row['extensions'],
                row['aliases'],
                f':class:`~{row["class"]}`'))
        out.append(write_seperator())

        return '\n'.join(out)

    def document_lexers(self):
        from pip._vendor.pygments.lexers._mapping import LEXERS
        from pip._vendor import pygments
        import inspect
        import pathlib

        out = []
        modules = {}
        moduledocstrings = {}
        for classname, data in sorted(LEXERS.items(), key=lambda x: x[0]):
            module = data[0]
            mod = __import__(module, None, None, [classname])
            self.filenames.add(mod.__file__)
            cls = getattr(mod, classname)
            if not cls.__doc__:
                print(f"Warning: {classname} does not have a docstring.")
            docstring = cls.__doc__
            if isinstance(docstring, bytes):
                docstring = docstring.decode('utf8')

            example_file = getattr(cls, '_example', None)
            if example_file:
                p = pathlib.Path(inspect.getabsfile(pygments)).parent.parent /\
                    'tests' / 'examplefiles' / example_file
                content = p.read_text(encoding='utf-8')
                if not content:
                    raise Exception(
                        f"Empty example file '{example_file}' for lexer "
                        f"{classname}")

                if data[2]:
                    lexer_name = data[2][0]
                    docstring += '\n\n    .. admonition:: Example\n'
                    docstring += f'\n      .. code-block:: {lexer_name}\n\n'
                    for line in content.splitlines():
                        docstring += f'          {line}\n'

            if cls.version_added:
                version_line = f'.. versionadded:: {cls.version_added}'
            else:
                version_line = ''

            modules.setdefault(module, []).append((
                classname,
                ', '.join(data[2]) or 'None',
                ', '.join(data[3]).replace('*', '\\*').replace('_', '\\') or 'None',
                ', '.join(data[4]) or 'None',
                docstring,
                version_line))
            if module not in moduledocstrings:
                moddoc = mod.__doc__
                if isinstance(moddoc, bytes):
                    moddoc = moddoc.decode('utf8')
                moduledocstrings[module] = moddoc

        for module, lexers in sorted(modules.items(), key=lambda x: x[0]):
            if moduledocstrings[module] is None:
                raise Exception(f"Missing docstring for {module}")
            heading = moduledocstrings[module].splitlines()[4].strip().rstrip('.')
            out.append(MODULEDOC % (module, heading, '-'*len(heading)))
            for data in lexers:
                out.append(LEXERDOC % data)

        return ''.join(out)

    def document_formatters(self):
        from pip._vendor.pygments.formatters import FORMATTERS

        out = []
        for classname, data in sorted(FORMATTERS.items(), key=lambda x: x[0]):
            module = data[0]
            mod = __import__(module, None, None, [classname])
            self.filenames.add(mod.__file__)
            cls = getattr(mod, classname)
            docstring = cls.__doc__
            if isinstance(docstring, bytes):
                docstring = docstring.decode('utf8')
            heading = cls.__name__
            out.append(FMTERDOC % (heading, ', '.join(data[2]) or 'None',
                                   ', '.join(data[3]).replace('*', '\\*') or 'None',
                                   docstring))
        return ''.join(out)

    def document_filters(self):
        from pip._vendor.pygments.filters import FILTERS

        out = []
        for name, cls in FILTERS.items():
            self.filenames.add(sys.modules[cls.__module__].__file__)
            docstring = cls.__doc__
            if isinstance(docstring, bytes):
                docstring = docstring.decode('utf8')
            out.append(FILTERDOC % (cls.__name__, name, docstring))
        return ''.join(out)


def setup(app):
    app.add_directive('pygmentsdoc', PygmentsDoc)

"""Android."""

from __future__ import annotations

import os
import re
import sys
from functools import lru_cache
from typing import TYPE_CHECKING, cast

from .api import PlatformDirsABC


class Android(PlatformDirsABC):
    """
    Follows the guidance `from here <https://android.stackexchange.com/a/216132>`_.

    Makes use of the `appname <platformdirs.api.PlatformDirsABC.appname>`, `version
    <platformdirs.api.PlatformDirsABC.version>`, `ensure_exists <platformdirs.api.PlatformDirsABC.ensure_exists>`.

    """

    @property
    def user_data_dir(self) -> str:
        """:return: data directory tied to the user, e.g. ``/data/user/<userid>/<packagename>/files/<AppName>``"""
        return self._append_app_name_and_version(cast(str, _android_folder()), "files")

    @property
    def site_data_dir(self) -> str:
        """:return: data directory shared by users, same as `user_data_dir`"""
        return self.user_data_dir

    @property
    def user_config_dir(self) -> str:
        """
        :return: config directory tied to the user, e.g. \
        ``/data/user/<userid>/<packagename>/shared_prefs/<AppName>``
        """
        return self._append_app_name_and_version(cast(str, _android_folder()), "shared_prefs")

    @property
    def site_config_dir(self) -> str:
        """:return: config directory shared by the users, same as `user_config_dir`"""
        return self.user_config_dir

    @property
    def user_cache_dir(self) -> str:
        """:return: cache directory tied to the user, e.g.,``/data/user/<userid>/<packagename>/cache/<AppName>``"""
        return self._append_app_name_and_version(cast(str, _android_folder()), "cache")

    @property
    def site_cache_dir(self) -> str:
        """:return: cache directory shared by users, same as `user_cache_dir`"""
        return self.user_cache_dir

    @property
    def user_state_dir(self) -> str:
        """:return: state directory tied to the user, same as `user_data_dir`"""
        return self.user_data_dir

    @property
    def user_log_dir(self) -> str:
        """
        :return: log directory tied to the user, same as `user_cache_dir` if not opinionated else ``log`` in it,
          e.g. ``/data/user/<userid>/<packagename>/cache/<AppName>/log``
        """
        path = self.user_cache_dir
        if self.opinion:
            path = os.path.join(path, "log")  # noqa: PTH118
        return path

    @property
    def user_documents_dir(self) -> str:
        """:return: documents directory tied to the user e.g. ``/storage/emulated/0/Documents``"""
        return _android_documents_folder()

    @property
    def user_downloads_dir(self) -> str:
        """:return: downloads directory tied to the user e.g. ``/storage/emulated/0/Downloads``"""
        return _android_downloads_folder()

    @property
    def user_pictures_dir(self) -> str:
        """:return: pictures directory tied to the user e.g. ``/storage/emulated/0/Pictures``"""
        return _android_pictures_folder()

    @property
    def user_videos_dir(self) -> str:
        """:return: videos directory tied to the user e.g. ``/storage/emulated/0/DCIM/Camera``"""
        return _android_videos_folder()

    @property
    def user_music_dir(self) -> str:
        """:return: music directory tied to the user e.g. ``/storage/emulated/0/Music``"""
        return _android_music_folder()

    @property
    def user_desktop_dir(self) -> str:
        """:return: desktop directory tied to the user e.g. ``/storage/emulated/0/Desktop``"""
        return "/storage/emulated/0/Desktop"

    @property
    def user_runtime_dir(self) -> str:
        """
        :return: runtime directory tied to the user, same as `user_cache_dir` if not opinionated else ``tmp`` in it,
          e.g. ``/data/user/<userid>/<packagename>/cache/<AppName>/tmp``
        """
        path = self.user_cache_dir
        if self.opinion:
            path = os.path.join(path, "tmp")  # noqa: PTH118
        return path

    @property
    def site_runtime_dir(self) -> str:
        """:return: runtime directory shared by users, same as `user_runtime_dir`"""
        return self.user_runtime_dir


@lru_cache(maxsize=1)
def _android_folder() -> str | None:  # noqa: C901, PLR0912
    """:return: base folder for the Android OS or None if it cannot be found"""
    result: str | None = None
    # type checker isn't happy with our "import android", just don't do this when type checking see
    # https://stackoverflow.com/a/61394121
    if not TYPE_CHECKING:
        try:
            # First try to get a path to android app using python4android (if available)...
            from android import mActivity  # noqa: PLC0415

            context = cast("android.content.Context", mActivity.getApplicationContext())  # noqa: F821
            result = context.getFilesDir().getParentFile().getAbsolutePath()
        except Exception:  # noqa: BLE001
            result = None
    if result is None:
        try:
            # ...and fall back to using plain pyjnius, if python4android isn't available or doesn't deliver any useful
            # result...
            from jnius import autoclass  # noqa: PLC0415

            context = autoclass("android.content.Context")
            result = context.getFilesDir().getParentFile().getAbsolutePath()
        except Exception:  # noqa: BLE001
            result = None
    if result is None:
        # and if that fails, too, find an android folder looking at path on the sys.path
        # warning: only works for apps installed under /data, not adopted storage etc.
        pattern = re.compile(r"/data/(data|user/\d+)/(.+)/files")
        for path in sys.path:
            if pattern.match(path):
                result = path.split("/files")[0]
                break
        else:
            result = None
    if result is None:
        # one last try: find an android folder looking at path on the sys.path taking adopted storage paths into
        # account
        pattern = re.compile(r"/mnt/expand/[a-fA-F0-9-]{36}/(data|user/\d+)/(.+)/files")
        for path in sys.path:
            if pattern.match(path):
                result = path.split("/files")[0]
                break
        else:
            result = None
    return result


@lru_cache(maxsize=1)
def _android_documents_folder() -> str:
    """:return: documents folder for the Android OS"""
    # Get directories with pyjnius
    try:
        from jnius import autoclass  # noqa: PLC0415

        context = autoclass("android.content.Context")
        environment = autoclass("android.os.Environment")
        documents_dir: str = context.getExternalFilesDir(environment.DIRECTORY_DOCUMENTS).getAbsolutePath()
    except Exception:  # noqa: BLE001
        documents_dir = "/storage/emulated/0/Documents"

    return documents_dir


@lru_cache(maxsize=1)
def _android_downloads_folder() -> str:
    """:return: downloads folder for the Android OS"""
    # Get directories with pyjnius
    try:
        from jnius import autoclass  # noqa: PLC0415

        context = autoclass("android.content.Context")
        environment = autoclass("android.os.Environment")
        downloads_dir: str = context.getExternalFilesDir(environment.DIRECTORY_DOWNLOADS).getAbsolutePath()
    except Exception:  # noqa: BLE001
        downloads_dir = "/storage/emulated/0/Downloads"

    return downloads_dir


@lru_cache(maxsize=1)
def _android_pictures_folder() -> str:
    """:return: pictures folder for the Android OS"""
    # Get directories with pyjnius
    try:
        from jnius import autoclass  # noqa: PLC0415

        context = autoclass("android.content.Context")
        environment = autoclass("android.os.Environment")
        pictures_dir: str = context.getExternalFilesDir(environment.DIRECTORY_PICTURES).getAbsolutePath()
    except Exception:  # noqa: BLE001
        pictures_dir = "/storage/emulated/0/Pictures"

    return pictures_dir


@lru_cache(maxsize=1)
def _android_videos_folder() -> str:
    """:return: videos folder for the Android OS"""
    # Get directories with pyjnius
    try:
        from jnius import autoclass  # noqa: PLC0415

        context = autoclass("android.content.Context")
        environment = autoclass("android.os.Environment")
        videos_dir: str = context.getExternalFilesDir(environment.DIRECTORY_DCIM).getAbsolutePath()
    except Exception:  # noqa: BLE001
        videos_dir = "/storage/emulated/0/DCIM/Camera"

    return videos_dir


@lru_cache(maxsize=1)
def _android_music_folder() -> str:
    """:return: music folder for the Android OS"""
    # Get directories with pyjnius
    try:
        from jnius import autoclass  # noqa: PLC0415

        context = autoclass("android.content.Context")
        environment = autoclass("android.os.Environment")
        music_dir: str = context.getExternalFilesDir(environment.DIRECTORY_MUSIC).getAbsolutePath()
    except Exception:  # noqa: BLE001
        music_dir = "/storage/emulated/0/Music"

    return music_dir


__all__ = [
    "Android",
]