-
Notifications
You must be signed in to change notification settings - Fork 0
/
ztwitgw.py
executable file
·239 lines (211 loc) · 9.83 KB
/
ztwitgw.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
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
#!/usr/bin/python3
# Copyright (c) 2008-2011 Mark Eichin <[email protected]>
# See ./LICENSE (MIT style.)
"""take recent twitters and zephyr them to me"""
__version__ = "0.5"
__author__ = "Mark Eichin <[email protected]>"
__license__ = "MIT"
import sys
import os
import getpass
import subprocess
import signal
import time
import tweepy
from lengthener import lengthen
def get_oauth_info(appname=None):
"""get this user's oauth info"""
filebase = os.path.expanduser("~/.ztwit_")
if appname:
# default path is ztwitgw
filebase += appname + "_"
key, secret = open(filebase + "oauth", "r").read().strip().split(":", 1)
return key, secret
# TODO: write get_verifier_X11
def get_verifier_tty(output_path, appname=None):
"""we don't have a verifier, ask the user to use a browser and get one"""
consumer_token, consumer_secret = get_oauth_info(appname=appname)
auth = tweepy.OAuthHandler(consumer_token, consumer_secret)
redirect_url = auth.get_authorization_url() # tweepy.TweepError
print("Open this URL in a browser where you're logged in to twitter:")
print(redirect_url)
verifier = input("Enter (cut&paste) the response code: ")
# use it...
auth.get_access_token(verifier)
# hmm, discard the verifier?
open(output_path, "wb").write(":".join([auth.request_token.key,
auth.request_token.secret,
auth.access_token.key,
auth.access_token.secret,
verifier]))
def get_just_verifier(output_path, appname=None):
"""ask for the verifier *without* having consumer info"""
auth = tweepy.OAuthHandler("", "")
# TODO: this can't work unless we first give the user a redirect to the
# URL to *get* the response code. and possibly not then?
verifier = input("Enter (cut&paste) the response code: ")
# use it...
auth.get_access_token(verifier)
# hmm, discard the verifier?
open(output_path, "wb").write(":".join([auth.request_token.key,
auth.request_token.secret,
auth.access_token.key,
auth.access_token.secret,
verifier]))
def get_oauth_verifier(fallback_mechanism, appname=None):
"""get the request token and verifier, using fallback_mechanism if we don't have one stashed"""
filebase = os.path.expanduser("~/.ztwit_")
if appname:
# default path is ztwitgw
filebase += appname + "_"
verifier_file = filebase + "oauth_verifier"
if not os.path.exists(verifier_file):
fallback_mechanism(verifier_file, appname=appname)
if not os.path.exists(verifier_file):
raise Exception("Fallback Failed")
rt_key, rt_secret, at_key, at_secret, verifier = open(verifier_file, "r").read().strip().split(":", 4)
return rt_key, rt_secret, at_key, at_secret, verifier
# do this with a localhost url?
def zwrite(username, body, tag, status_id=None):
"""deliver one twitter message to zephyr"""
# username... will get encoded when we see one
try:
body = body.encode("iso-8859-1", "xmlcharrefreplace")
except UnicodeDecodeError as ude:
body = repr(body) + ("\n[encode fail: %s]" % ude)
body = body.encode("iso-8859-1", "xmlcharrefreplace")
# example syntax: http://twitter.com/engadget/status/18164103530
zurl = " http://twitter.com/%s/status/%s" % (username, status_id) if status_id else ""
zsig = "%s %s%svia ztwitgw%s" % (username, tag, tag and " ", zurl)
# tag is from codde
cmd = ["zwrite",
"-q", # quiet
"-d", # Don't authenticate
"-s", zsig,
"-c", "%s.twitter" % getpass.getuser(),
"-i", username,
"-m", body]
subprocess.check_call(cmd)
def entity_decode(txt):
"""decode simple entities"""
# TODO: find out what ones twitter considers defined,
# or if sgmllib.entitydefs is enough...
return txt.replace(">", ">").replace("<", "<").replace("&", "&")
# turns out we don't actually see & in practice...
assert entity_decode("-> <3") == "-> <3"
def maybe_lengthen(url):
"""lengthen the url (with an old-ref) *or* leave it untouched"""
new_url = lengthen(url)
if not new_url:
return url
return "%s (via %s )" % (new_url, url)
def slice_substitute(target, offset, low, high, replacement):
"""substitute replacement into target in span low..high; return new target, new offset"""
target = target[:low+offset] + replacement + target[high+offset:]
offset += len(replacement) - (high - low)
return target, offset
assert slice_substitute("abcdefghij", 0, 3, 6, "DEF") == ('abcDEFghij', 0)
assert slice_substitute("abcdefghij", 0, 3, 6, "X") == ('abcXghij', -2)
assert slice_substitute("abcdefghij", 0, 3, 6, "__DEF__") == ('abc__DEF__ghij', 4)
def url_expander(twit, body):
"""expand urls in the body, safely"""
expcount = 0
urlcount = 0
longcount = 0
offset = 0
try:
# https://dev.twitter.com/docs/tweet-entities
# do media later, stick with urls for now
for urlblock in twit.entities.get("urls", []):
low, high = urlblock["indices"]
if urlblock.get("expanded_url"):
body, offset = slice_substitute(body, offset, low, high,
maybe_lengthen(urlblock["expanded_url"]))
expcount += 1
else:
raw_replacement = maybe_lengthen(urlblock["url"])
if raw_replacement != urlblock["url"]:
body, offset = slice_substitute(body, offset, low, high, raw_replacement)
longcount += 1
urlcount += 1
if expcount or urlcount or longcount:
return body + ("\n[expanded %s/%s urls, lengthened %s]" % (expcount, urlcount, longcount))
return body
except Exception as exc:
return body + ("[expander failed: %s]" % exc)
def process_new_twits(api, proto=None, tag=""):
"""process new messages, stashing markers"""
if proto is None:
proto = api.home_timeline
filebase = os.path.expanduser("~/.ztwit_")
if tag:
filebase = filebase + tag + "_"
sincefile = filebase + "since"
since_id = None
if os.path.exists(sincefile):
since_id = open(sincefile, "r").read().strip()
# if since_id: # allow for truncated file
# the iterators *have* a prev, but there's no way to "start" at since_id?
# favorites.json doesn't take an id arg, and it's not like we save anything
# (other than parsing) by walking up and then down, since the json for the
# entire set is loaded anyway...
for twit in reversed(list(tweepy.Cursor(proto, since_id=since_id, include_entities=1).items())):
# reversed?
if not twit:
print("huh? empty twit")
continue
# type(twit) == tweepy.models.Status
# type(twit.author) == tweepy.models.User
who = twit.author.screen_name
what = entity_decode(url_expander(twit, twit.text))
status_id = twit.id_str # to construct a link
zwrite(who, what, tag, status_id)
since_id = status_id
print("Sent:", since_id)
time.sleep(3)
signal.alarm(5*60) # if we're actually making progress, push back the timeout
# Note that since_id is just an ordering - if I favorite an old tweet (even
# something that showed up new because it was freshly retweeted) it doesn't
# show up. This isn't a new bug, I'm just noticing it...
with open(sincefile, "w") as newsince:
print(since_id, file=newsince)
def display_rate_limit(tag, limit):
"""Display rate limit if it isn't at maximum; return available count for convenience"""
if limit["remaining"] != limit["limit"]:
print(limit["remaining"], "out of", limit["limit"], tag, "queries available")
reset_time = limit["reset"]
print(" Will reset in", reset_time - time.time(), "seconds, at", time.ctime(reset_time))
return limit["remaining"]
if __name__ == "__main__":
# This conflicts with the long-lag retry mode, so just turn it off for now
# signal.alarm(5*60) # been seeing some hangs, give up after a bit
prog, = sys.argv
rt_key, rt_secret, at_key, at_secret, verifier = get_oauth_verifier(get_verifier_tty)
consumer_token, consumer_secret = get_oauth_info()
auth = tweepy.OAuthHandler(consumer_token, consumer_secret)
auth.set_request_token(rt_key, rt_secret)
auth.set_access_token(at_key, at_secret)
print("ct:", consumer_token)
print("cs:", consumer_secret)
print("rk:", rt_key)
print("rs:", rt_secret)
print("vf:", verifier)
print("ak:", at_key)
print("as:", at_secret)
# request limits reset every 15 minutes, so retry in 16
# retry 10 times to allow us to get 3000 messages behind
# set the timeout to match the retry count
api = tweepy.API(auth, retry_delay=16*60, retry_count=10, timeout=160*60)
limits = api.rate_limit_status()
home_left = display_rate_limit("home", limits["resources"]["statuses"]["/statuses/home_timeline"])
ment_left = display_rate_limit("mentions", limits["resources"]["statuses"]["/statuses/mentions_timeline"])
fave_left = display_rate_limit("favorites", limits["resources"]["favorites"]["/favorites/list"])
if home_left > 0:
process_new_twits(api)
if ment_left > 0:
process_new_twits(api, proto=api.mentions_timeline, tag="reply")
# replies_url = "http://twitter.com/statuses/replies.json"
# but that's not in tweepy... try hacking it?
# hmm, not in http://apiwiki.twitter.com/w/page/22554679/Twitter-API-Documentation either
if fave_left > 0:
process_new_twits(api, proto=api.favorites, tag="favorites")