Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Newer
Older
100644 259 lines (205 sloc) 7.396 kb
35eccd1 start
George Buckenham authored
1 """
2 twitterbot
3
4 A twitter IRC bot. Twitterbot connected to an IRC server and idles in
5 a channel, polling a twitter account and broadcasting all updates to
6 friends.
7
8 USAGE
9
10 twitterbot [config_file]
11
12 CONFIG_FILE
13
14 The config file is an ini-style file that must contain the following:
15
16 [twitter]
17 email: <twitter_account_email>
18 password: <twitter_account_password>
19
20 and optionally:
21 [debug]
22 debug: False
23
24 If no config file is given "twitterbot.ini" will be used by default.
25 """
26
27 BOT_VERSION = "TwitterBot 1.0 (http://mike.verdone.ca/twitter)"
28
29 IRC_BOLD = chr(0x02)
30 IRC_ITALIC = chr(0x16)
31 IRC_UNDERLINE = chr(0x1f)
32 IRC_REGULAR = chr(0x0f)
33
34 import sys
35 import time
36 from dateutil.parser import parse
37 from ConfigParser import SafeConfigParser
38 from heapq import heappop, heappush
39 import traceback
40 import os.path
41 import twitter
42
43 from twitter import TwitterError
44 import re
45 from htmlentitydefs import name2codepoint
46
34e4843 ResponsesDict
Boris the Brave authored
47 from bearuser import BearUser
48
35eccd1 start
George Buckenham authored
49
50
51
52 def htmlentitydecode(s):
53 return re.sub(
54 '&(%s);' % '|'.join(name2codepoint),
55 lambda m: unichr(name2codepoint[m.group(1)]), s)
56
57 debug_flag = False
58
59 def debug(msg):
60 # uncomment this for debug text stuff
61 if debug_flag:
62 print >> sys.stderr, msg
63 else:
64 pass
65
66
67 class SchedTask(object):
68 def __init__(self, task, delta, repeat):
69 self.task = task
70 self.delta = delta
71 self.next = time.time()
72 self.repeat = repeat
73
74 def __repr__(self):
75 return "<SchedTask %s next:%i delta:%i>" %(
76 self.task.__name__, self.next, self.delta)
77
78 def __cmp__(self, other):
79 return cmp(self.next, other.next)
80
81 def __call__(self):
82 return self.task()
83
84 class Scheduler(object):
85 def __init__(self, tasks):
86 self.task_heap = []
87 for task in tasks:
88 heappush(self.task_heap, task)
89
90 def next_task(self):
91 debug("next_task")
92 now = time.time()
93 task = heappop(self.task_heap)
94 wait = task.next - now
95 task.next = now + task.delta
96 if task.repeat:
97 heappush(self.task_heap, task)
98 if (wait > 0):
99 time.sleep(wait)
100 task()
101 debug("tasks: " + str(self.task_heap))
102
103 def add_task(self, task):
104 heappush(self.task_heap, task)
105
106 def run_forever(self):
107 while True:
108 self.next_task()
9f595ee Re-organize data
Boris the Brave authored
109
35eccd1 start
George Buckenham authored
110 class TwitterBot(object):
111 def __init__(self, configFilename):
112 self.configFilename = configFilename
113 self.config = load_config(self.configFilename)
114 self.twitter = twitter.Api(
115 username=self.config.get('twitter', 'email'),
116 password=self.config.get('twitter', 'password'))
117
118 self.sched = Scheduler(
119 #(SchedTask(self.process_events, 1),
120 (SchedTask(self.check_dms, 120, True),
121 SchedTask(self.start_game_to_v21, 30, False),
122 SchedTask(self.check_replies, 30, True)))
123 # SchedTask(self.stay_joined, 120)))
124 self.lastDMsUpdate = time.gmtime()
125 self.lastRepliesUpdate = time.gmtime()
126 self.lastUpdate = time.gmtime()
127
4f10d0e added bearuser class
George Buckenham authored
128 self.bearUserDict = {}
129
35eccd1 start
George Buckenham authored
130 def start_game_to_v21(self):
131 self.start_game("v21", "you should really generalize this bit")
132 def start_game(self, follower, message):
b3b326b fixed some things, added a test thing when it starts up
George Buckenham authored
133 class update:
134 sender_screen_name = "v21"
135 text = "i think youre a terrible person"
136
137 self.handle_dm(update)
35eccd1 start
George Buckenham authored
138 try:
139 self.twitter.CreateFriendship(follower)
140 self.twitter.PostDirectMessage(follower, message)
141 #some kind of database action
142 except Exception, e:
143 print >> sys.stderr, "Exception while querying twitter:"
144 traceback.print_exc(file=sys.stderr)
145 return
146
147 def check_dms(self):
4f10d0e added bearuser class
George Buckenham authored
148 debug("In check_dms")
35eccd1 start
George Buckenham authored
149 try:
150 updates = self.twitter.GetDirectMessages()
151 except Exception, e:
152 print >> sys.stderr, "Exception while querying twitter:"
153 traceback.print_exc(file=sys.stderr)
154 return
155
156 nextLastUpdate = self.lastDMsUpdate
157 for update in updates:
158 crt = parse(update.created_at).utctimetuple()
159 if (crt > self.lastUpdate):
160 text = (htmlentitydecode(
161 update.text.replace('\n', ' '))
162 .encode('utf-8', 'replace'))
163 self.handle_dm(update)
164
165 nextLastUpdate = crt
166 else:
167 break
168 self.lastDMsUpdate = nextLastUpdate
169
170 def check_replies(self):
171 debug("In check_replies")
172 try:
173 updates = self.twitter.GetReplies()
174 except Exception, e:
175 print >> sys.stderr, "Exception while querying twitter:"
176 traceback.print_exc(file=sys.stderr)
177 return
178
179 nextLastUpdate = self.lastRepliesUpdate
180 for update in updates:
181 crt = parse(update.created_at).utctimetuple()
182 if (crt > self.lastUpdate):
183 text = (htmlentitydecode(
184 update.text.replace('\n', ' '))
185 .encode('utf-8', 'replace'))
186 self.handle_replies(update)
187
188 nextLastUpdate = crt
189 else:
190 break
191 self.lastRepliesUpdate = nextLastUpdate
192
193
194 def handle_dm(self, update):
4f10d0e added bearuser class
George Buckenham authored
195 user = update.sender_screen_name
196
b3b326b fixed some things, added a test thing when it starts up
George Buckenham authored
197 if not user in self.bearUserDict:
4f10d0e added bearuser class
George Buckenham authored
198 self.bearUserDict[user] = BearUser(user=self.twitter.GetUser(user=user))
b3b326b fixed some things, added a test thing when it starts up
George Buckenham authored
199 message = self.bearUserDict[user].createReply(update.text)
200 self.twitter.PostDirectMessage(user, message)
35eccd1 start
George Buckenham authored
201
202 def handle_replies(self, update):
b3b326b fixed some things, added a test thing when it starts up
George Buckenham authored
203 user = update.user.screen_name
4f10d0e added bearuser class
George Buckenham authored
204
b3b326b fixed some things, added a test thing when it starts up
George Buckenham authored
205 if not user in self.bearUserDict:
4f10d0e added bearuser class
George Buckenham authored
206 self.bearUserDict[user] = BearUser(user=self.twitter.GetUser(user=user))
207 message = self.bearUserDict[update.user].createReply(update.text)
35eccd1 start
George Buckenham authored
208
209 self.twitter.PostUpdate(status="@%s %s"%(update.user.screen_name, message), in_reply_to_status_id=update.id)
210
211 def run(self):
212 while True:
213 try:
214 self.sched.run_forever()
215 except KeyboardInterrupt:
216 break
217 except TwitterError, e:
218 # twitter.com is probably down because it sucks. ignore the fault and keep going
219 print >> sys.stderr, e
220 #except Exception, e:
221 # print >> sys.stderr, e
222
223 def load_config(filename):
224 defaults = dict(debug=dict(debug=False))
225 cp = SafeConfigParser(defaults)
226 cp.read((filename,))
227
228 # attempt to read these properties-- they are required
229 cp.get('twitter', 'email'),
230 cp.get('twitter', 'password')
231 try:
232 global debug_flag
233 debug_flag = cp.getboolean('debug', 'debug')
234 except: #debug is optional
235 pass
236
237 return cp
238
239 def main():
240 configFilename = "twitterbot.ini"
241 if (sys.argv[1:]):
242 configFilename = sys.argv[1]
243
244 try:
245 if not os.path.exists(configFilename):
246 raise Exception()
247 load_config(configFilename)
248 except Exception, e:
249 print >> sys.stderr, "Error while loading ini file %s" %(
250 configFilename)
251 print >> sys.stderr, e
252 print >> sys.stderr, __doc__
253 sys.exit(1)
254
255 bot = TwitterBot(configFilename)
256 return bot.run()
257
258 main()
Something went wrong with that request. Please try again.