init. commit

Author: davidshen84

Dockerfile

@@ -0,0 +1,11 @@
+FROM python:3.6
+MAINTAINER Xi Shen <[email protected]>
+
+COPY requirements.txt /app/
+RUN pip install --requirement /app/requirements.txt
+
+COPY main.py /app/
+
+EXPOSE 80
+WORKDIR /app
+ENTRYPOINT ["python", "main.py"]

__init__.py

@@ -0,0 +1,96 @@
+"""
+Weibo Streaming
+===============
+
+Turn `weibo`_ statues into a HTTP Stream.
+
+- Please refer to *weibo* `API documents`_ for detail weibo statuses API
+- Please refer to `chunked`_ encoding for how HTTP streaming implementation specification
+
+.. _weibo: http://www.weibo.com
+.. _API documents: http://open.weibo.com/wiki/%E5%BE%AE%E5%8D%9AAPI
+.. _chunked: https://en.wikipedia.org/wiki/Chunked_transfer_encoding
+"""
+import logging
+
+from tornado.escape import json_decode
+from tornado.httpclient import AsyncHTTPClient
+
+
+class Fib(object):
+    """
+    Generate Fibonacci sequence
+
+    >>> fib = Fib()
+    >>> fib.next()
+    1
+
+    >>> fib.next()
+    1
+
+    >>> fib.next()
+    2
+    >>> fib.next()
+    3
+    >>> fib.next()
+    5
+    >>> fib.reset()
+    >>> fib.next()
+    1
+    """
+
+    def __init__(self):
+        self._counter = 0
+        self._a = 1
+        self._b = 1
+        self._c = -1
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        self._counter += 1
+        if self._counter <= 2:
+            return 1
+        else:
+            self._c = self._a + self._b
+            self._a = self._b
+            self._b = self._c
+
+            return self._c
+
+    def next(self):
+        return self.__next__()
+
+    def reset(self):
+        self._counter = 0
+        self._a = 1
+        self._b = 1
+        self._c = -1
+
+
+app_logger = logging.getLogger('tornado.application')
+
+
+class WeiboClient(object):
+    _weibo_public_timeline_url = 'https://api.weibo.com/2/statuses/public_timeline.json?access_token={}'
+
+    def __init__(self, access_token):
+        self._http_client = AsyncHTTPClient()
+        self._access_token = access_token
+        self.last_id = 0
+
+    def public_timeline(self):
+        response = yield self._http_client.fetch(WeiboClient._weibo_public_timeline_url.format(self._access_token))
+
+        if response.code == 200:
+            json_body = json_decode(response.body)
+            statuses = [s for s in json_body['statuses'] if s['id'] > self.last_id]
+
+            if len(statuses) > 0:
+                self.last_id = statuses[0]['id']
+
+            return statuses
+        else:
+            app_logger.error('weibo api respond {}'.format(response.code))
+            return []

main.py

@@ -0,0 +1,88 @@
+import logging
+
+from tornado import gen
+from tornado import ioloop
+from tornado import web
+from tornado.escape import json_encode
+from tornado.options import define, options
+
+from . import Fib, WeiboClient
+
+weibo_public_timeline_url = 'https://api.weibo.com/2/statuses/public_timeline.json?access_token={}'
+access_logger = logging.getLogger('tornado.access')
+app_logger = logging.getLogger('tornado.application')
+sleep_duration_origin = 3
+sleep_duration_increase_factor = 2
+
+# chunked transfer encoding delimiter
+CRLF = '\r\n'
+
+# define tornado command line arguments
+define('weibo_access_token')
+define('debug', default=False)
+
+
+def remote_ip(request):
+    return request.headers.get('X-Real-IP') or request.remote_ip
+
+
+# noinspection PyAbstractClass
+class MainHandler(web.RequestHandler):
+    def get(self):
+        self.write('''
+        access /public_timeline to get a weibo public status stream
+        ''')
+
+
+# noinspection PyAbstractClass
+class PublicTimelineHandler(web.RequestHandler):
+    @gen.coroutine
+    def get(self):
+        client = WeiboClient(options.weibo_access_token)
+        access_logger.info('start streaming to {}'.format(remote_ip(self.request)))
+        self.set_header('transfer-encoding', 'chunked')
+        self.set_header('content-type', 'application/json; charset=utf-8')
+        fib = Fib()
+
+        while True:
+            statuses = yield client.public_timeline()
+            if not self.request.connection.stream.closed():
+                statuses_count = len(statuses)
+                if statuses_count > 0:
+                    app_logger.info('received {} new statuses'.format(statuses_count))
+                    for s in statuses:
+                        chunked = json_encode(s)
+                        chunked_size = len(chunked)
+                        self.write('{:x}{}'.format(chunked_size + 1, CRLF))
+                        self.write('{}\n{}'.format(chunked, CRLF))
+                    self.flush()
+                    app_logger.info('last id updated to {}'.format(client.last_id))
+                    fib.reset()
+                    sleep_duration = fib.next()
+                else:
+                    sleep_duration = fib.next()
+                    app_logger.warn('no new statuses (sleeping {} seconds)'.format(sleep_duration))
+
+                yield gen.sleep(sleep_duration)
+            else:
+                # access_logger.info('stop streaming to {}'.format(remote_ip(self.request))
+                return
+                # self.write('0' + CRLF)
+                # self.write(CRLF)
+
+    def on_connection_close(self):
+        access_logger.info('close connection to {}'.format(remote_ip(self.request)))
+        self.finish()
+
+
+if __name__ == '__main__':
+    options.parse_command_line()
+
+    app = web.Application([
+        (r'/', MainHandler),
+        (r'/public_timeline', PublicTimelineHandler)
+    ],
+        debug=options.debug)
+    # listen to default HTTP port
+    app.listen(80)
+    ioloop.IOLoop.current().start()

requirements.txt

@@ -0,0 +1 @@
+tornado==4.4.1