aceproxy

.gitignore

@@ -0,0 +1,5 @@
+*.pyc
+.project
+.pydevproject
+.settings/
+.idea/

aceclient/aceclient.py

@@ -4,8 +4,14 @@
 import telnetlib
 import logging
 import json
+import time
+import threading
+import traceback
+import Queue
+from collections import deque
 from acemessages import *
-
+import aceconfig
+from aceconfig import AceConfig
 
 class AceException(Exception):
 
@@ -49,15 +55,24 @@ def __init__(self, host, port, connect_timeout=5, result_timeout=10):
         self._authevent = Event()
         # Result for getURL()
         self._urlresult = AsyncResult()
+        # Result for GETCID()
+        self._cidresult = AsyncResult()
         # Event for resuming from PAUSE
         self._resumeevent = Event()
         # Seekback seconds.
         self._seekback = 0
         # Did we get START command again? For seekback.
         self._started_again = False
+
+        self._idleSince = time.time()
+        self._lock = threading.Condition(threading.Lock())
+        self._streamReaderConnection = None
+        self._streamReaderState = None
+        self._streamReaderQueue = deque()
+        self._engine_version_code = 0;
 
         # Logger
-        logger = logging.getLogger('AceClient_init')
+        logger = logging.getLogger('AceClieimport tracebacknt_init')
 
         try:
             self._socket = telnetlib.Telnet(host, port, connect_timeout)
@@ -100,8 +115,15 @@ def destroy(self):
         finally:
             self._shuttingDown.set()
 
+    def reset(self):
+        self._started_again = False
+        self._idleSince = time.time()
+        self._streamReaderState = None
+
     def _write(self, message):
         try:
+            logger = logging.getLogger("AceClient_write")
+            logger.debug(message)
             self._socket.write(message + "\r\n")
         except EOFError as e:
             raise AceException("Write error! " + repr(e))
@@ -136,35 +158,50 @@ def aceInit(self, gender=AceConst.SEX_MALE, age=AceConst.AGE_18_24, product_key=
 
     def _getResult(self):
         # Logger
-        logger = logging.getLogger("AceClient_START")
-
         try:
             result = self._result.get(timeout=self._resulttimeout)
             if not result:
-                errmsg = "START error!"
-                logger.error(errmsg)
-                raise AceException(errmsg)
+                raise AceException("Result not received")
         except gevent.Timeout:
-            errmsg = "START timeout!"
-            logger.error(errmsg)
-            raise AceException(errmsg)
+            raise AceException("Timeout")
 
         return result
 
     def START(self, datatype, value):
         '''
         Start video method
         '''
-        self._result = AsyncResult()
+        if self._engine_version_code >= 3010500 and AceConfig.vlcuse:
+           stream_type = 'output_format=hls' + ' transcode_audio=' + str(AceConfig.transcode_audio) \
+                                             + ' transcode_mp3=' + str(AceConfig.transcode_mp3) \
+                                             + ' transcode_ac3=' + str(AceConfig.transcode_ac3)
+        else:
+           stream_type = 'output_format=http' 
+
         self._urlresult = AsyncResult()
+        self._write(AceMessage.request.START(datatype.upper(), value, stream_type))
+        self._getResult()
 
-        self._write(AceMessage.request.LOADASYNC(datatype.upper(), 0, value))
-        contentinfo = self._getResult()
+    def STOP(self):
+        '''
+        Stop video method
+        '''
+        if self._state is not None and self._state != '0':
+            self._result = AsyncResult()
+            self._write(AceMessage.request.STOP)
+            self._getResult()
 
-        self._write(AceMessage.request.START(datatype.upper(), value))
-        self._getResult()
+    def LOADASYNC(self, datatype, url):
+        self._result = AsyncResult()
+        self._write(AceMessage.request.LOADASYNC(datatype.upper(), 0, {'url': url}))
+        return self._getResult()
 
-        return contentinfo
+    def GETCID(self, datatype, url):
+        contentinfo = self.LOADASYNC(datatype, url)
+        self._cidresult = AsyncResult()
+        self._write(AceMessage.request.GETCID(contentinfo.get('checksum'), contentinfo.get('infohash'), 0, 0, 0))
+        cid = self._cidresult.get(True, 5)
+        return '' if not cid or cid == '' else cid[2:]
 
     def getUrl(self, timeout=40):
         # Logger
@@ -178,6 +215,79 @@ def getUrl(self, timeout=40):
             logger.error(errmsg)
             raise AceException(errmsg)
 
+    def startStreamReader(self, url, cid, counter):
+        logger = logging.getLogger("StreamReader")
+        self._streamReaderState = 1
+        logger.debug("Opening video stream: %s" % url)
+
+        try:
+            connection = self._streamReaderConnection = urllib2.urlopen(url)
+
+            if url.endswith('.m3u8'):
+                logger.debug("Can't stream HLS in non VLC mode: %s" % url)
+                return
+
+            if connection.getcode() != 200:
+                logger.error("Failed to open video stream %s" % connection)
+                return
+
+            with self._lock:
+                self._streamReaderState = 2
+                self._lock.notifyAll()
+
+            while True:
+                data = None
+                clients = counter.getClients(cid)
+
+                try:
+                    data = connection.read(AceConfig.readchunksize)
+                except:
+                    break;
+
+                if data and clients:
+                    with self._lock:
+                        if len(self._streamReaderQueue) == AceConfig.readcachesize:
+                            self._streamReaderQueue.popleft()
+                        self._streamReaderQueue.append(data)
+
+                    for c in clients:
+                        try:
+                            c.addChunk(data, 5.0)
+                        except Queue.Full:
+                            if len(clients) > 1:
+                                logger.debug("Disconnecting client: %s" % str(c))
+                                c.destroy()
+                elif not clients:
+                    logger.debug("All clients disconnected - closing video stream")
+                    break
+                else:
+                    logger.warning("No data received")
+                    break
+        except urllib2.URLError:
+            logger.error("Failed to open video stream")
+            logger.error(traceback.format_exc())
+        except:
+            logger.error(traceback.format_exc())
+            if counter.getClients(cid):
+                logger.error("Failed to read video stream")
+        finally:
+            self.closeStreamReader()
+            with self._lock:
+                self._streamReaderState = 3
+                self._lock.notifyAll()
+            counter.deleteAll(cid)
+
+    def closeStreamReader(self):
+        logger = logging.getLogger("StreamReader")
+        c = self._streamReaderConnection
+
+        if c:
+            self._streamReaderConnection = None
+            c.close()
+            logger.debug("Video stream closed")
+
+        self._streamReaderQueue.clear()
+
     def getPlayEvent(self, timeout=None):
         '''
         Blocking while in PAUSE, non-blocking while in RESUME
@@ -201,7 +311,7 @@ def _recvData(self):
             try:
                 self._recvbuffer = self._socket.read_until("\r\n")
                 self._recvbuffer = self._recvbuffer.strip()
-                #logger.debug('<<< ' + self._recvbuffer)
+                # logger.debug('<<< ' + self._recvbuffer)
             except:
                 # If something happened during read, abandon reader.
                 if not self._shuttingDown.isSet():
@@ -213,10 +323,14 @@ def _recvData(self):
                 # Parsing everything only if the string is not empty
                 if self._recvbuffer.startswith(AceMessage.response.HELLO):
                     # Parse HELLO
+                    if 'version_code=' in self._recvbuffer:
+                        v = self._recvbuffer.find('version_code=')
+                        self._engine_version_code = int(self._recvbuffer[v+13:v+20])
+
                     if 'key=' in self._recvbuffer:
                         self._request_key_begin = self._recvbuffer.find('key=')
                         self._request_key = \
-                            self._recvbuffer[self._request_key_begin+4:self._request_key_begin+14]
+                            self._recvbuffer[self._request_key_begin + 4:self._request_key_begin + 14]
                         try:
                             self._write(AceMessage.request.READY_key(
                                 self._request_key, self._product_key))
@@ -246,12 +360,11 @@ def _recvData(self):
                         self._result.set(False)
                     else:
                         logger.debug("Content info: %s", _contentinfo)
-                        _filename = urllib2.unquote(_contentinfo.get('files')[0][0])
-                        self._result.set(_filename)
+                        self._result.set(_contentinfo)
 
                 elif self._recvbuffer.startswith(AceMessage.response.START):
                     # START
-                    if not self._seekback or (self._seekback and self._started_again):
+                    if not self._seekback or self._started_again or not self._recvbuffer.endswith(' stream=1'):
                         # If seekback is disabled, we use link in first START command.
                         # If seekback is enabled, we wait for first START command and
                         # ignore it, then do seeback in first EVENT position command
@@ -263,6 +376,8 @@ def _recvData(self):
                             self._resumeevent.set()
                         except IndexError as e:
                             self._url = None
+                    else:
+                        logger.debug("START received. Waiting for %s." % AceMessage.response.LIVEPOS)
 
                 elif self._recvbuffer.startswith(AceMessage.response.STOP):
                     pass
@@ -290,10 +405,7 @@ def _recvData(self):
                     self._position_last = self._position[2].split('=')[1]
                     self._position_buf = self._position[9].split('=')[1]
                     self._position = self._position[4].split('=')[1]
-                    logger.debug('Current position/last/buf: %s/%s/%s' % (self._position,
-                                                                          self._position_last,
-                                                                          self._position_buf)
-                    )
+
                     if self._seekback and not self._started_again:
                         self._write(AceMessage.request.SEEK(str(int(self._position_last) - \
                             self._seekback)))
@@ -318,6 +430,8 @@ def _recvData(self):
                             AceException(self._status + ' with message ' + self._recvbuffer.split(';')[2]))
                     elif self._status == 'main:starting':
                         self._result.set(True)
+                    elif self._status == 'main:idle':
+                        self._result.set(True)
 
                 elif self._recvbuffer.startswith(AceMessage.response.PAUSE):
                     logger.debug("PAUSE event")
@@ -327,3 +441,7 @@ def _recvData(self):
                     logger.debug("RESUME event")
                     gevent.sleep(self._pausedelay)
                     self._resumeevent.set()
+
+                elif self._recvbuffer.startswith('##') or len(self._recvbuffer) == 0:
+                    self._cidresult.set(self._recvbuffer)
+                    logger.debug("CID: %s" %self._recvbuffer)

aceclient/acemessages.py

@@ -5,8 +5,7 @@
 import hashlib
 import platform
 import urllib2
-
-
+
 class AceConst(object):
     APIVERSION = 3
 
@@ -78,7 +77,7 @@ def LOADASYNC(command, request_id, params_dict):
         # End LOADASYNC
 
         @staticmethod
-        def START(command, params_dict):
+        def START(command, params_dict, stream_type):
             if command == 'TORRENT':
                 return 'START TORRENT ' + str(params_dict.get('url')) + ' ' + \
                     str(params_dict.get('file_indexes', '0')) + ' ' + \
@@ -92,12 +91,11 @@ def START(command, params_dict):
                     str(params_dict.get('file_indexes', '0')) + ' ' + \
                     str(params_dict.get('developer_id', '0')) + ' ' + \
                     str(params_dict.get('affiliate_id', '0')) + ' ' + \
-                    str(params_dict.get('zone_id', '0')) + ' ' + \
-                    str(params_dict.get('stream_id', '0'))
+                    str(params_dict.get('zone_id', '0'))
 
             elif command == 'PID':
                 return 'START PID ' + str(params_dict.get('content_id')) + ' ' + \
-                    str(params_dict.get('file_indexes', '0'))
+                    str(params_dict.get('file_indexes', '0')) + ' ' + stream_type
 
             elif command == 'RAW':
                 return 'START RAW ' + str(params_dict.get('data')) + ' ' + \