kopia lustrzana https://github.com/ihabunek/toot
665 wiersze
20 KiB
Python
665 wiersze
20 KiB
Python
from itertools import chain
|
|
import json
|
|
import sys
|
|
import platform
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from time import sleep, time
|
|
|
|
from toot import api, config, __version__
|
|
from toot.auth import login_interactive, login_browser_interactive, create_app_interactive
|
|
from toot.entities import Account, Instance, Notification, Status, from_dict
|
|
from toot.exceptions import ApiError, ConsoleError
|
|
from toot.output import (print_lists, print_out, print_instance, print_account, print_acct_list,
|
|
print_search_results, print_status, print_table, print_timeline, print_notifications,
|
|
print_tag_list, print_list_accounts, print_user_list)
|
|
from toot.utils import args_get_instance, delete_tmp_status_file, editor_input, multiline_input, EOF_KEY
|
|
from toot.utils.datetime import parse_datetime
|
|
|
|
|
|
def get_timeline_generator(app, user, args):
|
|
if len([arg for arg in [args.tag, args.list, args.public, args.account] if arg]) > 1:
|
|
raise ConsoleError("Only one of --public, --tag, --account, or --list can be used at one time.")
|
|
|
|
if args.local and not (args.public or args.tag):
|
|
raise ConsoleError("The --local option is only valid alongside --public or --tag.")
|
|
|
|
if args.instance and not (args.public or args.tag):
|
|
raise ConsoleError("The --instance option is only valid alongside --public or --tag.")
|
|
|
|
if args.public:
|
|
if args.instance:
|
|
return api.anon_public_timeline_generator(args.instance, local=args.local, limit=args.count)
|
|
else:
|
|
return api.public_timeline_generator(app, user, local=args.local, limit=args.count)
|
|
elif args.tag:
|
|
if args.instance:
|
|
return api.anon_tag_timeline_generator(args.instance, args.tag, limit=args.count)
|
|
else:
|
|
return api.tag_timeline_generator(app, user, args.tag, local=args.local, limit=args.count)
|
|
elif args.account:
|
|
return api.account_timeline_generator(app, user, args.account, limit=args.count)
|
|
elif args.list:
|
|
return api.timeline_list_generator(app, user, args.list, limit=args.count)
|
|
else:
|
|
return api.home_timeline_generator(app, user, limit=args.count)
|
|
|
|
|
|
def timeline(app, user, args, generator=None):
|
|
if not generator:
|
|
generator = get_timeline_generator(app, user, args)
|
|
|
|
while True:
|
|
try:
|
|
items = next(generator)
|
|
except StopIteration:
|
|
print_out("That's all folks.")
|
|
return
|
|
|
|
if args.reverse:
|
|
items = reversed(items)
|
|
|
|
statuses = [from_dict(Status, item) for item in items]
|
|
print_timeline(statuses)
|
|
|
|
if args.once or not sys.stdout.isatty():
|
|
break
|
|
|
|
char = input("\nContinue? [Y/n] ")
|
|
if char.lower() == "n":
|
|
break
|
|
|
|
|
|
def status(app, user, args):
|
|
response = api.fetch_status(app, user, args.status_id)
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
status = from_dict(Status, response.json())
|
|
print_status(status)
|
|
|
|
|
|
def thread(app, user, args):
|
|
context_response = api.context(app, user, args.status_id)
|
|
|
|
if args.json:
|
|
print(context_response.text)
|
|
else:
|
|
toot = api.fetch_status(app, user, args.status_id).json()
|
|
context = context_response.json()
|
|
|
|
statuses = chain(context["ancestors"], [toot], context["descendants"])
|
|
print_timeline(from_dict(Status, s) for s in statuses)
|
|
|
|
|
|
def post(app, user, args):
|
|
if args.editor and not sys.stdin.isatty():
|
|
raise ConsoleError("Cannot run editor if not in tty.")
|
|
|
|
if args.media and len(args.media) > 4:
|
|
raise ConsoleError("Cannot attach more than 4 files.")
|
|
|
|
media_ids = _upload_media(app, user, args)
|
|
status_text = _get_status_text(args.text, args.editor, args.media)
|
|
scheduled_at = _get_scheduled_at(args.scheduled_at, args.scheduled_in)
|
|
|
|
if not status_text and not media_ids:
|
|
raise ConsoleError("You must specify either text or media to post.")
|
|
|
|
response = api.post_status(
|
|
app, user, status_text,
|
|
visibility=args.visibility,
|
|
media_ids=media_ids,
|
|
sensitive=args.sensitive,
|
|
spoiler_text=args.spoiler_text,
|
|
in_reply_to_id=args.reply_to,
|
|
language=args.language,
|
|
scheduled_at=scheduled_at,
|
|
content_type=args.content_type,
|
|
poll_options=args.poll_option,
|
|
poll_expires_in=args.poll_expires_in,
|
|
poll_multiple=args.poll_multiple,
|
|
poll_hide_totals=args.poll_hide_totals,
|
|
)
|
|
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
status = response.json()
|
|
if "scheduled_at" in status:
|
|
scheduled_at = parse_datetime(status["scheduled_at"])
|
|
scheduled_at = datetime.strftime(scheduled_at, "%Y-%m-%d %H:%M:%S%z")
|
|
print_out(f"Toot scheduled for: <green>{scheduled_at}</green>")
|
|
else:
|
|
print_out(f"Toot posted: <green>{status['url']}")
|
|
|
|
delete_tmp_status_file()
|
|
|
|
|
|
def _get_status_text(text, editor, media):
|
|
isatty = sys.stdin.isatty()
|
|
|
|
if not text and not isatty:
|
|
text = sys.stdin.read().rstrip()
|
|
|
|
if isatty:
|
|
if editor:
|
|
text = editor_input(editor, text)
|
|
elif not text and not media:
|
|
print_out("Write or paste your toot. Press <yellow>{}</yellow> to post it.".format(EOF_KEY))
|
|
text = multiline_input()
|
|
|
|
return text
|
|
|
|
|
|
def _get_scheduled_at(scheduled_at, scheduled_in):
|
|
if scheduled_at:
|
|
return scheduled_at
|
|
|
|
if scheduled_in:
|
|
scheduled_at = datetime.now(timezone.utc) + timedelta(seconds=scheduled_in)
|
|
return scheduled_at.replace(microsecond=0).isoformat()
|
|
|
|
return None
|
|
|
|
|
|
def _upload_media(app, user, args):
|
|
# Match media to corresponding description and thumbnail
|
|
media = args.media or []
|
|
descriptions = args.description or []
|
|
thumbnails = args.thumbnail or []
|
|
uploaded_media = []
|
|
|
|
for idx, file in enumerate(media):
|
|
description = descriptions[idx].strip() if idx < len(descriptions) else None
|
|
thumbnail = thumbnails[idx] if idx < len(thumbnails) else None
|
|
result = _do_upload(app, user, file, description, thumbnail)
|
|
uploaded_media.append(result)
|
|
|
|
_wait_until_all_processed(app, user, uploaded_media)
|
|
|
|
return [m["id"] for m in uploaded_media]
|
|
|
|
|
|
def _wait_until_all_processed(app, user, uploaded_media):
|
|
"""
|
|
Media is uploaded asynchronously, and cannot be attached until the server
|
|
has finished processing it. This function waits for that to happen.
|
|
|
|
Once media is processed, it will have the URL populated.
|
|
"""
|
|
if all(m["url"] for m in uploaded_media):
|
|
return
|
|
|
|
# Timeout after waiting 1 minute
|
|
start_time = time()
|
|
timeout = 60
|
|
|
|
print_out("<dim>Waiting for media to finish processing...</dim>")
|
|
for media in uploaded_media:
|
|
_wait_until_processed(app, user, media, start_time, timeout)
|
|
|
|
|
|
def _wait_until_processed(app, user, media, start_time, timeout):
|
|
if media["url"]:
|
|
return
|
|
|
|
media = api.get_media(app, user, media["id"])
|
|
while not media["url"]:
|
|
sleep(1)
|
|
if time() > start_time + timeout:
|
|
raise ConsoleError(f"Media not processed by server after {timeout} seconds. Aborting.")
|
|
media = api.get_media(app, user, media["id"])
|
|
|
|
|
|
def delete(app, user, args):
|
|
response = api.delete_status(app, user, args.status_id)
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out("<green>✓ Status deleted</green>")
|
|
|
|
|
|
def favourite(app, user, args):
|
|
response = api.favourite(app, user, args.status_id)
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out("<green>✓ Status favourited</green>")
|
|
|
|
|
|
def unfavourite(app, user, args):
|
|
response = api.unfavourite(app, user, args.status_id)
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out("<green>✓ Status unfavourited</green>")
|
|
|
|
|
|
def reblog(app, user, args):
|
|
response = api.reblog(app, user, args.status_id, visibility=args.visibility)
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out("<green>✓ Status reblogged</green>")
|
|
|
|
|
|
def unreblog(app, user, args):
|
|
response = api.unreblog(app, user, args.status_id)
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out("<green>✓ Status unreblogged</green>")
|
|
|
|
|
|
def pin(app, user, args):
|
|
response = api.pin(app, user, args.status_id)
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out("<green>✓ Status pinned</green>")
|
|
|
|
|
|
def unpin(app, user, args):
|
|
response = api.unpin(app, user, args.status_id)
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out("<green>✓ Status unpinned</green>")
|
|
|
|
|
|
def bookmark(app, user, args):
|
|
response = api.bookmark(app, user, args.status_id)
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out("<green>✓ Status bookmarked</green>")
|
|
|
|
|
|
def unbookmark(app, user, args):
|
|
response = api.unbookmark(app, user, args.status_id)
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out("<green>✓ Status unbookmarked</green>")
|
|
|
|
|
|
def bookmarks(app, user, args):
|
|
timeline(app, user, args, api.bookmark_timeline_generator(app, user, limit=args.count))
|
|
|
|
|
|
def reblogged_by(app, user, args):
|
|
response = api.reblogged_by(app, user, args.status_id)
|
|
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
headers = ["Account", "Display name"]
|
|
rows = [[a["acct"], a["display_name"]] for a in response.json()]
|
|
print_table(headers, rows)
|
|
|
|
|
|
def auth(app, user, args):
|
|
config_data = config.load_config()
|
|
|
|
if not config_data["users"]:
|
|
print_out("You are not logged in to any accounts")
|
|
return
|
|
|
|
active_user = config_data["active_user"]
|
|
|
|
print_out("Authenticated accounts:")
|
|
for uid, u in config_data["users"].items():
|
|
active_label = "ACTIVE" if active_user == uid else ""
|
|
print_out("* <green>{}</green> <yellow>{}</yellow>".format(uid, active_label))
|
|
|
|
path = config.get_config_file_path()
|
|
print_out("\nAuth tokens are stored in: <blue>{}</blue>".format(path))
|
|
|
|
|
|
def env(app, user, args):
|
|
print_out(f"toot {__version__}")
|
|
print_out(f"Python {sys.version}")
|
|
print_out(platform.platform())
|
|
|
|
|
|
def update_account(app, user, args):
|
|
options = [
|
|
args.avatar,
|
|
args.bot,
|
|
args.discoverable,
|
|
args.display_name,
|
|
args.header,
|
|
args.language,
|
|
args.locked,
|
|
args.note,
|
|
args.privacy,
|
|
args.sensitive,
|
|
]
|
|
|
|
if all(option is None for option in options):
|
|
raise ConsoleError("Please specify at least one option to update the account")
|
|
|
|
api.update_account(
|
|
app,
|
|
user,
|
|
avatar=args.avatar,
|
|
bot=args.bot,
|
|
discoverable=args.discoverable,
|
|
display_name=args.display_name,
|
|
header=args.header,
|
|
language=args.language,
|
|
locked=args.locked,
|
|
note=args.note,
|
|
privacy=args.privacy,
|
|
sensitive=args.sensitive,
|
|
)
|
|
|
|
print_out("<green>✓ Account updated</green>")
|
|
|
|
|
|
def login_cli(app, user, args):
|
|
base_url = args_get_instance(args.instance, args.scheme)
|
|
app = create_app_interactive(base_url)
|
|
login_interactive(app, args.email)
|
|
|
|
print_out()
|
|
print_out("<green>✓ Successfully logged in.</green>")
|
|
|
|
|
|
def login(app, user, args):
|
|
base_url = args_get_instance(args.instance, args.scheme)
|
|
app = create_app_interactive(base_url)
|
|
login_browser_interactive(app)
|
|
|
|
print_out()
|
|
print_out("<green>✓ Successfully logged in.</green>")
|
|
|
|
|
|
def logout(app, user, args):
|
|
user = config.load_user(args.account, throw=True)
|
|
config.delete_user(user)
|
|
print_out("<green>✓ User {} logged out</green>".format(config.user_id(user)))
|
|
|
|
|
|
def activate(app, user, args):
|
|
if not args.account:
|
|
print_out("Specify one of the following user accounts to activate:\n")
|
|
print_user_list(config.get_user_list())
|
|
return
|
|
|
|
user = config.load_user(args.account, throw=True)
|
|
config.activate_user(user)
|
|
print_out("<green>✓ User {} active</green>".format(config.user_id(user)))
|
|
|
|
|
|
def upload(app, user, args):
|
|
response = _do_upload(app, user, args.file, args.description, None)
|
|
|
|
msg = "Successfully uploaded media ID <yellow>{}</yellow>, type '<yellow>{}</yellow>'"
|
|
|
|
print_out()
|
|
print_out(msg.format(response['id'], response['type']))
|
|
print_out("URL: <green>{}</green>".format(response['url']))
|
|
print_out("Preview URL: <green>{}</green>".format(response['preview_url']))
|
|
|
|
|
|
def search(app, user, args):
|
|
response = api.search(app, user, args.query, args.resolve)
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_search_results(response.json())
|
|
|
|
|
|
def _do_upload(app, user, file, description, thumbnail):
|
|
print_out("Uploading media: <green>{}</green>".format(file.name))
|
|
return api.upload_media(app, user, file, description=description, thumbnail=thumbnail)
|
|
|
|
|
|
def follow(app, user, args):
|
|
account = api.find_account(app, user, args.account)
|
|
response = api.follow(app, user, account["id"])
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out(f"<green>✓ You are now following {args.account}</green>")
|
|
|
|
|
|
def unfollow(app, user, args):
|
|
account = api.find_account(app, user, args.account)
|
|
response = api.unfollow(app, user, account["id"])
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out(f"<green>✓ You are no longer following {args.account}</green>")
|
|
|
|
|
|
def following(app, user, args):
|
|
account = api.find_account(app, user, args.account)
|
|
accounts = api.following(app, user, account["id"])
|
|
if args.json:
|
|
print(json.dumps(accounts))
|
|
else:
|
|
print_acct_list(accounts)
|
|
|
|
|
|
def followers(app, user, args):
|
|
account = api.find_account(app, user, args.account)
|
|
accounts = api.followers(app, user, account["id"])
|
|
if args.json:
|
|
print(json.dumps(accounts))
|
|
else:
|
|
print_acct_list(accounts)
|
|
|
|
|
|
def tags_follow(app, user, args):
|
|
tn = args.tag_name if not args.tag_name.startswith("#") else args.tag_name[1:]
|
|
api.follow_tag(app, user, tn)
|
|
print_out("<green>✓ You are now following #{}</green>".format(tn))
|
|
|
|
|
|
def tags_unfollow(app, user, args):
|
|
tn = args.tag_name if not args.tag_name.startswith("#") else args.tag_name[1:]
|
|
api.unfollow_tag(app, user, tn)
|
|
print_out("<green>✓ You are no longer following #{}</green>".format(tn))
|
|
|
|
|
|
def tags_followed(app, user, args):
|
|
response = api.followed_tags(app, user)
|
|
print_tag_list(response)
|
|
|
|
|
|
def lists(app, user, args):
|
|
lists = api.get_lists(app, user)
|
|
|
|
if lists:
|
|
print_lists(lists)
|
|
else:
|
|
print_out("You have no lists defined.")
|
|
|
|
|
|
def list_accounts(app, user, args):
|
|
list_id = _get_list_id(app, user, args)
|
|
response = api.get_list_accounts(app, user, list_id)
|
|
print_list_accounts(response)
|
|
|
|
|
|
def list_create(app, user, args):
|
|
api.create_list(app, user, title=args.title, replies_policy=args.replies_policy)
|
|
print_out(f"<green>✓ List \"{args.title}\" created.</green>")
|
|
|
|
|
|
def list_delete(app, user, args):
|
|
list_id = _get_list_id(app, user, args)
|
|
api.delete_list(app, user, list_id)
|
|
print_out(f"<green>✓ List \"{args.title if args.title else args.id}\"</green> <red>deleted.</red>")
|
|
|
|
|
|
def list_add(app, user, args):
|
|
list_id = _get_list_id(app, user, args)
|
|
account = api.find_account(app, user, args.account)
|
|
|
|
try:
|
|
api.add_accounts_to_list(app, user, list_id, [account['id']])
|
|
except Exception as ex:
|
|
# if we failed to add the account, try to give a
|
|
# more specific error message than "record not found"
|
|
my_accounts = api.followers(app, user, account['id'])
|
|
found = False
|
|
if my_accounts:
|
|
for my_account in my_accounts:
|
|
if my_account['id'] == account['id']:
|
|
found = True
|
|
break
|
|
if found is False:
|
|
print_out(f"<red>You must follow @{account['acct']} before adding this account to a list.</red>")
|
|
else:
|
|
print_out(f"<red>{ex}</red>")
|
|
return
|
|
|
|
print_out(f"<green>✓ Added account \"{args.account}\"</green>")
|
|
|
|
|
|
def list_remove(app, user, args):
|
|
list_id = _get_list_id(app, user, args)
|
|
account = api.find_account(app, user, args.account)
|
|
api.remove_accounts_from_list(app, user, list_id, [account['id']])
|
|
print_out(f"<green>✓ Removed account \"{args.account}\"</green>")
|
|
|
|
|
|
def _get_list_id(app, user, args):
|
|
list_id = args.id or api.find_list_id(app, user, args.title)
|
|
if not list_id:
|
|
raise ConsoleError("List not found")
|
|
return list_id
|
|
|
|
|
|
def mute(app, user, args):
|
|
account = api.find_account(app, user, args.account)
|
|
response = api.mute(app, user, account['id'])
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out("<green>✓ You have muted {}</green>".format(args.account))
|
|
|
|
|
|
def unmute(app, user, args):
|
|
account = api.find_account(app, user, args.account)
|
|
response = api.unmute(app, user, account['id'])
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out("<green>✓ {} is no longer muted</green>".format(args.account))
|
|
|
|
|
|
def muted(app, user, args):
|
|
response = api.muted(app, user)
|
|
if args.json:
|
|
print(json.dumps(response))
|
|
else:
|
|
if len(response) > 0:
|
|
print("Muted accounts:")
|
|
print_acct_list(response)
|
|
else:
|
|
print("No accounts muted")
|
|
|
|
|
|
def block(app, user, args):
|
|
account = api.find_account(app, user, args.account)
|
|
response = api.block(app, user, account['id'])
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out("<green>✓ You are now blocking {}</green>".format(args.account))
|
|
|
|
|
|
def unblock(app, user, args):
|
|
account = api.find_account(app, user, args.account)
|
|
response = api.unblock(app, user, account['id'])
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
print_out("<green>✓ {} is no longer blocked</green>".format(args.account))
|
|
|
|
|
|
def blocked(app, user, args):
|
|
response = api.blocked(app, user)
|
|
if args.json:
|
|
print(json.dumps(response))
|
|
else:
|
|
if len(response) > 0:
|
|
print("Blocked accounts:")
|
|
print_acct_list(response)
|
|
else:
|
|
print("No accounts blocked")
|
|
|
|
|
|
def whoami(app, user, args):
|
|
response = api.verify_credentials(app, user)
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
account = from_dict(Account, response.json())
|
|
print_account(account)
|
|
|
|
|
|
def whois(app, user, args):
|
|
account = api.find_account(app, user, args.account)
|
|
# Here it's not possible to avoid parsing json since it's needed to find the account.
|
|
if args.json:
|
|
print(json.dumps(account))
|
|
else:
|
|
account = from_dict(Account, account)
|
|
print_account(account)
|
|
|
|
|
|
def instance(app, user, args):
|
|
default = app.base_url if app else None
|
|
base_url = args_get_instance(args.instance, args.scheme, default)
|
|
|
|
if not base_url:
|
|
raise ConsoleError("Please specify an instance.")
|
|
|
|
try:
|
|
response = api.get_instance(base_url)
|
|
except ApiError:
|
|
raise ConsoleError(
|
|
f"Instance not found at {base_url}.\n"
|
|
"The given domain probably does not host a Mastodon instance."
|
|
)
|
|
|
|
if args.json:
|
|
print(response.text)
|
|
else:
|
|
instance = from_dict(Instance, response.json())
|
|
print_instance(instance)
|
|
|
|
|
|
def notifications(app, user, args):
|
|
if args.clear:
|
|
api.clear_notifications(app, user)
|
|
print_out("<green>Cleared notifications</green>")
|
|
return
|
|
|
|
exclude = []
|
|
if args.mentions:
|
|
# Filter everything except mentions
|
|
# https://docs.joinmastodon.org/methods/notifications/
|
|
exclude = ["follow", "favourite", "reblog", "poll", "follow_request"]
|
|
notifications = api.get_notifications(app, user, exclude_types=exclude)
|
|
if not notifications:
|
|
print_out("<yellow>No notification</yellow>")
|
|
return
|
|
|
|
if args.reverse:
|
|
notifications = reversed(notifications)
|
|
|
|
notifications = [from_dict(Notification, n) for n in notifications]
|
|
print_notifications(notifications)
|
|
|
|
|
|
def tui(app, user, args):
|
|
from .tui.app import TUI
|
|
TUI.create(app, user, args).run()
|