-
Notifications
You must be signed in to change notification settings - Fork 0
/
doukichan.py
95 lines (84 loc) · 2.54 KB
/
doukichan.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
import json
import requests
import json
import os
import time
from dotenv import load_dotenv
# Load env
load_dotenv()
bearer_token = os.getenv("BEARER_TOKEN")
##chatid = os.getenv("CHATID")
def bearer_oauth(r):
r.headers["Authorization"] = f"Bearer {bearer_token}"
r.headers["User-Agent"] = "v2FilteredStreamPython"
return r
def get_rules():
response = requests.get(
"https://api.twitter.com/2/tweets/search/stream/rules", auth=bearer_oauth
)
if response.status_code != 200:
raise Exception(
"Cannot get rules (HTTP {}): {}".format(response.status_code, response.text)
)
print(json.dumps(response.json()))
return response.json()
def delete_all_rules(rules):
if rules is None or "data" not in rules:
return None
ids = list(map(lambda rule: rule["id"], rules["data"]))
payload = {"delete": {"ids": ids}}
response = requests.post(
"https://api.twitter.com/2/tweets/search/stream/rules",
auth=bearer_oauth,
json=payload
)
if response.status_code != 200:
raise Exception(
"Cannot delete rules (HTTP {}): {}".format(
response.status_code, response.text
)
)
print(json.dumps(response.json()))
def set_rules(delete):
sample_rules = [
{"value" : "from:y_o_m_y_o_m"},
]
payload = {"add": sample_rules}
response = requests.post(
"https://api.twitter.com/2/tweets/search/stream/rules",
auth = bearer_oauth,
json = payload,
)
if response.status_code != 201:
raise Exception(
"Cannot add rules : {} {}".format(response.status_code, response.text)
)
print(json.dumps(response.json()))
def get_stream(set):
tweet_fields = "tweet.fields=public_metrics,entities"
url = "https://api.twitter.com/2/tweets/search/stream?{}&expansions=author_id".format(tweet_fields)
response = requests.get(
url,auth=bearer_oauth, stream=True,
)
#print(response.status_code)
if response.status_code != 200:
raise Exception(
"Cannot add stream, : {} {}".format(response.status_code, response.text)
)
for response_line in response.iter_lines():
if response_line:
json_response = json.loads(response_line)
# print(json.dumps(json_response, indent=4, sort_keys=True))
if "RT " not in json_response["data"]["text"]:
print(json_response["data"]["text"])
fp = open("tweet", "w")
fp.write(json_response["data"]["text"])
fp.close()
os._exit(0)
def main():
rules = get_rules()
delete = delete_all_rules(rules)
set = set_rules(delete)
get_stream(set)
if __name__ == "__main__":
main()