diff options
| -rw-r--r-- | .gitignore | 2 | ||||
| -rw-r--r-- | LICENCE.txt | 13 | ||||
| -rw-r--r-- | README.md | 111 | ||||
| -rw-r--r-- | __init__.py | 3 | ||||
| -rwxr-xr-x | plann.py | 190 | ||||
| -rw-r--r-- | poll.py | 71 | ||||
| -rw-r--r-- | requirements.txt | 2 | ||||
| -rw-r--r-- | status.py | 116 |
8 files changed, 508 insertions, 0 deletions
diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d50a09f --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +.env +__pycache__/ diff --git a/LICENCE.txt b/LICENCE.txt new file mode 100644 index 0000000..edec74b --- /dev/null +++ b/LICENCE.txt @@ -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. diff --git a/README.md b/README.md new file mode 100644 index 0000000..cbb1863 --- /dev/null +++ b/README.md @@ -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. diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..f972568 --- /dev/null +++ b/__init__.py @@ -0,0 +1,3 @@ +"""pleroma_announce""" + +from .plann import * 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() @@ -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})" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..4564349 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +dotenv +requests diff --git a/status.py b/status.py new file mode 100644 index 0000000..55b20d1 --- /dev/null +++ b/status.py @@ -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)})" |