-
Notifications
You must be signed in to change notification settings - Fork 0
/
botkit.py
199 lines (159 loc) · 7.22 KB
/
botkit.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
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
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
import confighandler
import oauthclient
import time
import twitterconnector
import logging
import platformutils
import utils
import os
class BotKit( object ):
def __init__( self, bot_name, debug_mode=False, tweet_interval=None, reply_direct_only=True ):
self._bot_name = bot_name
self._last_mention_id = None
self._debug_mode = debug_mode
self._tweet_interval = tweet_interval
self._reply_direct_only = reply_direct_only
def do_replies( self, connector, generator ):
# get a list of mentions
mentions = None
if self._last_mention_id is not None:
# if we've got a last_mention_id, only reply to mentions since that
mentions = connector.mentions( since_id=self._last_mention_id )
else:
# otherwise, replty to everything
mentions = connector.mentions()
last_id = self._last_mention_id
for mention in mentions:
mention_body = mention.text
process_mention = True
if self._reply_direct_only:
# if reply_direct_only is set, only reply to tweets that begin with this bot's username
self_username = connector.self_username()
process_mention = mention_body.startswith( "@%s" % self_username )
if process_mention:
# strip @usernames from beginning of mentions
if mention_body.startswith( "@" ):
mention_body = mention_body[mention_body.index(" ")+1:].lstrip()
try:
# ask generator for a reply to this mention
tweet = generator.reply(
mention_body,
to_username=mention.author.screen_name,
to_userid=mention.author.id_str
)
# tweet out generated reply
self.do_tweet(
connector,
tweet,
to_username=mention.author.screen_name,
in_reply_to_status_id=mention.id_str
)
except Exception as e:
logging.exception( e )
if (last_id is None) or (mention.id > last_id):
last_id = mention.id
self._last_mention_id = last_id
def do_tweet( self, connector, tweet, to_username=None, in_reply_to_status_id=None ):
if tweet is None:
return
if self._debug_mode:
logging.info( "do_tweet DEBUG MODE, would tweet: %s" % tweet )
elif tweet and (len(tweet)>1):
logging.info( tweet )
# split tweet up into chunk_length sized chunks
prefix = None
chunk_length = 140
if to_username:
# if we're replying, set prefix to @username
# and subtract the length of prefix from chunk_length
prefix = "@%s " % to_username
chunk_length -= len(prefix)
chunks = utils.chunk_string( tweet, chunk_length )
reply_to_status_id = in_reply_to_status_id
# tweet out chunks
for chunk in chunks:
if prefix is not None:
# prepend prefix to each chunk
chunk = "%s%s" % (prefix, chunk)
new_status = connector.tweet(
chunk,
in_reply_to_status_id=reply_to_status_id
)
# each new chunk replies to the last, make threaded replies make sense
reply_to_status_id = new_status.id_str
def run( self, generator ):
# init logging
platformutils.init_logging()
# init config
config = confighandler.ConfigHandler( bot_name=self._bot_name )
# twitter connector & tweet generator
consumer_token = config.twitter_consumer_token()
# get last mention id from environment, if set
last_mention_id = os.environ.get( "LAST_MENTION_ID" )
if last_mention_id is not None:
self._last_mention_id = int(last_mention_id)
# start web UI if enabled
if config.oauth_enabled():
logging.info( "Start oauth interface..." )
# handler for token save from oauth web UI
def save_access_token( token, token_secret ):
global access_token
logging.info( "Token received, saving...")
config.save_twitter_access_token(
token=token,
secret=token_secret
)
# init oauth web UI
oauth = oauthclient.WebInterface(
api_key=consumer_token[0],
api_secret=consumer_token[1],
save_token_callback=save_access_token
)
oauth.start()
RUNNING = True
logging.info( "Enter main runloop..." )
TWEET_INTERVAL = config.tweet_interval( default_value=self._tweet_interval )
try:
# main runloop
while RUNNING:
logging.info( "Main runloop tick...")
# reload access token in case it's changed
access_token = config.twitter_access_token()
if (access_token is not None) and (consumer_token is not None):
logging.info( "Token set, will tweet..." )
# set up a twitter connector
connector = twitterconnector.TwitterConnector(
consumer_key=consumer_token[0],
consumer_secret=consumer_token[1],
access_key=access_token[0],
access_secret=access_token[1]
)
try:
# generate standalone tweet
tweet = None
try:
tweet = generator.generate()
except Exception as e:
logging.exception( e )
self.do_tweet( connector, tweet )
# do replies
self.do_replies( connector, generator )
except Exception as e:
logging.exception( e )
# sleep for TWEET_INTERVAL, potentially a long time
logging.info( "Sleeping for %d seconds..." % TWEET_INTERVAL )
time.sleep( TWEET_INTERVAL )
else:
logging.debug( "No token set, doing nothing..." )
logging.debug(
"-> access_token: %s, consumer_token: %s",
access_token,
consumer_token
)
# sleep for a short period, waiting for token
time.sleep( 1 )
except KeyboardInterrupt:
pass
# close down web UI on exit
if config.oauth_enabled():
oauth.stop()