mirror of
https://gitlab.com/octospacc/WinDog.git
synced 2025-06-05 22:09:20 +02:00
Start Matrix support, more refactoring, cleaner /help and /translate
This commit is contained in:
@@ -1,36 +1,45 @@
|
||||
# ================================== #
|
||||
# WinDog multi-purpose chatbot #
|
||||
# Licensed under AGPLv3 by OctoSpacc #
|
||||
# ================================== #
|
||||
|
||||
MastodonUrl, MastodonToken = None, None
|
||||
|
||||
import mastodon
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
def MastodonSender(event, manager, Data, Destination, TextPlain, TextMarkdown) -> None:
|
||||
if InDict(Data, 'Media'):
|
||||
Media = manager.media_post(Data['Media'], Magic(mime=True).from_buffer(Data['Media']))
|
||||
while Media['url'] == 'null':
|
||||
Media = manager.media(Media)
|
||||
if TextPlain:
|
||||
manager.status_post(
|
||||
status=(TextPlain + '\n\n@' + event['account']['acct']),
|
||||
media_ids=(Media if InDict(Data, 'Media') else None),
|
||||
in_reply_to_id=event['status']['id'],
|
||||
visibility=('direct' if event['status']['visibility'] == 'direct' else 'unlisted'),
|
||||
)
|
||||
|
||||
def MastodonMain() -> None:
|
||||
def MastodonMain() -> bool:
|
||||
if not (MastodonUrl and MastodonToken):
|
||||
return
|
||||
return False
|
||||
Mastodon = mastodon.Mastodon(api_base_url=MastodonUrl, access_token=MastodonToken)
|
||||
class MastodonListener(mastodon.StreamListener):
|
||||
def on_notification(self, event):
|
||||
if event['type'] == 'mention':
|
||||
Msg = BeautifulSoup(event['status']['content'], 'html.parser').get_text(' ').strip().replace('\t', ' ')
|
||||
if not Msg.split('@')[0]:
|
||||
Msg = ' '.join('@'.join(Msg.split('@')[1:]).strip().split(' ')[1:]).strip()
|
||||
if Msg[0] in CmdPrefixes:
|
||||
cmd = ParseCmd(Msg)
|
||||
if cmd:
|
||||
cmd.messageId = event['status']['id']
|
||||
if cmd.Name in Endpoints:
|
||||
Endpoints[cmd.Name]({"Event": event, "Manager": Mastodon}, cmd)
|
||||
OnMessageReceived()
|
||||
message = BeautifulSoup(event['status']['content'], 'html.parser').get_text(' ').strip().replace('\t', ' ')
|
||||
if not message.split('@')[0]:
|
||||
message = ' '.join('@'.join(message.split('@')[1:]).strip().split(' ')[1:]).strip()
|
||||
if message[0] in CmdPrefixes:
|
||||
command = ParseCmd(message)
|
||||
if command:
|
||||
command.messageId = event['status']['id']
|
||||
if command.Name in Endpoints:
|
||||
Endpoints[command.Name]({"Event": event, "Manager": Mastodon}, command)
|
||||
Mastodon.stream_user(MastodonListener(), run_async=True)
|
||||
return True
|
||||
|
||||
def MastodonSender(event, manager, data, destination, textPlain, textMarkdown) -> None:
|
||||
if InDict(data, 'Media'):
|
||||
Media = manager.media_post(data['Media'], Magic(mime=True).from_buffer(data['Media']))
|
||||
while Media['url'] == 'null':
|
||||
Media = manager.media(Media)
|
||||
if textPlain or Media:
|
||||
manager.status_post(
|
||||
status=(textPlain + '\n\n@' + event['account']['acct']),
|
||||
media_ids=(Media if InDict(data, 'Media') else None),
|
||||
in_reply_to_id=event['status']['id'],
|
||||
visibility=('direct' if event['status']['visibility'] == 'direct' else 'unlisted'),
|
||||
)
|
||||
|
||||
RegisterPlatform(name="Mastodon", main=MastodonMain, sender=MastodonSender, managerClass=mastodon.Mastodon)
|
||||
|
||||
|
Reference in New Issue
Block a user