@@ -0,0 +1,2 @@
+.env
+__pycache__/

@@ -0,0 +1,13 @@
+Copyright 2022 la-ninpre
+
+Permission to use, copy, modify, and/or distribute this software for any purpose
+with or without fee is hereby granted, provided that the above copyright notice
+and this permission notice appear in all copies.
+
+THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
+REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND
+FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
+INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
+OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+PERFORMANCE OF THIS SOFTWARE.

@@ -0,0 +1,111 @@
+# pleroma announce
+
+this is a small quick and dirty tool that intended to be used to post
+announcements and polls onto bot on pleroma.
+
+## prerequisites
+
+- python 3
+
+- pip
+
+- curl (optional, if you can replace it with other tool)
+
+- registered pleroma app and authentication token (see next section)
+
+## pleroma app registration
+
+1. create pleroma user and mark it as bot.
+
+2. create an application for pleroma to get `client_id` and `client_secret`.
+refer to [pleroma api specification][1] and [mastodon api specification][2].
+this tool requires `write:statuses` and `read:statuses` scopes, so don't forget
+to specify them, otherwise it will fail.
+
+  ```terminal
+  curl -X POST \
+      -F 'client_name={YOUR_APP_NAME}' \
+      -F 'redirect_uris=urn:ietf:wg:oauth:2.0:oob' \
+      -F 'scopes=write:statuses read:statuses' \
+      https://yourdomain.tld/api/v1/apps
+  ```
+  [1]: https://api.pleroma.social/#operation/AppController.create
+  [2]: https://docs.joinmastodon.org/client/authorized/
+
+3. with your `client_id` authorize your application with
+account you are planning to use for posting.
+
+  in browser enter:
+  ```
+  https://yourdomain.tld/oauth/authorize?client_id={YOUR_CLIENT_ID}
+  &scope=write:statuses+read:statuses
+  &redirect_uri=urn:ietf:wg:oauth:2.0:oob
+  &response_type=code
+  ```
+
+  after entering credentials, you will receive access code which
+  will be used in the next step.
+
+4. with obtained access code you have to aquire authentication token for
+the application.
+
+  ```terminal
+  curl -X POST \
+      -F 'client_id={YOUR_CLIENT_ID}' \
+      -F 'client_secret={YOUR_CLIENT_SECRET}' \
+      -F 'redirect_uri=urn:ietf:wg:oauth:2.0:oob' \
+      -F 'grant_type=authorization_code' \
+      -F 'code={YOUR_AUTHERIZATION_CODE}' \
+      -F 'scope=write:statuses read:statuses' \
+      https://yourdomain.tld/oauth/token
+  ```
+
+  you will json response with authentication token, now save it to `.env` file,
+  see `.env.example`.
+
+## usage
+
+1. git clone the repo
+
+  ```terminal
+  $ git clone https://git.aaoth.xyz/pleroma_announce.git
+  $ cd pleroma_announce
+  ```
+
+2. install required pip packages
+(though they are likely to be installed already)
+
+  ```terminal
+  $ pip install -r requirements.txt
+  ```
+
+3. now you can use the script, see `--help` for options. for example:
+
+  ```terminal
+  $ python plann.py post -d 2022-02-14T14:00+00:00 -p opt1 opt2 -e 3 -m test
+  ```
+
+  this will schedule a status with text `test` and poll with options `opt1` and
+  `opt2` to 14 feb 2022 14:00 UTC.
+
+## todos
+
+- add setup.py
+
+- figure out versions of pip packages
+
+- handle media attachments
+
+- add man page
+
+## contacts
+
+[email me][3] or reach me on [fedi][4] if you have found a bug or
+just something weird.
+
+[3]:mailto:aaoth@aaoth.xyz
+[4]:https://pleroma.aaoth.xyz/users/la_ninpre
+
+## licence
+
+this program is licensed under an isc licence, see `LICENCE.txt` for details.

@@ -0,0 +1,3 @@
+"""pleroma_announce"""
+
+from .plann import *

@@ -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()

@@ -0,0 +1,71 @@
+"""poll wrapper class"""
+
+import datetime
+
+class PleromaPoll:
+    """
+    wrapper class for handling polls
+    """
+    def __init__(self,
+            options: list[str],
+            expires_in: datetime.timedelta,
+            multiple: bool = None,
+            hide_totals: bool = None):
+        self._options = options
+        self._expires_in = expires_in.total_seconds()
+        self._multiple = multiple
+        self._hide_totals = hide_totals
+
+    @classmethod
+    def from_response(cls, data: dict):
+        """
+        alternative constructor to create object from response of a server
+
+        as the app is used to only view scheduled posts, this constructor doesn't
+        retain information about vote counts.
+        """
+        return cls(expires_in=datetime.datetime.fromisoformat(data["expires_in"]),
+                multiple=data["multiple"],
+                options=[i["title"] for i in data["options"]])
+
+    @property
+    def options(self):
+        """
+        a list of strings, which are options for poll
+        """
+        return self._options
+
+    @property
+    def expires_in(self):
+        """
+        timedelta, which represents time in which poll will expire
+        """
+        return self._expires_in
+
+    @property
+    def multiple(self):
+        """
+        boolean flag, True if poll allows multiple choices
+        """
+        return self._multiple
+
+    @multiple.setter
+    def multiple(self, flag: bool):
+        self._multiple = flag
+
+    @property
+    def hide_totals(self):
+        """
+        boolean flag, True if counts are hidden until poll ends
+        """
+        return self._hide_totals
+
+    @hide_totals.setter
+    def hide_totals(self, flag: bool):
+        self._hide_totals = flag
+
+    def __repr__(self):
+        return f"PleromaPoll(options={self._options},\
+expires_in={self._expires_in},\
+multiple={self._multiple},\
+hide_totals={self._hide_totals})"

@@ -0,0 +1,2 @@
+dotenv
+requests

@@ -0,0 +1,116 @@
+"""classes for status objects"""
+
+import datetime as dt
+
+from poll import PleromaPoll
+
+class PleromaScheduledStatus:
+    """
+    scheduled status type
+    """
+    def __init__(self,
+            s_id: str,
+            scheduled_at: dt.datetime,
+            params: dict,
+            media_attachments: list):
+        self._s_id = s_id
+        self._scheduled_at = scheduled_at
+        self._params = params
+        self._media_attachments = media_attachments
+
+    @classmethod
+    def from_response(cls, data: dict):
+        """
+        alternative constructor to create object from response of a server
+        """
+        try:
+            media_attachments = data["media_attachments"]
+        except KeyError:
+            media_attachments = None
+        return cls(s_id=data["id"],
+                scheduled_at=dt.datetime.fromisoformat(data["scheduled_at"].replace("Z","")),
+                params=PleromaScheduledStatusParams.from_response(data["params"]),
+                media_attachments=media_attachments)
+
+    @property
+    def s_id(self):
+        """status id"""
+        return self._s_id
+
+    @property
+    def scheduled_at(self):
+        """time to which post is scheduled"""
+        return self._scheduled_at
+
+    @property
+    def params(self):
+        """params object"""
+        return self.params
+
+    @property
+    def media_attachments(self):
+        """attachments"""
+        return self.media_attachments
+
+    def __repr__(self):
+        return f"PleromaScheduledStatus(s_id={self._s_id},\
+scheduled_at={repr(self._scheduled_at)},\
+params={repr(self._params)},\
+media_attachments={self._media_attachments})"
+
+    def __str__(self):
+        # TODO: add poll
+        return f"[#{self._s_id}] - {self._params.text}"
+
+
+
+class PleromaScheduledStatusParams:
+    """
+    wrapper class for params of scheduled status
+
+    note: only requided attributes implemented here, because i won't ever need
+    other things just for announcing. i don't want to implement full pleroma
+    api and spend here another eternity.
+    """
+    def __init__(self,
+            text: str,
+            visibility: str,
+            poll: PleromaPoll = None):
+        self._poll = poll
+        self._text = text
+        self._visibility = visibility
+
+    @classmethod
+    def from_response(cls, data: dict):
+        """
+        alternative constructor to make object from response of a server
+        """
+        if data["poll"] is not None:
+            poll = PleromaPoll.from_response(data["poll"])
+        else:
+            poll = None
+        return cls(text=data["text"],
+                visibility=data["visibility"],
+                poll=poll)
+
+    @property
+    def poll(self):
+        """poll object if any"""
+        return self._poll
+
+    @property
+    def text(self):
+        """text of a status"""
+        return self._text
+
+    @property
+    def visibility(self):
+        """
+        visibility of a status
+        """
+        return self._visibility
+
+    def __repr__(self):
+        return f"PleromaScheduledStatusParams(text={self._text},\
+visibility={self._visibility},\
+poll={repr(self._poll)})"