Last active
September 30, 2018 12:59
-
-
Save YuzuRyo61/65a2f06e12b1118c76bbf1989cea2810 to your computer and use it in GitHub Desktop.
端末で見れるMastodonStreaming
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
# -*- coding: utf-8 -*- | |
""" | |
初回セットアップスクリプトを追加したスクリプトです。これ一つでタイムラインをリアルタイムで見ることができるようになります。 | |
""" | |
""" | |
Code by YuzuRyo61. | |
mastodon.timeline.read.py の改良型。差分がそこそこあるので新しく貼り直し。 | |
CWの表記追加と、引数で使用するStreamを選択することができるようにしました。 | |
ちなみに、Streamの選択優先度は"home > local > public"の順になってます。 | |
""" | |
import sys, warnings, traceback, re, os | |
import webbrowser, requests | |
from mastodon import Mastodon,StreamListener | |
from xml.sax.saxutils import unescape | |
from urllib.parse import urlencode | |
# Replace | |
def Replacetext(text): | |
text = re.sub('<p>|</p>|<a.+"tag">|<a.+"_blank">|<a.+mention">|<span>|</span>|</a>|<span class="[a-z-]+">|<span class="">', "", str(text)) | |
text = re.sub('<br>|<br />', "\n", str(text)) | |
text = text.replace("'", "'") | |
text = text.replace(""", "\"") | |
text = unescape(text) | |
return (text) | |
# StreamListener | |
class SListen(StreamListener): | |
def on_update(self, status): | |
try: | |
account = status["account"] | |
non_bmp_map = dict.fromkeys(range(0x10000, sys.maxunicode + 1), 0x0000) | |
if str(status["spoiler_text"]) == "" : | |
print("=== " + str((account["display_name"]).translate(non_bmp_map)) + " @" +str((account["acct"]).translate(non_bmp_map)) + " ===") | |
print(Replacetext(str(status["content"])).translate(non_bmp_map)) | |
else: | |
print("=!= " + str((account["display_name"]).translate(non_bmp_map)) + " @" +str((account["acct"]).translate(non_bmp_map)) + " =!=") | |
print("SPOILER: " + Replacetext(str(status["spoiler_text"])).translate(non_bmp_map)) | |
print("==========") | |
print(Replacetext(str(status["content"])).translate(non_bmp_map)) | |
except KeyboardInterrupt: | |
print("Connection Closing") | |
quit() | |
except: | |
print("FATAL: \n" + traceback.format_exc()) | |
pass | |
def on_delete(self, status_id): | |
print("=-= Deleted =-=\n" + str(status_id)) | |
# Notify StreamListener | |
class NotifySListen(StreamListener): | |
def on_notification(self, notification): | |
try: | |
if notification["type"] in "mention": | |
account = notification["account"] | |
status = notification["status"] | |
non_bmp_map = dict.fromkeys(range(0x10000, sys.maxunicode + 1), 0x0000) | |
if str(status["spoiler_text"]) == "": | |
print("=*= Mention: @" + str((account["acct"]).translate(non_bmp_map)) + " =*=") | |
print(Replacetext(str(status["content"])).translate(non_bmp_map)) | |
else: | |
print("=*= Mention: @" + str((account["acct"]).translate(non_bmp_map)) + " =*=") | |
print("SPOILER: " + Replacetext(str(status["spoiler_text"])).translate(non_bmp_map)) | |
print("==========") | |
print(Replacetext(str(status["content"])).translate(non_bmp_map)) | |
elif notification["type"] in "reblog": | |
account = notification["account"] | |
status = notification["status"] | |
non_bmp_map = dict.fromkeys(range(0x10000, sys.maxunicode + 1), 0x0000) | |
if str(status["spoiler_text"]) == "": | |
print("=*= Boost: @" + str((account["acct"]).translate(non_bmp_map)) + " =*=") | |
print(Replacetext(str(status["content"])).translate(non_bmp_map)) | |
else: | |
print("=*= Boost: @" + str((account["acct"]).translate(non_bmp_map)) + " =*=") | |
print("SPOILER: " + Replacetext(str(status["spoiler_text"])).translate(non_bmp_map)) | |
print("==========") | |
print(Replacetext(str(status["content"])).translate(non_bmp_map)) | |
elif notification["type"] in "favourite": | |
account = notification["account"] | |
status = notification["status"] | |
non_bmp_map = dict.fromkeys(range(0x10000, sys.maxunicode + 1), 0x0000) | |
if str(status["spoiler_text"]) == "": | |
print("=*= Favourite: @" + str((account["acct"]).translate(non_bmp_map)) + " =*=") | |
print(Replacetext(str(status["content"])).translate(non_bmp_map)) | |
else: | |
print("=*= Favourite: @" + str((account["acct"]).translate(non_bmp_map)) + " =*=") | |
print("SPOILER: " + Replacetext(str(status["spoiler_text"])).translate(non_bmp_map)) | |
print("==========") | |
print(Replacetext(str(status["content"])).translate(non_bmp_map)) | |
elif notification["type"] in "follow": | |
account = notification["account"] | |
non_bmp_map = dict.fromkeys(range(0x10000, sys.maxunicode + 1), 0x0000) | |
print("=*= Follow: @" + str((account["acct"]).translate(non_bmp_map)) + " =*=") | |
print("Name: " + Replacetext(str(account["display_name"])).translate(non_bmp_map) + "\n") | |
print(Replacetext(str(account["note"])).translate(non_bmp_map)) | |
pass | |
else: | |
print(notification) | |
pass | |
except KeyboardInterrupt: | |
print("Connection Closing") | |
quit() | |
except: | |
print("FATAL: \n" + traceback.format_exc()) | |
pass | |
if __name__ == "__main__": | |
# Check exists depend files | |
instance = None | |
client = None | |
secret = None | |
if not os.path.isfile("instance.txt") and not os.path.isfile("key.txt"): | |
print("*** WARNING, DEPENDCY FILES NOT FOUND! ***") | |
print("Start setup.") | |
instance = input("Your instance address: ") | |
print("Sending create app API request to {}...".format(str(instance))) | |
Mastodon.create_app( | |
'Python-TimeLineReader', | |
scopes=['read'], | |
api_base_url = str(instance), | |
to_file = 'key.txt' | |
) | |
with open("instance.txt", "w") as w: | |
w.write(str(instance)) | |
with open("key.txt", 'r') as k: | |
client = k.readline() | |
secret = k.readline() | |
print("Created key.txt and instance.txt") | |
if not os.path.isfile("userkey.txt"): | |
if client == None or secret == None or instance == None: | |
with open("instance.txt", 'r') as i: | |
instance = i.readline() | |
with open("key.txt", 'r') as k: | |
client = k.readline() | |
secret = k.readline() | |
client = re.sub('\n', '', client) | |
secret = re.sub('\n', '', secret) | |
params = urlencode(dict( | |
client_id=client, | |
response_type="code", | |
redirect_uri="urn:ietf:wg:oauth:2.0:oob", | |
scope="read" | |
)) | |
aurl = 'https://'+instance+'/oauth/authorize?'+params | |
print("Please authencate your account. You will opening authencation page.") | |
print("If you didn't open browser. Use this URL: \n{}".format(str(aurl))) | |
webbrowser.open(str(aurl)) | |
ctoken = input("Token: ") | |
res = requests.post('https://'+str(instance)+'/oauth/token', dict( | |
grant_type="authorization_code", | |
redirect_uri="urn:ietf:wg:oauth:2.0:oob", | |
client_id=client, | |
client_secret=secret, | |
code=str(ctoken) | |
)).json() | |
with open("userkey.txt", "w") as u: | |
u.write(str(res["access_token"])) | |
print("Created userkey.txt") | |
print("Setup Complete! You have to restart this script.") | |
input("Please Enter to finish setup...") | |
sys.exit() | |
# Read instance address | |
with open("instance.txt") as a: | |
address = a.readlines() | |
# init Mastodon.py | |
try: | |
print("Initializing") | |
mastodon = Mastodon( | |
access_token = "userkey.txt", | |
api_base_url = str(address[0]) | |
) | |
# include args | |
args = sys.argv | |
# Start Listen | |
listen = SListen() | |
notifylisten = NotifySListen() | |
if "home" in args or "notify" in args or "local" in args or "public" in args: | |
if "home" in args: | |
print("Streaming Start: Home") | |
mastodon.stream_user(listen) | |
elif "notify" in args: | |
print("Streaming Start: Notify") | |
mastodon.stream_user(notifylisten) | |
elif "local" in args: | |
print("Streaming Start: Local") | |
mastodon.stream_local(listen) | |
elif "public" in args: | |
print("Streaming Start: Public") | |
mastodon.stream_public(listen) | |
else: | |
print("Selecting Home timeline.") | |
print("Streaming Start: Home") | |
mastodon.stream_user(listen) | |
except KeyboardInterrupt: | |
print("Connection Closing") | |
quit() | |
except: | |
print("FATAL: " + traceback.format_exc()) | |
quit() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
changelog
2018/02/11
2018/02/12
2018/04/21
2018/09/30