streaming.py 5.42 KB
Newer Older
1
2
3
4
5
# Tweepy
# Copyright 2009 Joshua Roesslein
# See LICENSE

import httplib
6
from socket import timeout
7
from threading import Thread
8
from time import sleep
9
10

from . auth import BasicAuthHandler
11
12
from . parsers import parse_status
from . api import API
13
from . error import TweepError
14
15

try:
16
    import json #Python >= 2.6
17
except ImportError:
18
19
20
21
22
23
24
    try:
        import simplejson as json #Python < 2.6
    except ImportError:
        try:
            from django.utils import simplejson as json #Google App Engine
        except ImportError:
            raise ImportError, "Can't load a json library"
25

26
27
STREAM_VERSION = 1

Josh Roesslein's avatar
Josh Roesslein committed
28

29
30
class StreamListener(object):

Josh Roesslein's avatar
Josh Roesslein committed
31
32
33
34
35
36
37
    def on_status(self, status):
        """Called when a new status arrives"""
        return

    def on_delete(self, status_id, user_id):
        """Called when a delete notice arrives for a status"""
        return
38

Josh Roesslein's avatar
Josh Roesslein committed
39
40
41
    def on_limit(self, track):
        """Called when a limitation notice arrvies"""
        return
42

Josh Roesslein's avatar
Josh Roesslein committed
43
44
45
    def on_error(self, status_code):
        """Called when a non-200 status code is returned"""
        return False
46

Josh Roesslein's avatar
Josh Roesslein committed
47
48
49
    def on_timeout(self):
        """Called when stream connection times out"""
        return
50

51

52
53
class Stream(object):

Josh Roesslein's avatar
Josh Roesslein committed
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
    host = 'stream.twitter.com'

    def __init__(self, username, password, listener, timeout=5.0, retry_count = None,
                    retry_time = 10.0, snooze_time = 5.0, buffer_size=1500):
        self.auth = BasicAuthHandler(username, password)
        self.running = False
        self.timeout = timeout
        self.retry_count = retry_count
        self.retry_time = retry_time
        self.snooze_time = snooze_time
        self.buffer_size = buffer_size
        self.listener = listener
        self.api = API()

    def _run(self):
        # setup
        headers = {}
        self.auth.apply_auth(None, None, headers, None)

        # enter loop
        error_counter = 0
        conn = None
        while self.running:
            if self.retry_count and error_counter > self.retry_count:
                # quit if error count greater than retry count
                break
            try:
                conn = httplib.HTTPConnection(self.host)
                conn.connect()
                conn.sock.settimeout(self.timeout)
                conn.request('POST', self.url, headers=headers)
                resp = conn.getresponse()
                if resp.status != 200:
                    if self.listener.on_error(resp.status) is False:
                        break
                    error_counter += 1
                    sleep(self.retry_time)
                else:
                    error_counter = 0
                    self._read_loop(resp)
            except timeout:
                if self.listener.on_timeout() == False:
                    break
                if self.running is False:
                    break
                conn.close()
                sleep(self.snooze_time)
            except Exception:
                # any other exception is fatal, so kill loop
                break

        # cleanup
        self.running = False
        if conn:
            conn.close()

    def _read_loop(self, resp):
        data = ''
        while self.running:
            if resp.isclosed():
                break

            # read length
            length = ''
            while True:
                c = resp.read(1)
                if c == '\n':
                    break
                length += c
            length = length.strip()
            if length.isdigit():
                length = int(length)
            else:
                continue

            # read data
            data = resp.read(length)

            # turn json data into status object
            if 'in_reply_to_status_id' in data:
                status = parse_status(data, self.api)
                if self.listener.on_status(status) == False:
                    self.running = False
            elif 'delete' in data:
                delete = json.loads(data)['delete']['status']
                if self.listener.on_delete(delete['id'], delete['user_id']) == False:
                    self.running = False
            elif 'limit' in data:
                if self.listener.on_limit(json.loads(data)['limit']['track']) == False:
                    self.running = False

    def firehose(self, count=None):
        if self.running:
            raise TweepError('Stream object already connected!')
        self.url = '/%i/statuses/firehose.json?delimited=length' % STREAM_VERSION
        if count:
            self.url += '&count=%s' % count
        self.running = True
        Thread(target=self._run).start()

    def sample(self, count=None):
        if self.running:
            raise TweepError('Stream object already connected!')
        self.url = '/%i/statuses/sample.json?delimited=length' % STREAM_VERSION
        if count:
            self.url += '&count=%s' % count
        self.running = True
        Thread(target=self._run).start()

    def filter(self, follow=None, track=None):
        if self.running:
            raise TweepError('Stream object already connected!')
        self.url = '/%i/statuses/filter.json?delimited=length' % STREAM_VERSION
        if follow:
            self.url += '&follow=%s' % ','.join(follow)
        if track:
            self.url += '&track=%s' % ','.join(track)
        print self.url
        self.running = True
        Thread(target=self._run).start()

    def disconnect(self):
176
        if self.running is False:
Josh Roesslein's avatar
Josh Roesslein committed
177
178
179
            return
        self.running = False