Skip to content

Instantly share code, notes, and snippets.

@YuzuRyo61
Last active September 30, 2018 12:59
Show Gist options
  • Save YuzuRyo61/65a2f06e12b1118c76bbf1989cea2810 to your computer and use it in GitHub Desktop.
Save YuzuRyo61/65a2f06e12b1118c76bbf1989cea2810 to your computer and use it in GitHub Desktop.
端末で見れるMastodonStreaming
#!/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("&apos;", "'")
text = text.replace("&quot;", "\"")
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()
@YuzuRyo61
Copy link
Author

YuzuRyo61 commented Feb 11, 2018

changelog

2018/02/11

  • < や > 等の記号がエスケープ文字になっていた問題を解決させるためのスクリプトを追加。
  • 提案により、instance.txtを読み取るための記述をwith型に変更。

2018/02/12

  • 新しく新ファイル追加。内容についてはそこに記載済みです。

2018/04/21

  • 新ファイル追加。セットアップスクリプトを追加して、単体で動作することが可能になりました。(ファイル名 : STANDALONE.py)

2018/09/30

  • 他ファイル削除。
  • クライアントキーは認証時のみ必要なため、通常初期化時にクライアントキーの読み込みをしないように変更
  • " ' "の変換が &apos; のままになるため、replaceで訂正。
  • " の変換が &quot; のままになるため、replaceで訂正。
  • 文字変換をクラス扱いではなく関数に変更。
  • ホームタイムラインと通知タイムラインを分離。通知タイムラインは別途別クラスを使用。

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment