this repo has no description
New mentions-timeline command, refs #26
+59
-19
+53
-5
twitter_to_sqlite/cli.py
+53
-5
twitter_to_sqlite/cli.py
···
251
251
)
252
252
def home_timeline(db_path, auth, since, since_id):
253
253
"Save tweets from timeline for authenticated user"
254
+
_shared_timeline(
255
+
db_path,
256
+
auth,
257
+
since,
258
+
since_id,
259
+
table="timeline_tweets",
260
+
api_url="https://api.twitter.com/1.1/statuses/home_timeline.json",
261
+
)
262
+
263
+
264
+
@cli.command(name="mentions-timeline")
265
+
@click.argument(
266
+
"db_path",
267
+
type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
268
+
required=True,
269
+
)
270
+
@click.option(
271
+
"-a",
272
+
"--auth",
273
+
type=click.Path(file_okay=True, dir_okay=False, allow_dash=True, exists=True),
274
+
default="auth.json",
275
+
help="Path to auth.json token file",
276
+
)
277
+
@click.option(
278
+
"--since",
279
+
is_flag=True,
280
+
default=False,
281
+
help="Pull tweets since last retrieved mention",
282
+
)
283
+
@click.option(
284
+
"--since_id", type=str, default=False, help="Pull mentions since this Tweet ID"
285
+
)
286
+
def mentions_timeline(db_path, auth, since, since_id):
287
+
"Save tweets that mention the authenticated user"
288
+
_shared_timeline(
289
+
db_path,
290
+
auth,
291
+
since,
292
+
since_id,
293
+
table="mentions_tweets",
294
+
api_url="https://api.twitter.com/1.1/statuses/mentions_timeline.json",
295
+
sleep=10,
296
+
)
297
+
298
+
299
+
def _shared_timeline(db_path, auth, since, since_id, table, api_url, sleep=1):
254
300
if since and since_id:
255
301
raise click.ClickException("Use either --since or --since_id, not both")
256
302
auth = json.load(open(auth))
···
258
304
db = utils.open_database(db_path)
259
305
profile = utils.get_profile(db, session)
260
306
expected_length = 800
261
-
if since and db["timeline_tweets"].exists:
307
+
if since and db[table].exists:
262
308
# Set since_id to highest value for this timeline
263
309
try:
264
310
since_id = db.conn.execute(
265
-
"select max(tweet) from timeline_tweets where user = ?", [profile["id"]]
311
+
"select max(tweet) from {} where user = ?".format(table),
312
+
[profile["id"]],
266
313
).fetchall()[0][0]
267
314
expected_length = None
268
315
except IndexError:
269
316
pass
317
+
270
318
with click.progressbar(
271
-
utils.fetch_home_timeline(session, since_id=since_id),
319
+
utils.fetch_timeline(session, api_url, sleep=sleep, since_id=since_id),
272
320
length=expected_length,
273
-
label="Importing timeline",
321
+
label="Importing tweets",
274
322
show_pos=True,
275
323
) as bar:
276
324
# Save them 100 at a time
277
325
def save_chunk(db, chunk):
278
326
utils.save_tweets(db, chunk)
279
327
# Record who's timeline they came from
280
-
db["timeline_tweets"].upsert_all(
328
+
db[table].upsert_all(
281
329
[{"user": profile["id"], "tweet": tweet["id"]} for tweet in chunk],
282
330
pk=("user", "tweet"),
283
331
foreign_keys=("user", "tweet"),
+6
-14
twitter_to_sqlite/utils.py
+6
-14
twitter_to_sqlite/utils.py
···
89
89
return profile
90
90
91
91
92
-
def fetch_timeline(session, url, args, sleep=1, stop_after=None, key=None):
92
+
def fetch_timeline(
93
+
session, url, args=None, sleep=1, stop_after=None, key=None, since_id=None
94
+
):
93
95
# See https://developer.twitter.com/en/docs/tweets/timelines/guides/working-with-timelines
94
-
args = dict(args)
96
+
args = dict(args or {})
95
97
args["count"] = 200
96
98
if stop_after is not None:
97
99
args["count"] = stop_after
100
+
if since_id:
101
+
args["since_id"] = since_id
98
102
args["tweet_mode"] = "extended"
99
103
min_seen_id = None
100
104
num_rate_limit_errors = 0
···
139
143
args,
140
144
sleep=1,
141
145
stop_after=stop_after,
142
-
)
143
-
144
-
145
-
def fetch_home_timeline(session, since_id=None):
146
-
args = {}
147
-
if since_id:
148
-
args["since_id"] = since_id
149
-
yield from fetch_timeline(
150
-
session,
151
-
"https://api.twitter.com/1.1/statuses/home_timeline.json",
152
-
args,
153
-
sleep=1,
154
146
)
155
147
156
148