from contextlib import contextmanager
import attr
import socket
import yaml
import re

import six


@attr.s(frozen=True)
class Job(object):
    """Structure holding a job returned from Beanstalk

    :ivar job_id: Opaque identifier for the job (to be passed to :func:`BeanstalkClient.release_job()` or
        :func:`BeanstalkClient.stats_job()`).

    :ivar job_data: Blob of the data. str if :attr:`BeanstalkClient.auto_decode` is True; otherwise bytes
    """

    job_id = attr.ib(validator=attr.validators.instance_of(int))

    job_data = attr.ib()


if getattr(attr, "__version_info__", (0,)) >= (19, 2):
    _attrs_kwargs = dict(eq=True)
else:
    _attrs_kwargs = dict(cmp=True)


@attr.s(frozen=True, hash=True, **_attrs_kwargs)
class BeanstalkError(Exception):
    """Common error raised when something goes wrong with beanstalk"""

    message = attr.ib(converter=lambda m: m.decode('ascii'))


def yaml_load(fo):
    # yaml.safe_load will never use the C loader; we have to detect it ourselves
    if hasattr(yaml, 'CSafeLoader'):
        return yaml.load(fo, Loader=yaml.CSafeLoader)
    else:
        return yaml.safe_load(fo)


@attr.s(frozen=True)
class BeanstalkInsertingProxy(object):
    """Proxy object yielded by :func:`BeanstalkClient.using()`"""

    beanstalk_client = attr.ib()
    tube = attr.ib()

    def put_job(self, data, pri=65536, delay=0, ttr=120):
        """Method to insert a job into the tube selected with :func:`BeanstalkClient.using`.

        :param data: Job body
        :type data: Text (either str which will be encoded as utf-8, or bytes which are already utf-8
        :param pri: Priority for the job
        :type pri: int
        :param delay: Delay in seconds before the job should be placed on the ready queue
        :type delay: int
        :param ttr: Time to reserve (how long a worker may work on this job before we assume the worker is blocked
            and give the job to another worker
        :type ttr: int
        """

        self.beanstalk_client.use(self.tube)
        return self.beanstalk_client.put_job(data=data, pri=pri, delay=delay, ttr=ttr)


class BeanstalkClient(object):
    """Simple wrapper around the Beanstalk API.

    :param host: Hostname or IP address to connect to
    :type host: str
    :param port: Port to connect to
    :type port: int
    :param socket_timeout: Timeout to set on the socket.
    :type socket_timeout: float
    :param auto_decode: Attempt to decode job bodies as UTF-8 when reading them
    :type auto_decode: bool

    Doesn't provide any fanciness for writing consumers or producers. Just lets you invoke methods to call beanstalk
    functions.

      .. warning::

         Setting socket timeout to a value lower than the value you pass to blocking functions like
         :func:`reserve_job()` will cause errors!
    """
    def __init__(self, host, port=11300, socket_timeout=None, auto_decode=False):
        """Construct a synchronous Beanstalk Client. Does not connect!"""
        self.host = host
        self.port = port
        self.socket_timeout = socket_timeout
        self._reset_state()
        self.desired_tube = 'default'
        self.desired_watchlist = set(['default'])
        self.auto_decode = auto_decode

    @classmethod
    def from_uri(cls, uri, socket_timeout=None, auto_decode=False):
        """Construct a synchronous Beanstalk Client from a URI.

        The URI may be of the form beanstalk://host:port or beanstalkd://host:port

        IPv6 literals must be wrapped in brackets as per RFC 2732.
        """
        parts = six.moves.urllib.parse.urlparse(uri)
        if parts.scheme.lower() not in ('beanstalk', 'beanstalkd'):
            raise ValueError('Invalid scheme %s' % parts.scheme)
        ipv6_md = re.match(r'^\[([0-9a-fA-F:]+)\](:[0-9]+)?$', parts.netloc)
        if ipv6_md:
            host = ipv6_md.group(1)
            port = ipv6_md.group(2) or '11300'
            port = port.lstrip(':')
        elif ':' in parts.netloc:
            host, port = parts.netloc.rsplit(':', 1)
        else:
            host = parts.netloc
            port = 11300
        port = int(port)
        return cls(host, port, socket_timeout=socket_timeout, auto_decode=auto_decode)

    def _reset_state(self):
        self._watchlist = set(['default'])
        self.current_tube = 'default'
        self.initial_watch = True
        self.socket = None

    def __repr__(self):
        return '{0}({1!r}, {2!r})'.format(self.__class__.__name__, self.host, self.port)  # pragma: no cover

    def __str__(self):
        return '{0} - watching:{1}, current:{2}'.format(  # pragma: no cover
            repr(self), self._watchlist, self.current_tube  # pragma: no cover
        )  # pragma: no cover

    @property
    def _socket(self):
        if self.socket is None:
            self.socket = socket.create_connection((self.host, self.port), timeout=self.socket_timeout)
            self._re_establish_use_watch()
        return self.socket

    def _re_establish_use_watch(self):
        """Call after a close/re-connect.

        Automatically re-establishes the USE and WATCH configs previously setup.
        """
        if self.current_tube != self.desired_tube:
            self.use(self.desired_tube)
        if self._watchlist != self.desired_watchlist:
            self.watchlist = self.desired_watchlist

    def close(self):
        """Close any open connection to the Beanstalk server.

        This object is still safe to use after calling :func:`close()` ; it will automatically reconnect
        and re-establish any open watches / uses.

        It is a logic error to close the connection while you have a reserved job
        """
        if self.socket is not None:
            self.socket.close()
            self._reset_state()

    @contextmanager
    def _sock_ctx(self):
        yield self._socket

    def _receive_data_with_prefix(self, prefix, sock):
        buf = b''
        target_len = len(prefix) + 28
        while b'\r\n' not in buf:
            message = sock.recv(target_len - len(buf))
            if not message:
                break
            buf += message
        if b' ' not in buf:
            error = buf.rstrip()
            raise BeanstalkError(error)
        first_word, rest = buf.split(b' ', 1)
        if first_word != prefix:
            raise BeanstalkError(first_word)
        return self._receive_data(sock, rest)

    def _receive_id_and_data_with_prefix(self, prefix, sock):
        buf = b''
        target_len = len(prefix) + 28
        while b'\r\n' not in buf:
            message = sock.recv(target_len - len(buf))
            if not message:
                break
            buf += message
        if b' ' not in buf:
            error = buf.rstrip()
            raise BeanstalkError(error)
        first_word, rest = buf.split(b' ', 1)
        if first_word != prefix:
            raise BeanstalkError(first_word)
        the_id, rest = rest.split(b' ', 1)
        return int(the_id), self._receive_data(sock, rest)

    def _receive_data(self, sock, initial=None):
        if initial is None:
            initial = sock.recv(12)
        byte_length, rest = initial.split(b'\r\n', 1)
        byte_length = int(byte_length) + 2
        buf = [rest]
        bytes_read = len(rest)
        while bytes_read < byte_length:
            message = sock.recv(min(4096, byte_length - bytes_read))
            if not message:
                break
            bytes_read += len(message)
            buf.append(message)
        bytez = b''.join(buf)[:-2]
        if self.auto_decode:
            return bytez.decode('utf-8')
        else:
            return bytez

    def _receive_id(self, sock):
        status, gid = self._receive_name(sock)
        return status, int(gid)

    def _receive_name(self, sock):
        message = sock.recv(1024)
        if b' ' in message:
            status, rest = message.split(b' ', 1)
            return status, rest.rstrip()
        else:
            raise BeanstalkError(message.rstrip())

    def _receive_word(self, sock, *expected_words):
        message = sock.recv(1024).rstrip()
        if message not in expected_words:
            raise BeanstalkError(message)
        return message

    def _send_message(self, message, sock):
        if isinstance(message, bytes):
            if not message.endswith(b'\r\n'):
                message += b'\r\n'
            return sock.sendall(message)
        else:
            if not message.endswith('\r\n'):
                message += '\r\n'
            return sock.sendall(message.encode('utf-8'))

    def list_tubes(self):
        """Return a list of tubes that this beanstalk instance knows about

        :rtype: list of tubes
        """
        with self._sock_ctx() as sock:
            self._send_message('list-tubes', sock)
            body = self._receive_data_with_prefix(b'OK', sock)
            tubes = yaml_load(body)
            return tubes

    def stats(self):
        """Return a dictionary with a bunch of instance-wide statistics

        :rtype: dict
        """
        with self._sock_ctx() as socket:
            self._send_message('stats', socket)
            body = self._receive_data_with_prefix(b'OK', socket)
            stats = yaml_load(body)
            return stats

    def put_job(self, data, pri=65536, delay=0, ttr=120):
        """Insert a new job into whatever queue is currently USEd

        :param data: Job body
        :type data: Text (either str which will be encoded as utf-8, or bytes which are already utf-8
        :param pri: Priority for the job
        :type pri: int
        :param delay: Delay in seconds before the job should be placed on the ready queue
        :type delay: int
        :param ttr: Time to reserve (how long a worker may work on this job before we assume the worker is blocked
            and give the job to another worker
        :type ttr: int

        .. seealso::

           :func:`put_job_into()`
              Put a job into a specific tube

           :func:`using()`
              Insert a job using an external guard
        """
        with self._sock_ctx() as socket:
            message = 'put {pri} {delay} {ttr} {datalen}\r\n'.format(
                pri=pri, delay=delay, ttr=ttr, datalen=len(data)
            ).encode('utf-8')
            if not isinstance(data, bytes):
                data = data.encode('utf-8')
            message += data
            message += b'\r\n'
            self._send_message(message, socket)
            return self._receive_id(socket)

    def put_job_into(self, tube_name, data, pri=65536, delay=0, ttr=120):
        """Insert a new job into a specific queue. Wrapper around :func:`put_job`.

        :param tube_name: Tube name
        :type tube_name: str
        :param data: Job body
        :type data: Text (either str which will be encoded as utf-8, or bytes which are already utf-8
        :param pri: Priority for the job
        :type pri: int
        :param delay: Delay in seconds before the job should be placed on the ready queue
        :type delay: int
        :param ttr: Time to reserve (how long a worker may work on this job before we assume the worker is blocked
            and give the job to another worker
        :type ttr: int

        .. seealso::

           :func:`put_job()`
              Put a job into whatever the current tube is

           :func:`using()`
              Insert a job using an external guard
        """
        with self.using(tube_name) as inserter:
            return inserter.put_job(data=data, pri=pri, delay=delay, ttr=ttr)

    @property
    def watchlist(self):
        return self._watchlist

    @watchlist.setter
    def watchlist(self, tubes):
        """Set the watchlist to the given tubes

        :param tubes: A list of tubes to watch

        Automatically un-watches any tubes that are not on the target list
        """
        tubes = set(tubes)
        for tube in tubes - self._watchlist:
            self.watch(tube)
        for tube in self._watchlist - tubes:
            self.ignore(tube)

    def watch(self, tube):
        """Add the given tube to the watchlist.

        :param tube: Name of the tube to add to the watchlist

        Note: Initially, all connections are watching a tube named "default". If
        you manually call :func:`watch()`, we will un-watch the "default" tube.
        To keep it in your list, first call :func:`watch()` with the other tubes, then
        call :func:`watch()` with "default".
        """
        with self._sock_ctx() as socket:
            self.desired_watchlist.add(tube)
            if tube not in self._watchlist:
                self._send_message('watch {0}'.format(tube), socket)
                self._receive_id(socket)
                self._watchlist.add(tube)
                if self.initial_watch:
                    if tube != 'default':
                        self.ignore('default')
                    self.initial_watch = False

    def ignore(self, tube):