initial commitHEADv0.1main

1 files changed, 190 insertions, 0 deletions

diff --git a/plann.py b/plann.py
new file mode 100755
index 0000000..1ce87ac
--- /dev/null
+++ b/plann.py
@@ -0,0 +1,190 @@
+#!/usr/bin/env python3
+
+"""small tool to post, schedule and cancel announcements on pleroma"""
+
+import argparse
+import datetime
+import json
+import logging
+import os
+
+import dotenv
+import requests
+
+from poll import PleromaPoll
+from status import PleromaScheduledStatus
+
+__all__ = ["publish_status", "list_scheduled_statuses", "cancel_scheduled_status"]
+
+dotenv.load_dotenv()
+APP_TOKEN=os.getenv("APP_TOKEN")
+HOST=os.getenv("HOST")
+
+
+def publish_status(text: str,
+        visibility: str = "unlisted",
+        scheduled_at: datetime.datetime = None,
+        poll: PleromaPoll = None):
+    """publish or schedule a status"""
+
+    url = f"{HOST}/api/v1/statuses"
+    headers = {"Authorization": f"Bearer {APP_TOKEN}"}
+
+    payload = {
+            "status": text,
+            "visibilty": visibility
+            }
+
+    if scheduled_at is not None:
+        payload["scheduled_at"] = scheduled_at.isoformat()
+    elif poll is not None:
+        payload["poll"] = {
+                "options": poll.options,
+                "expires_in": poll.expires_in,
+                "multiple": poll.multiple,
+                "hide_totals": poll.hide_totals
+                }
+    req = requests.post(url, headers=headers, json=payload)
+    if req.ok:
+        logging.debug("publish_status: %s", req.text)
+        logging.info("publish_status: published status")
+    else:
+        logging.error("publish_status: %s", req.text)
+
+
+def publish_status_helper(args: dict):
+    """publish_status wrapper to use with argparse"""
+
+    if args.poll_option is not None:
+        poll = PleromaPoll(options=args.poll_option,
+                expires_in=datetime.timedelta(days=args.expires_in),
+                multiple=args.multiple,
+                hide_totals=args.hide_totals)
+        publish_status(args.text, visibility=args.visibility,
+                scheduled_at=args.datetime, poll=poll)
+    else:
+        publish_status(args.text, visibility=args.visibility,
+                scheduled_at=args.datetime)
+
+
+def list_scheduled_statuses() -> list[PleromaScheduledStatus]:
+    """get a list of scheduled statuses"""
+    url = f"{HOST}/api/v1/scheduled_statuses"
+    headers = {"Authorization": f"Bearer {APP_TOKEN}"}
+
+    out = []
+    req = requests.get(url, headers=headers)
+    if req.ok:
+        logging.debug("list_scheduled_statuses: %s", req.text)
+        logging.info("got a list of scheduled statuses")
+        statuses_raw = json.loads(req.text)
+        for s in statuses_raw:
+            out.append(PleromaScheduledStatus.from_response(s))
+    else:
+        logging.error("list_scheduled_statuses: %s", req.text)
+
+    return out
+
+
+def list_scheduled_statuses_helper(args: dict):
+    """list_scheduled_statuses wrapper to use with argparse"""
+
+    for status in list_scheduled_statuses():
+        print(status)
+
+
+def cancel_scheduled_status(s_id: str):
+    """cancel a scheduled status with id"""
+    url = f"{HOST}/api/v1/scheduled_statuses/{s_id}"
+    headers = {"Authorization": f"Bearer {APP_TOKEN}"}
+
+    req = requests.delete(url, headers=headers)
+    if req.ok:
+        logging.debug("cancel_scheduled_status: %s", req.text)
+        logging.info("cancel_scheduled_status: cancelled status %s", s_id)
+    else:
+        logging.error("cancel_scheduled_status: %s", req.text)
+
+
+def cancel_scheduled_status_helper(args: dict):
+    """cancel_scheduled_status wrapper to use with argparse"""
+    prompt = "are you sure you want to cancel ALL these statuses? [Yy] "
+
+    statuses = list_scheduled_statuses()
+
+    if args.id is not None:
+        if args.id in [s.s_id for s in statuses]:
+            cancel_scheduled_status(args.id)
+        else:
+            logging.error("no such status")
+            return
+    elif args.all:
+        for s in statuses:
+            print(s)
+        if not args.yes:
+            try:
+                assert input(prompt).lower() == 'y'
+            except AssertionError:
+                return
+
+        for s in statuses:
+            cancel_scheduled_status(s.s_id)
+
+
+def main():
+    """entry point for interactive use"""
+
+    parser = argparse.ArgumentParser(prog="plann")
+    parser.add_argument("-v", "--verbose", action="count", default=0,
+            help="increase verbosity")
+    subparsers = parser.add_subparsers(help="subcommands (see individual ones for help)")
+
+    publish_parser = subparsers.add_parser("post",
+            description="post or schedule a status")
+    publish_parser.add_argument("text", help="text of a status")
+    publish_parser.add_argument("-d", "--datetime", type=datetime.datetime.fromisoformat,
+            help="date and time to which status will be scheduled")
+    publish_parser.add_argument("-s", "--visibility", type=str,
+            choices=["public", "unlisted", "private", "direct"], default="unlisted",
+            help="status visibility (default: unlisted)")
+    publish_poll_group = publish_parser.add_argument_group("poll")
+    publish_poll_group.add_argument("-p", "--poll-option", type=str, metavar="OPT",
+            action="extend", nargs="+",
+            help="add a poll option")
+    publish_poll_group.add_argument("-e", "--expires-in", type=int, metavar="DAYS",
+            default=1,
+            help="number of days in which poll will expire (default: 1 day)")
+    publish_poll_group.add_argument("-m", "--multiple", action="store_true",
+            help="allow multiple choices")
+    publish_poll_group.add_argument("-t", "--hide_totals", action="store_true",
+            help="hide totals until poll is expired")
+    publish_parser.set_defaults(func=publish_status_helper)
+
+    cancel_parser = subparsers.add_parser("cancel",
+            description="cancel one or all scheduled statuses")
+    cancel_parser.add_argument("-y", "--yes", action="store_true",
+            help="don't ask for confirmation (only for -a option)")
+    cancel_group = cancel_parser.add_mutually_exclusive_group()
+    cancel_group.add_argument("-i", "--id", type=str,
+            help="scheduled post id to cancel")
+    cancel_group.add_argument("-a", "--all", action="store_true",
+            help="cancel all scheduled posts")
+    cancel_parser.set_defaults(func=cancel_scheduled_status_helper)
+
+    list_parser = subparsers.add_parser("list",
+            description="list scheduled statuses")
+    list_parser.set_defaults(func=list_scheduled_statuses_helper)
+
+    args = parser.parse_args()
+
+    loglevels = (logging.ERROR, logging.WARNING, logging.INFO, logging.DEBUG)
+    logging.basicConfig(level=loglevels[min(args.verbose, len(loglevels))])
+
+    try:
+        args.func(args)
+    except AttributeError:
+        logging.error("please specify a subcommand (one of {post, list, cancel})")
+
+
+if __name__ == '__main__':
+    main()