<commit_before>"""Various helper functions"""

import asyncio
import base64
import binascii
import cgi
import datetime
import functools
import inspect
import netrc
import os
import re
import sys
import time
import weakref
from collections import namedtuple
from collections.abc import Mapping as ABCMapping
from contextlib import suppress
from math import ceil
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple  # noqa
from urllib.parse import quote
from urllib.request import getproxies

import async_timeout
import attr
from multidict import MultiDict
from yarl import URL

from . import hdrs
from .abc import AbstractAccessLogger
from .log import client_logger


__all__ = ('BasicAuth', 'ChainMapProxy')

PY_36 = sys.version_info >= (3, 6)
PY_37 = sys.version_info >= (3, 7)

if not PY_37:
    import idna_ssl
    idna_ssl.patch_match_hostname()


sentinel = object()  # type: Any
NO_EXTENSIONS = bool(os.environ.get('AIOHTTP_NO_EXTENSIONS'))  # type: bool

# N.B. sys.flags.dev_mode is available on Python 3.7+, use getattr
# for compatibility with older versions
DEBUG = (getattr(sys.flags, 'dev_mode', False) or
         (not sys.flags.ignore_environment and
          bool(os.environ.get('PYTHONASYNCIODEBUG'))))


CHAR = set(chr(i) for i in range(0, 128))
CTL = set(chr(i) for i in range(0, 32)) | {chr(127), }
SEPARATORS = {'(', ')', '<', '>', '@', ',', ';', ':', '\\', '"', '/', '[', ']',
              '?', '=', '{', '}', ' ', chr(9)}
TOKEN = CHAR ^ CTL ^ SEPARATORS


coroutines = asyncio.coroutines
old_debug = coroutines._DEBUG  # type: ignore

# prevent "coroutine noop was never awaited" warning.
coroutines._DEBUG = False  # type: ignore


@asyncio.coroutine
def noop(*args, **kwargs):
    return


coroutines._DEBUG = old_debug  # type: ignore


class BasicAuth(namedtuple('BasicAuth', ['login', 'password', 'encoding'])):
    """Http basic authentication helper."""

    def __new__(cls, login: str,
                password: str='',
                encoding: str='latin1') -> 'BasicAuth':
        if login is None:
            raise ValueError('None is not allowed as login value')

        if password is None:
            raise ValueError('None is not allowed as password value')

        if ':' in login:
            raise ValueError(
                'A ":" is not allowed in login (RFC 1945#section-11.1)')

        return super().__new__(cls, login, password, encoding)

    @classmethod
    def decode(cls, auth_header: str, encoding: str='latin1') -> 'BasicAuth':
        """Create a BasicAuth object from an Authorization HTTP header."""
        try:
            auth_type, encoded_credentials = auth_header.split(' ', 1)
        except ValueError:
            raise ValueError('Could not parse authorization header.')

        if auth_type.lower() != 'basic':
            raise ValueError('Unknown authorization method %s' % auth_type)

        try:
            decoded = base64.b64decode(
                encoded_credentials.encode('ascii'), validate=True
            ).decode(encoding)
        except binascii.Error:
            raise ValueError('Invalid base64 encoding.')

        try:
            # RFC 2617 HTTP Authentication
            # https://www.ietf.org/rfc/rfc2617.txt
            # the colon must be present, but the username and password may be
            # otherwise blank.
            username, password = decoded.split(':', 1)
        except ValueError:
            raise ValueError('Invalid credentials.')

        return cls(username, password, encoding=encoding)

    @classmethod
    def from_url(cls, url: URL,
                 *, encoding: str='latin1') -> Optional['BasicAuth']:
        """Create BasicAuth from url."""
        if not isinstance(url, URL):
            raise TypeError("url should be yarl.URL instance")
        if url.user is None:
            return None
        return cls(url.user, url.password or '', encoding=encoding)

    def encode(self) -> str:
        """Encode credentials."""
        creds = ('%s:%s' % (self.login, self.password)).encode(self.encoding)
        return 'Basic %s' % base64.b64encode(creds).decode(self.encoding)


def strip_auth_from_url(url: URL) -> Tuple[URL, Optional[BasicAuth]]:
    auth = BasicAuth.from_url(url)
    if auth is None:
        return url, None
    else:
        return url.with_user(None), auth


def netrc_from_env():
    netrc_obj = None
    netrc_path = os.environ.get('NETRC')
    try:
        if netrc_path is not None:
            netrc_path = Path(netrc_path)
        else:
            home_dir = Path.home()
            if os.name == 'nt':  # pragma: no cover
                netrc_path = home_dir.joinpath('_netrc')
            else:
                netrc_path = home_dir.joinpath('.netrc')

        if netrc_path and netrc_path.is_file():
            try:
                netrc_obj = netrc.netrc(str(netrc_path))
            except (netrc.NetrcParseError, OSError) as e:
                client_logger.warning(".netrc file parses fail: %s", e)

        if netrc_obj is None:
            client_logger.warning("could't find .netrc file")
    except RuntimeError as e:  # pragma: no cover
        """ handle error raised by pathlib """
        client_logger.warning("could't find .netrc file: %s", e)
    return netrc_obj


@attr.s(frozen=True, slots=True)
class ProxyInfo:
    proxy = attr.ib(type=str)
    proxy_auth = attr.ib(type=BasicAuth)
<commit_msg>Quiet warnings about netrc<commit_after>"""Various helper functions"""

import asyncio
import base64
import binascii
import cgi
import datetime
import functools
import inspect
import netrc
import os
import platform
import re
import sys
import time
import weakref
from collections import namedtuple
from collections.abc import Mapping as ABCMapping
from contextlib import suppress
from math import ceil
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple  # noqa
from urllib.parse import quote
from urllib.request import getproxies

import async_timeout
import attr
from multidict import MultiDict
from yarl import URL

from . import hdrs
from .abc import AbstractAccessLogger
from .log import client_logger


__all__ = ('BasicAuth', 'ChainMapProxy')

PY_36 = sys.version_info >= (3, 6)
PY_37 = sys.version_info >= (3, 7)

if not PY_37:
    import idna_ssl
    idna_ssl.patch_match_hostname()


sentinel = object()  # type: Any
NO_EXTENSIONS = bool(os.environ.get('AIOHTTP_NO_EXTENSIONS'))  # type: bool

# N.B. sys.flags.dev_mode is available on Python 3.7+, use getattr
# for compatibility with older versions
DEBUG = (getattr(sys.flags, 'dev_mode', False) or
         (not sys.flags.ignore_environment and
          bool(os.environ.get('PYTHONASYNCIODEBUG'))))


CHAR = set(chr(i) for i in range(0, 128))
CTL = set(chr(i) for i in range(0, 32)) | {chr(127), }
SEPARATORS = {'(', ')', '<', '>', '@', ',', ';', ':', '\\', '"', '/', '[', ']',
              '?', '=', '{', '}', ' ', chr(9)}
TOKEN = CHAR ^ CTL ^ SEPARATORS


coroutines = asyncio.coroutines
old_debug = coroutines._DEBUG  # type: ignore

# prevent "coroutine noop was never awaited" warning.
coroutines._DEBUG = False  # type: ignore


@asyncio.coroutine
def noop(*args, **kwargs):
    return


coroutines._DEBUG = old_debug  # type: ignore


class BasicAuth(namedtuple('BasicAuth', ['login', 'password', 'encoding'])):
    """Http basic authentication helper."""

    def __new__(cls, login: str,
                password: str='',
                encoding: str='latin1') -> 'BasicAuth':
        if login is None:
            raise ValueError('None is not allowed as login value')

        if password is None:
            raise ValueError('None is not allowed as password value')

        if ':' in login:
            raise ValueError(
                'A ":" is not allowed in login (RFC 1945#section-11.1)')

        return super().__new__(cls, login, password, encoding)

    @classmethod
    def decode(cls, auth_header: str, encoding: str='latin1') -> 'BasicAuth':
        """Create a BasicAuth object from an Authorization HTTP header."""
        try:
            auth_type, encoded_credentials = auth_header.split(' ', 1)
        except ValueError:
            raise ValueError('Could not parse authorization header.')

        if auth_type.lower() != 'basic':
            raise ValueError('Unknown authorization method %s' % auth_type)

        try:
            decoded = base64.b64decode(
                encoded_credentials.encode('ascii'), validate=True
            ).decode(encoding)
        except binascii.Error:
            raise ValueError('Invalid base64 encoding.')

        try:
            # RFC 2617 HTTP Authentication
            # https://www.ietf.org/rfc/rfc2617.txt
            # the colon must be present, but the username and password may be
            # otherwise blank.
            username, password = decoded.split(':', 1)
        except ValueError:
            raise ValueError('Invalid credentials.')

        return cls(username, password, encoding=encoding)

    @classmethod
    def from_url(cls, url: URL,
                 *, encoding: str='latin1') -> Optional['BasicAuth']:
        """Create BasicAuth from url."""
        if not isinstance(url, URL):
            raise TypeError("url should be yarl.URL instance")
        if url.user is None:
            return None
        return cls(url.user, url.password or '', encoding=encoding)

    def encode(self) -> str:
        """Encode credentials."""
        creds = ('%s:%s' % (self.login, self.password)).encode(self.encoding)
        return 'Basic %s' % base64.b64encode(creds).decode(self.encoding)


def strip_auth_from_url(url: URL) -> Tuple[URL, Optional[BasicAuth]]:
    auth = BasicAuth.from_url(url)
    if auth is None:
        return url, None
    else:
        return url.with_user(None), auth


def netrc_from_env():
    """Attempt to load the netrc file from the path specified by the env-var
    NETRC or in the default location in the user's home directory.

    Returns ``None`` if it couldn't be found or fails to parse.
    """
    netrc_env = os.environ.get('NETRC')

    if netrc_env is not None:
        netrc_path = Path(netrc_env)
    else:
        try:
            home_dir = Path.home()
        except RuntimeError as e:  # pragma: no cover
            # if pathlib can't resolve home, it may raise a RuntimeError
            client_logger.warning('Could not resolve home directory when '
                                  'trying to look for .netrc file: %s', e)
            return None

        netrc_path = home_dir / (
            '_netrc' if platform.system() == 'Windows' else '.netrc')

    try:
        return netrc.netrc(str(netrc_path))
    except netrc.NetrcParseError as e:
        client_logger.warning('Could not parse .netrc file: %s', e)
    except OSError as e:
        # we couldn't read the file (doesn't exist, permissions, etc.)
        if netrc_env or netrc_path.is_file():
            # only warn if the enviroment wanted us to load it,
            # or it appears like the default file does actually exist
            client_logger.warning('Could not read .netrc file: %s', e)

    return None


@attr.s(frozen=True, slots=True)
class ProxyInfo:
    proxy = attr.ib(type=str)
    proxy_auth = attr.ib(type=BasicAuth)

{'fraction_non_alphanumeric': 0.11322262676494095, 'fraction_numerical': 0.00976822662285765, 'mean_word_length': 2.68520942408377, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 19, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7085222', 'n_tokens_mistral': 3449, 'n_tokens_neox': 3310, 'n_words': 1034}