# -*- coding: utf-8 -*-
'''
Created on Mar 22, 2012

@author: ymh

Module directly inspired by tweetstream

'''
import time
import requests
from requests.utils import stream_decode_response_unicode
import anyjson
import select

from . import  USER_AGENT, ConnectionError, AuthenticationError


def iter_content_non_blocking(req, max_chunk_size=4096, decode_unicode=False, timeout=-1):
    
    if req._content_consumed:
        raise RuntimeError(
            'The content for this response was already consumed'
        )
    
    req.raw._fp.fp._sock.setblocking(False)
    
    def generate():
        chunk_size = 1        
        while True:
            if timeout < 0:
                rlist,_,_ = select.select([req.raw._fp.fp._sock], [], [])
            else:
                rlist,_,_ = select.select([req.raw._fp.fp._sock], [], [], timeout)
                
            if not rlist:                 
                continue
            
            try:
                chunk = req.raw.read(chunk_size, decode_content=True)            
                if not chunk:
                    break
                if len(chunk) >= chunk_size and chunk_size < max_chunk_size:
                    chunk_size = min(chunk_size*2, max_chunk_size)
                elif len(chunk) < chunk_size/2 and chunk_size < max_chunk_size:
                    chunk_size = max(chunk_size/2,1)
                yield chunk
            except requests.exceptions.SSLError as e:
                if e.errno == 2:
                    # Apparently this means there was nothing in the socket buf
                    pass
                else:
                    raise                
            
        req._content_consumed = True

    gen = generate()

    if decode_unicode:
        gen = stream_decode_response_unicode(gen, req)

    return gen

    
    

class BaseStream(object):

    """A network connection to Twitters streaming API

    :param auth: requests auth object.
    :keyword catchup: Number of tweets from the past to get before switching to
      live stream.
    :keyword raw: If True, return each tweet's raw data direct from the socket,
      without UTF8 decoding or parsing, rather than a parsed object. The
      default is False.
    :keyword timeout: If non-None, set a timeout in seconds on the receiving
      socket. Certain types of network problems (e.g., disconnecting a VPN)
      can cause the connection to hang, leading to indefinite blocking that
      requires kill -9 to resolve. Setting a timeout leads to an orderly
      shutdown in these cases. The default is None (i.e., no timeout).
    :keyword url: Endpoint URL for the object. Note: you should not
      need to edit this. It's present to make testing easier.

    .. attribute:: connected

        True if the object is currently connected to the stream.

    .. attribute:: url

        The URL to which the object is connected

    .. attribute:: starttime

        The timestamp, in seconds since the epoch, the object connected to the
        streaming api.

    .. attribute:: count

        The number of tweets that have been returned by the object.

    .. attribute:: rate

        The rate at which tweets have been returned from the object as a
        float. see also :attr: `rate_period`.

    .. attribute:: rate_period

        The ammount of time to sample tweets to calculate tweet rate. By
        default 10 seconds. Changes to this attribute will not be reflected
        until the next time the rate is calculated. The rate of tweets vary
        with time of day etc. so it's usefull to set this to something
        sensible.

    .. attribute:: user_agent

        User agent string that will be included in the request. NOTE: This can
        not be changed after the connection has been made. This property must
        thus be set before accessing the iterator. The default is set in
        :attr: `USER_AGENT`.
    """

    def __init__(self, auth,
                  raw=False, timeout=-1, url=None, compressed=False, chunk_size=4096, logger=None):
        self._conn = None
        self._rate_ts = None
        self._rate_cnt = 0
        self._auth = auth
        self.raw_mode = raw
        self.timeout = timeout
        self._compressed = compressed

        self.rate_period = 10  # in seconds
        self.connected = False
        self.starttime = None
        self.count = 0
        self.rate = 0
        self.user_agent = USER_AGENT
        self.chunk_size = chunk_size
        if url: self.url = url
        
        self.muststop = False
        self._logger = logger
        
        self._iter = self.__iter__()
         

    def __enter__(self):
        return self

    def __exit__(self, *params):
        self.close()
        return False

    def _init_conn(self):
        """Open the connection to the twitter server"""
        
        if self._logger : self._logger.debug("BaseStream Open the connection to the twitter server")
        
        headers = {'User-Agent': self.user_agent}
        
        if self._compressed:
            headers['Accept-Encoding'] = "deflate, gzip"

        postdata = self._get_post_data() or {}
        postdata['stall_warnings'] = 'true'
        
        if self._logger : self._logger.debug("BaseStream init connection url " + repr(self.url))
        if self._logger : self._logger.debug("BaseStream init connection headers " + repr(headers))
        if self._logger : self._logger.debug("BaseStream init connection data " + repr(postdata))
        
        self._resp = requests.post(self.url, auth=self._auth, headers=headers, data=postdata, stream=True)
        if self._logger : self._logger.debug("BaseStream init connection " + repr(self._resp))
        
        self._resp.raise_for_status()
        self.connected = True

        if not self._rate_ts:
            self._rate_ts = time.time()
        if not self.starttime:
            self.starttime = time.time()


    def _get_post_data(self):
        """Subclasses that need to add post data to the request can override
        this method and return post data. The data should be in the format
        returned by urllib.urlencode."""
        return None

    def testmuststop(self):
        if callable(self.muststop):
            return self.muststop()
        else:
            return self.muststop
    
    def _update_rate(self):
        rate_time = time.time() - self._rate_ts
        if not self._rate_ts or rate_time > self.rate_period:
            self.rate = self._rate_cnt / rate_time
            self._rate_cnt = 0
            self._rate_ts = time.time()
            
    def _iter_object(self):
   
#        for line in self._resp.iter_lines():
#            yield line     
#         pending = None
# 
#         for chunk in self._resp.iter_content(chunk_size=self.chunk_size, decode_unicode=None):
# 
#             if pending is not None:
#                 chunk = pending + chunk 
#             lines = chunk.splitlines()
# 
#             if lines and lines[-1] and chunk and lines[-1][-1] == chunk[-1]:
#                 pending = lines.pop()
#             else:
#                 pending = None
# 
#             for line in lines:
#                 yield line
# 
#         if pending is not None:
#             yield pending
            
        pending = None
        has_stopped = False
         
        if self._logger : self._logger.debug("BaseStream _iter_object")
 
        for chunk in self._resp.iter_content(
            chunk_size=self.chunk_size,
            decode_unicode=None):
 
            if self._logger : self._logger.debug("BaseStream _iter_object loop")
            if self.testmuststop():
                has_stopped = True
                break
 
            if pending is not None:
                chunk = pending + chunk
            lines = chunk.split('\r')
 
            if lines and lines[-1] and chunk and lines[-1][-1] == chunk[-1]:
                pending = lines.pop()
            else:
                pending = None
 
            for line in lines:
                yield line.strip('\n')
 
            if self.testmuststop():
                has_stopped = True
                break
 
        if pending is not None:
            yield pending
        if has_stopped:
            raise StopIteration()
            
    def __iter__(self):
        
        if self._logger : self._logger.debug("BaseStream __iter__")
        if not self.connected:
            if self._logger : self._logger.debug("BaseStream __iter__ not connected, connecting")
            self._init_conn()

        if self._logger : self._logger.debug("BaseStream __iter__ connected")
        
        for line in self._iter_object():

            if self._logger : self._logger.debug("BaseStream __iter__ line %s " % repr(line))
            
            if not line:
                continue

            if (self.raw_mode):
                tweet = line
            else:
                line = line.decode("utf8")
                try:
                    tweet = anyjson.deserialize(line)
                except ValueError:
                    self.close()
                    raise ConnectionError("Got invalid data from twitter", details=line)
            if 'text' in tweet:
                self.count += 1
                self._rate_cnt += 1
            self._update_rate()
            yield tweet


    def next(self):
        """Return the next available tweet. This call is blocking!"""
        return self._iter.next()


    def close(self):
        """
        Close the connection to the streaming server.
        """
        self.connected = False


class FilterStream(BaseStream):
    url = "https://stream.twitter.com/1.1/statuses/filter.json"

    def __init__(self, auth, follow=None, locations=None,
                 track=None, url=None, raw=False, timeout=None, compressed=False, chunk_size=requests.models.ITER_CHUNK_SIZE, logger=None):
        self._follow = follow
        self._locations = locations
        self._track = track
        # remove follow, locations, track
        BaseStream.__init__(self, auth, url=url, raw=raw, timeout=timeout, compressed=compressed, chunk_size=chunk_size, logger=logger)

    def _get_post_data(self):
        postdata = {}
        if self._follow: postdata["follow"] = ",".join([str(e) for e in self._follow])
        if self._locations: postdata["locations"] = ",".join(self._locations)
        if self._track: postdata["track"] = ",".join(self._track)
        return postdata


class SafeStreamWrapper(object):
    
    def __init__(self, base_stream, logger=None, error_cb=None, max_reconnects=-1, initial_tcp_wait=250, initial_http_wait=5000, max_wait=240000):
        self._stream = base_stream
        self._logger = logger
        self._error_cb = error_cb
        self._max_reconnects = max_reconnects
        self._initial_tcp_wait = initial_tcp_wait
        self._initial_http_wait = initial_http_wait
        self._max_wait = max_wait
        self._retry_wait = 0
        self._retry_nb = 0
        self._reconnects = 0

    def __post_process_error(self,e):
        # Note: error_cb is not called on the last error since we
        # raise a ConnectionError instead
        if  callable(self._error_cb):
            self._error_cb(e)
        if self._logger: self._logger.info("stream sleeping for %d ms " % self._retry_wait)
        time.sleep(float(self._retry_wait)/1000.0)
        
        
    def __process_tcp_error(self,e):
        if self._logger: self._logger.debug("connection error type :" + repr(type(e)))
        if self._logger: self._logger.debug("connection error :" + repr(e))
        
        self._reconnects += 1
        if self._max_reconnects >= 0 and self._reconnects > self._max_reconnects:
            raise ConnectionError("Too many retries")
        if self._retry_wait < self._max_wait:
            self._retry_wait += self._initial_tcp_wait
            if self._retry_wait > self._max_wait:
                self._retry_wait = self._max_wait
        
        self.__post_process_error(e)

        
    def __process_http_error(self,e):
        if self._logger: self._logger.debug("http error type %s" % (repr(type(e))))
        if self._logger: self._logger.debug("http error on %s : %s" % (e.response.url,e.message))

        if self._retry_wait < self._max_wait:
            self._retry_wait = 2*self._retry_wait if self._retry_wait > 0 else self._initial_http_wait
            if self._retry_wait > self._max_wait:
                self._retry_wait = self._max_wait
        
        self.__post_process_error(e)
        
    def __iter__(self):
        while not self._stream.testmuststop():
            self._retry_nb += 1
            try:
                if self._logger: self._logger.debug("inner loop")
                for tweet in self._stream:
                    if self._logger: self._logger.debug("tweet : " + repr(tweet))
                    self._reconnects = 0
                    self._retry_wait = 0
                    if not tweet.strip():
                        if self._logger: self._logger.debug("Empty Tweet received : PING")
                        continue
                    yield tweet
            except requests.exceptions.HTTPError as e:
                if e.response.status_code == 401:
                    if self._logger: self._logger.debug("SafeStreamWrapper Connection Error http error on %s : %s" % (e.response.url,e.message))
                    raise AuthenticationError("Error connecting to %s : %s : %s - %s" % (e.response.url,e.message, repr(e.response.headers),repr(e.response.text)))
                if e.response.status_code > 200:
                    self.__process_http_error(e)
                else:
                    self.__process_tcp_error(e)
            except (ConnectionError, requests.exceptions.ConnectionError, requests.exceptions.Timeout, requests.exceptions.RequestException) as e:
                self.__process_tcp_error(e)

        
    
    