Repository navigation
Expand file tree
/
Copy pathsearch_tweet_type.py
More file actions
155 lines (135 loc) · 6.35 KB
/
Copy pathsearch_tweet_type.py
File metadata and controls
155 lines (135 loc) · 6.35 KB
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
# Enterprise Search Tweets - Data request with Tweet type function to demonstrate how to classify Tweets
# Supports data request only (not counts) and returns parsed Tweet payload (only select fields)
import argparse
import json
import os
import sys
import requests
from dotenv import load_dotenv
load_dotenv(verbose=True) # Throws error if it can't find .env file
# Argparse for cli options. Run `python search_tweet_type.py -h` to see the list of arguments.
parser = argparse.ArgumentParser()
parser.add_argument("-r", "--request_file", help="Use json file for request body",
action="store_true")
parser.add_argument("-q", "--query", help="A valid query up to 2,048 characters")
parser.add_argument("-f", "--from_date", help="Oldest date from which results will be provided")
parser.add_argument("-t", "--to_date", help="Most recent date to which results will be provided")
parser.add_argument("-m", "--max_results", help="Maximum number of results returned by a single\
request/response cycle (range: 10-500, default: 100)")
parser.add_argument("-b", "--bucket", choices=['day', 'hour', 'minute'],
help="The unit of time for which count data will be provided.")
parser.add_argument("-n", "--next", help="Auto paginate through next tokens", action="store_true")
args = parser.parse_args()
# Retrieves and stores credential information from the '.env' file
USERNAME = os.getenv("USERNAME")
PASSWORD = os.getenv("PASSWORD")
ACCOUNT_NAME = os.getenv("ACCOUNT_NAME")
ENDPOINT_LABEL = os.getenv("SEARCH_LABEL")
ARCHIVE = os.getenv ("SEARCH_ARCHIVE")
def main():
search_endpoint = f"https://gnip-api.twitter.com/search/{ARCHIVE}/accounts/{ACCOUNT_NAME}/{ENDPOINT_LABEL}.json"
# Build request body from file if it exists, else use cli args
request_body = build_request_body()
# Make first request
first_response = make_request(search_endpoint, request_body)
# Deserialize json response
json_response = (json.loads(first_response.text))
# Create Python dict from results list
tweet_results = json_response["results"]
parsed_results = { "parsed_results": [] }
# Loop through Tweet results to test for type, extended Tweet, and parse JSON
for tweet in tweet_results:
extended_tweet = check_for_extended_tweet(tweet)
tweet_type = determine_tweet_type(tweet)
if extended_tweet is True:
text = tweet["extended_tweet"]["full_text"]
else:
text = tweet["text"]
custom_dict = {
"tweet_id": tweet["id_str"],
"text": text,
"tweet_type": tweet_type,
"hyperlink": "https://twitter.com/twitter/status/" + tweet["id_str"]
}
parsed_results["parsed_results"].append(custom_dict) # Add Tweet to parsed_results list
print(json.dumps(parsed_results, indent=2, sort_keys=True))
# Pagination logic (if -n flag is passed, paginate through the results)
if json_response.get("next") is None or args.next is False:
print(f"Request complete.")
elif json_response.get("next") is not None and args.next:
next_token = json_response.get("next")
request_count = 1 # Keep track of the number of requests being made (pagination)
while next_token is not None:
# Update request_body with next token
request_body.update(next=next_token)
# Make the request with the next token
response = make_request(search_endpoint, request_body)
parsed_results = { "parsed_results": [] }
# Loop through Tweet results to test for type, extended Tweet, and parse JSON
for tweet in tweet_results:
extended_tweet = check_for_extended_tweet(tweet)
tweet_type = determine_tweet_type(tweet)
if extended_tweet is True:
text = tweet["extended_tweet"]["full_text"]
else:
text = tweet["text"]
custom_dict = {
"tweet_id": tweet["id_str"],
"text": text,
"tweet_type": tweet_type,
"hyperlink": "https://twitter.com/twitter/status/" + tweet["id_str"]
}
parsed_results["parsed_results"].append(custom_dict) # Add Tweet to parsed_results
print(json.dumps(parsed_results, indent=2, sort_keys=True))
# Parse n response and it's 'next' token
n_response = (json.loads(response.text))
next_token = n_response.get("next")
# Iterates the request counter
request_count += 1
print(f"Done paginating.\nTotal requests made: {request_count}")
def build_request_body():
# Request file will override CLI options
if args.request_file is True:
with open("request.json", "r") as read_file:
request_body = json.load(read_file)
else:
request_body = {}
if args.query:
request_body.update(query=args.query)
if args.from_date:
request_body.update(fromDate=args.from_date)
if args.to_date:
request_body.update(toDate=args.to_date)
if args.max_results:
request_body.update(maxResults=args.max_results)
if args.bucket:
request_body.update(bucket=args.bucket)
return request_body
def make_request(endpoint, request_body):
try:
response = requests.post(url=endpoint, auth=(USERNAME, PASSWORD), json=request_body)
except requests.exceptions.RequestException as e:
print(e)
sys.exit(120)
return response
def determine_tweet_type(tweet):
# Check for reply indicator first
if tweet["in_reply_to_status_id"] is not None:
tweet_type = "Reply Tweet"
# Check boolean quote status field but make sure it's not a Retweet (of a Quote Tweet)
elif tweet["is_quote_status"] is True and not tweet["text"].startswith("RT"):
tweet_type = "Quote Tweet"
# Check both indicators of a Retweet
elif tweet["text"].startswith("RT") and tweet.get("retweeted_status") is not None:
tweet_type = "Retweet"
else:
tweet_type = "Original Tweet"
return tweet_type
def check_for_extended_tweet(tweet):
try:
value = tweet["extended_tweet"]
return True
except KeyError:
return False
if __name__ == '__main__':
main()