111 lines
3.1 KiB
Python
111 lines
3.1 KiB
Python
|
import pkg_resources
|
||
|
from urllib.parse import urljoin
|
||
|
|
||
|
import requests
|
||
|
from requests import Response
|
||
|
|
||
|
from mobilizon_reshare.event.event import MobilizonEvent
|
||
|
from mobilizon_reshare.publishers.abstract import (
|
||
|
AbstractPlatform,
|
||
|
AbstractEventFormatter,
|
||
|
)
|
||
|
from mobilizon_reshare.publishers.exceptions import (
|
||
|
InvalidBot,
|
||
|
InvalidEvent,
|
||
|
InvalidResponse,
|
||
|
PublisherError,
|
||
|
HTTPResponseError,
|
||
|
)
|
||
|
|
||
|
|
||
|
class MastodonFormatter(AbstractEventFormatter):
|
||
|
|
||
|
_conf = ("publisher", "mastodon")
|
||
|
default_template_path = pkg_resources.resource_filename(
|
||
|
"mobilizon_reshare.publishers.templates", "mastodon.tmpl.j2"
|
||
|
)
|
||
|
|
||
|
default_recap_template_path = pkg_resources.resource_filename(
|
||
|
"mobilizon_reshare.publishers.templates", "mastodon_recap.tmpl.j2"
|
||
|
)
|
||
|
|
||
|
default_recap_header_template_path = pkg_resources.resource_filename(
|
||
|
"mobilizon_reshare.publishers.templates", "mastodon_recap_header.tmpl.j2"
|
||
|
)
|
||
|
|
||
|
def validate_event(self, event: MobilizonEvent) -> None:
|
||
|
text = event.description
|
||
|
if not (text and text.strip()):
|
||
|
self._log_error("No text was found", raise_error=InvalidEvent)
|
||
|
|
||
|
def validate_message(self, message) -> None:
|
||
|
if len(message.encode("utf-8")) >= self.conf.toot_length:
|
||
|
raise PublisherError("Message is too long")
|
||
|
|
||
|
|
||
|
class MastodonPlatform(AbstractPlatform):
|
||
|
"""
|
||
|
Mastodon publisher class.
|
||
|
"""
|
||
|
|
||
|
_conf = ("publisher", "mastodon")
|
||
|
api_uri = "api/v1/"
|
||
|
|
||
|
def _send(self, message: str) -> Response:
|
||
|
"""
|
||
|
Send messages
|
||
|
"""
|
||
|
return requests.post(
|
||
|
url=urljoin(self.conf.instance, self.api_uri) + "statuses",
|
||
|
headers={"Authorization": f"Bearer {self.conf.token}"},
|
||
|
data={
|
||
|
"status": message,
|
||
|
"visibility": "public",
|
||
|
},
|
||
|
)
|
||
|
|
||
|
def validate_credentials(self):
|
||
|
res = requests.get(
|
||
|
headers={"Authorization": f"Bearer {self.conf.token}"},
|
||
|
url=urljoin(self.conf.instance, self.api_uri) + "apps/verify_credentials",
|
||
|
)
|
||
|
data = self._validate_response(res)
|
||
|
|
||
|
if not self.conf.name == data["name"]:
|
||
|
self._log_error(
|
||
|
"Found a different bot than the expected one"
|
||
|
f"\n\tfound: {data['name']}"
|
||
|
f"\n\texpected: {self.conf.name}",
|
||
|
raise_error=InvalidBot,
|
||
|
)
|
||
|
|
||
|
def _validate_response(self, res: Response) -> dict:
|
||
|
try:
|
||
|
res.raise_for_status()
|
||
|
except requests.exceptions.HTTPError as e:
|
||
|
self._log_debug(str(res))
|
||
|
self._log_error(
|
||
|
str(e),
|
||
|
raise_error=HTTPResponseError,
|
||
|
)
|
||
|
|
||
|
try:
|
||
|
data = res.json()
|
||
|
except Exception as e:
|
||
|
self._log_error(
|
||
|
f"Server returned invalid json data: {str(e)}",
|
||
|
raise_error=InvalidResponse,
|
||
|
)
|
||
|
|
||
|
return data
|
||
|
|
||
|
|
||
|
class MastodonPublisher(MastodonPlatform):
|
||
|
|
||
|
_conf = ("publisher", "mastodon")
|
||
|
|
||
|
|
||
|
class MastodonNotifier(MastodonPlatform):
|
||
|
|
||
|
_conf = ("notifier", "mastodon")
|