1#!/usr/bin/env python3
  2
  3"""small tool to post, schedule and cancel announcements on pleroma"""
  4
  5import argparse
  6import datetime
  7import json
  8import logging
  9import os
 10
 11import dotenv
 12import requests
 13
 14from poll import PleromaPoll
 15from status import PleromaScheduledStatus
 16
 17__all__ = ["publish_status", "list_scheduled_statuses", "cancel_scheduled_status"]
 18
 19dotenv.load_dotenv()
 20APP_TOKEN=os.getenv("APP_TOKEN")
 21HOST=os.getenv("HOST")
 22
 23
 24def publish_status(text: str,
 25        visibility: str = "unlisted",
 26        scheduled_at: datetime.datetime = None,
 27        poll: PleromaPoll = None):
 28    """publish or schedule a status"""
 29
 30    url = f"{HOST}/api/v1/statuses"
 31    headers = {"Authorization": f"Bearer {APP_TOKEN}"}
 32
 33    payload = {
 34            "status": text,
 35            "visibilty": visibility
 36            }
 37
 38    if scheduled_at is not None:
 39        payload["scheduled_at"] = scheduled_at.isoformat()
 40    elif poll is not None:
 41        payload["poll"] = {
 42                "options": poll.options,
 43                "expires_in": poll.expires_in,
 44                "multiple": poll.multiple,
 45                "hide_totals": poll.hide_totals
 46                }
 47    req = requests.post(url, headers=headers, json=payload)
 48    if req.ok:
 49        logging.debug("publish_status: %s", req.text)
 50        logging.info("publish_status: published status")
 51    else:
 52        logging.error("publish_status: %s", req.text)
 53
 54
 55def publish_status_helper(args: dict):
 56    """publish_status wrapper to use with argparse"""
 57
 58    if args.poll_option is not None:
 59        poll = PleromaPoll(options=args.poll_option,
 60                expires_in=datetime.timedelta(days=args.expires_in),
 61                multiple=args.multiple,
 62                hide_totals=args.hide_totals)
 63        publish_status(args.text, visibility=args.visibility,
 64                scheduled_at=args.datetime, poll=poll)
 65    else:
 66        publish_status(args.text, visibility=args.visibility,
 67                scheduled_at=args.datetime)
 68
 69
 70def list_scheduled_statuses() -> list[PleromaScheduledStatus]:
 71    """get a list of scheduled statuses"""
 72    url = f"{HOST}/api/v1/scheduled_statuses"
 73    headers = {"Authorization": f"Bearer {APP_TOKEN}"}
 74
 75    out = []
 76    req = requests.get(url, headers=headers)
 77    if req.ok:
 78        logging.debug("list_scheduled_statuses: %s", req.text)
 79        logging.info("got a list of scheduled statuses")
 80        statuses_raw = json.loads(req.text)
 81        for s in statuses_raw:
 82            out.append(PleromaScheduledStatus.from_response(s))
 83    else:
 84        logging.error("list_scheduled_statuses: %s", req.text)
 85
 86    return out
 87
 88
 89def list_scheduled_statuses_helper(args: dict):
 90    """list_scheduled_statuses wrapper to use with argparse"""
 91
 92    for status in list_scheduled_statuses():
 93        print(status)
 94
 95
 96def cancel_scheduled_status(s_id: str):
 97    """cancel a scheduled status with id"""
 98    url = f"{HOST}/api/v1/scheduled_statuses/{s_id}"
 99    headers = {"Authorization": f"Bearer {APP_TOKEN}"}
100
101    req = requests.delete(url, headers=headers)
102    if req.ok:
103        logging.debug("cancel_scheduled_status: %s", req.text)
104        logging.info("cancel_scheduled_status: cancelled status %s", s_id)
105    else:
106        logging.error("cancel_scheduled_status: %s", req.text)
107
108
109def cancel_scheduled_status_helper(args: dict):
110    """cancel_scheduled_status wrapper to use with argparse"""
111    prompt = "are you sure you want to cancel ALL these statuses? [Yy] "
112
113    statuses = list_scheduled_statuses()
114
115    if args.id is not None:
116        if args.id in [s.s_id for s in statuses]:
117            cancel_scheduled_status(args.id)
118        else:
119            logging.error("no such status")
120            return
121    elif args.all:
122        for s in statuses:
123            print(s)
124        if not args.yes:
125            try:
126                assert input(prompt).lower() == 'y'
127            except AssertionError:
128                return
129
130        for s in statuses:
131            cancel_scheduled_status(s.s_id)
132
133
134def main():
135    """entry point for interactive use"""
136
137    parser = argparse.ArgumentParser(prog="plann")
138    parser.add_argument("-v", "--verbose", action="count", default=0,
139            help="increase verbosity")
140    subparsers = parser.add_subparsers(help="subcommands (see individual ones for help)")
141
142    publish_parser = subparsers.add_parser("post",
143            description="post or schedule a status")
144    publish_parser.add_argument("text", help="text of a status")
145    publish_parser.add_argument("-d", "--datetime", type=datetime.datetime.fromisoformat,
146            help="date and time to which status will be scheduled")
147    publish_parser.add_argument("-s", "--visibility", type=str,
148            choices=["public", "unlisted", "private", "direct"], default="unlisted",
149            help="status visibility (default: unlisted)")
150    publish_poll_group = publish_parser.add_argument_group("poll")
151    publish_poll_group.add_argument("-p", "--poll-option", type=str, metavar="OPT",
152            action="extend", nargs="+",
153            help="add a poll option")
154    publish_poll_group.add_argument("-e", "--expires-in", type=int, metavar="DAYS",
155            default=1,
156            help="number of days in which poll will expire (default: 1 day)")
157    publish_poll_group.add_argument("-m", "--multiple", action="store_true",
158            help="allow multiple choices")
159    publish_poll_group.add_argument("-t", "--hide_totals", action="store_true",
160            help="hide totals until poll is expired")
161    publish_parser.set_defaults(func=publish_status_helper)
162
163    cancel_parser = subparsers.add_parser("cancel",
164            description="cancel one or all scheduled statuses")
165    cancel_parser.add_argument("-y", "--yes", action="store_true",
166            help="don't ask for confirmation (only for -a option)")
167    cancel_group = cancel_parser.add_mutually_exclusive_group()
168    cancel_group.add_argument("-i", "--id", type=str,
169            help="scheduled post id to cancel")
170    cancel_group.add_argument("-a", "--all", action="store_true",
171            help="cancel all scheduled posts")
172    cancel_parser.set_defaults(func=cancel_scheduled_status_helper)
173
174    list_parser = subparsers.add_parser("list",
175            description="list scheduled statuses")
176    list_parser.set_defaults(func=list_scheduled_statuses_helper)
177
178    args = parser.parse_args()
179
180    loglevels = (logging.ERROR, logging.WARNING, logging.INFO, logging.DEBUG)
181    logging.basicConfig(level=loglevels[min(args.verbose, len(loglevels))])
182
183    try:
184        args.func(args)
185    except AttributeError:
186        logging.error("please specify a subcommand (one of {post, list, cancel})")
187
188
189if __name__ == '__main__':
190    main()