mathdatech/exemple_mastodon.py
1
0
Fork 0
Code Projects Releases Activity

add 1 another directory

Browse Source
This commit is contained in:
Math DaTech 2022-05-23 20:47:07 +02:00
parent 5fc3884101
commit 6d029d1d01
4 changed files with 926 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
delibird-master/LICENSE Executable file
View File

@ -0,0 +1,26 @@
Copyright (c) The Regents of the University of California.
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:
1. Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.
3. Neither the name of the University nor the names of its contributors
may be used to endorse or promote products derived from this software
without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
SUCH DAMAGE.

21
delibird-master/README.md Executable file
View File

@ -0,0 +1,21 @@
# Delibird / Birb-bot
Delibird / Birb-bot is a Mastodon bot flying from user to user!
It can be sent by an user to another by mentioning the bird with a message like
“Go see user@domain” or with similar keywords. It can only be at one place at a
time, though, meaning that when it is flying away to someone or it is visiting
someone else than you, you cannot send it flying to another user!
## Dependencies
- Python 3
- Mastodon.py: https://github.com/halcy/Mastodon.py/
## Writing your own birbs
To avoid spoilers around the “rewards” posted by my own birb-bot, the repo does
not include the full “data.py” I use.
Instead, you can find a “data.py.example” which is pretty close but omits some
rewards and media definitions.

394
delibird-master/data.py.example Executable file
View File

@ -0,0 +1,394 @@
MEDIA = {
'FLYING_AWAY': {
'file': 'bye.mp4',
'source': 'https://www.youtube.com/watch?v=6Qp4wafJ8_I',
},
'FLYING_IN': {
'file': 'hello.mp4',
'source': 'https://www.youtube.com/watch?v=Gu8EudiPLZw',
},
'STAIRS': {
'file': 'stairs.mp4',
'source': 'https://www.youtube.com/watch?v=coDxZZjTt4w',
},
'BELLY_RUB': {
'file': 'rewards/belly-rub.mp4',
'source': 'https://www.youtube.com/watch?v=-bTaj37E79k',
},
'DJBIRB': {
'file': 'rewards/dj.mp4',
'source': 'https://www.youtube.com/watch?v=1v2SN06DnPo',
},
'SHOPPING': {
'file': 'rewards/shopping.mp4',
'source': 'https://www.youtube.com/watch?v=BfhlPEkNm7o',
},
'DANCINGQUEEN': {
'file': 'rewards/queen.mp4',
'source': 'https://www.youtube.com/watch?v=9DaGgzJm6is',
},
'MICROWAVE': {
'file': 'rewards/microwave.mp4',
'source': 'https://www.youtube.com/watch?v=NVY0qNe3BUE',
},
'JUMPING': {
'file': 'rewards/jumping.mp4',
'source': 'https://www.youtube.com/watch?v=_62o7u-BYRA',
},
'RIDING': {
'file': 'rewards/riding.mp4',
'source': 'https://www.youtube.com/watch?v=dVngVDTYbOQ',
},
'RIVERDANCE': {
'file': 'rewards/riverdance.webm',
'source': 'https://www.youtube.com/watch?v=KF7gk10jgEw',
},
'PAPERTOWEL': {
'file': 'rewards/papertowel.mp4',
'source': 'https://www.youtube.com/watch?v=6_-4KArFR08',
},
'BIRDVSPHONE': {
'file': 'rewards/phonecover.mp4',
'source': 'https://www.youtube.com/watch?v=ktzfC5ZcOtE',
},
'KILLERBIRD': {
'file': 'rewards/killerbird.mp4',
'source': 'https://twitter.com/colloritz/status/798470429308424192',
},
'WITHFRIENDS': {
'file': 'rewards/friends.mp4',
'source': 'https://twitter.com/s_hatachan/status/879946102878883840',
},
}
MSGS = {
'ERROR_NOBOT': {
'text': '''@{sender_acct} Pwii! Pwi.
[FR] Désolé, je n'irai pas voir {receiver_acct}, car cette personne ne semble pas vouloir de la visite de bots.
[EN] Sorry, I won't visit {receiver_acct} as they do not seem to like bots very much.''',
'privacy': 'direct',
},
'ERROR_UNDELIVERABLE': {
'text': '''@{sender_acct} Pwiii…
[FR] Désolé, je n'ai pas réussi à joindre {receiver_acct}… essaie avec quelqu'un d'autre ?
[EN] Sorry, I can't reach {receiver_acct}… maybe try with someone else?''',
'privacy': 'direct',
},
'ERROR_OWNED': {
'text': '''@{sender_acct} Pwi pwi pwi!!!
[FR] Je suis chez quelqu'un d'autre !
Je serai de nouveau disponible si cette personne ne m'envoie pas me promener d'ici {hours}h{minutes:02d} !
Tu peux me demander de te dire quand je serai disponible en me disant « notifie moi » !
[EN] I'm currently visiting someone else!
I'll be available in {hours}h{minutes:02d} if that person doesn't send me around in the meantime!
You can request that I notify you when I'm available by telling me “notify me”!''',
'privacy': 'direct',
},
'ERROR_OWNED2': {
'text': '''@{sender_acct} Pwi pwi pwi!!!
[FR] Je suis chez quelqu'un d'autre !
Je serai de nouveau disponible si cette personne ne m'envoie pas me promener d'ici {hours}h{minutes:02d} !
Je te dirai la prochaine fois que je serai disponible !
[EN] I'm currently visiting someone else!
I'll be available in {hours}h{minutes:02d} if that person doesn't send me around in the meantime!
I'll tell you next time I'm available!''',
'privacy': 'direct',
},
'ERROR_DELIVERY': {
'text': '''@{sender_acct} Pwiii pwii pwii pwiii!
[FR] Je suis déjà en train de m'envoler pour voir quelqu'un d'autre !
Tu peux me demander de te dire quand je serai disponible en me disant « notifie moi » !
[EN] I'm already flying away to someone else!
You can request that I notify you when I'm available by telling me “notify me”!''',
'privacy': 'direct',
},
'ERROR_DELIVERY2': {
'text': '''@{sender_acct} Pwiii pwii pwii pwiii!
[FR] Tu m'as déjà envoyé voir quelqu'un d'autre ! Tu peux encore revenir sur ce choix en me disant « reviens » !
[EN] You already sent me to someone else! If that was a mistake, you can tell me to “come back”!''',
'privacy': 'direct',
},
'ERROR_DELIVERY3': {
'text': '''@{sender_acct} Pwiii pwii pwii pwiii!
[FR] Je suis déjà en train de m'envoler pour voir quelqu'un d'autre !
Je te dirai la prochaine fois que je serai disponible !
[EN] I'm already flying away to someone else!
I'll tell you next time I'm available!''',
'privacy': 'direct',
},
'ERROR_INVALID_FORMAT': {
'text': '''@{sender_acct} Pwii? Pwiii!
[FR] Je n'ai pas compris vers qui je dois m'envoler ? Il faut me dire « va voir utilisateur@domaine » !
[EN] I don't understand who I'm supposed to be flying to? You have to tell me “go see user@domain”!''',
'privacy': 'direct',
},
'ERROR_UNKNOWN_ACCOUNT': {
'text': '''@{sender_acct} Pwiii! Pwiii?
[FR] Je n'ai pas trouvé {acct} ! Vers qui d'autre dois-je aller ?
[EN] I can't find {acct}! Who else should I visit?''',
'privacy': 'direct',
},
'ERROR_UNKNOWN_ACCOUNT2': {
'text': '''@{sender_acct} Pwiii! Pwiii?
[FR] Je n'ai pas trouvé {acct} ! Vers qui d'autre dois-je aller ? Peut-être voulais-tu dire {suggested_acct} ?
[EN] I can't find {acct}! Who else should I visit? Maybe you meant {suggested_acct}?''',
'privacy': 'direct',
},
'ERROR_INTERNAL': {
'text': '''@{sender_acct} Pwiiii…
[FR] Je suis désolé, je n'arrive pas à contacter {acct}… il va falloir essayer avec quelqu'un d'autre.
[EN] I'm sorry I cannot reach {acct}… who should I be flying to?''',
'privacy': 'direct',
},
'ERROR_SAME_ACCOUNT': {
'text': '''@{sender_acct} Pwiii!
[FR] Hé, ce n'est pas du jeu de vouloir me garder comme ça ! Vers qui d'autre dois-je aller ?
[EN] Hey, it's not fair to try to keep me like this! Who else should I visit?''',
'privacy': 'direct',
},
'ERROR_OWN_ACCOUNT': {
'text': '''@{sender_acct} Pwii. Pwii?
[FR] Je ne peux pas aller me voir moi-même ?
[EN] I cannot fly to see myself?''',
'privacy': 'direct',
},
'DELIVERY_START': {
'text': '''@{sender_acct} Pwiipwii pwiipwiii!
[FR] En route vers chez {acct} ! Ça peut me prendre un peu de temps !
Si je me suis trompé de personne, tu peux me dire de revenir en me disant « reviens » !
[EN] On my way to {acct}! This may take me a while!
If I'm not flying to the correct person, you can still tell me to “come back”!''',
'privacy': 'direct',
'media': ['FLYING_AWAY'],
},
'DELIVERY_CANCELLED': {
'text': '''@{sender_acct} Pwiii…
[FR] Tu ne veux plus que j'aille voir {acct} ? D'accord… qui dois-je aller voir ?
[EN] You don't want me to go visit {acct} anymore? Ok… but who should I fly to?''',
'privacy': 'direct',
},
'DELIVERED': {
'text': '''@{receiver_acct} :caique: Pwii pwiii! :caique:
[FR] Bonjour de la part de {sender_acct} ! Je ne suis qu'à toi pour les {nb_hours} prochaines heures ! Envoie-moi voir qui tu veux en répondant « va voir utilisateur@domaine » !
Tu peux aussi me dire « va jouer ailleurs » !
[EN] Hello from {sender_acct_short}! Send me to anyone by replying “go see user@domain”! Until then, I'll be yours for the next {nb_hours} hours!
You can also tell me to “go play somewhere else”!''',
'privacy': 'direct',
'media': {('FLYING_IN',): 3, ('STAIRS',): 1},
},
'IDLE': {
'text': ''':caique: Pwiii… pwii pwii! :caique:
[FR] Personne ne m'a envoyé me promener depuis un moment… du coup, n'importe qui peut le faire en me répondant « va voir utilisateur@domaine » !
[EN] Noone sent me flying away… but now, anybody can! Reply me with “go see user@domain”, and I'll fly to them!''',
'privacy': 'public',
},
'IDLE2': {
'text': ''':caique: Pwiii… pwii pwii! :caique:
[FR] La personne chez qui j'ai été envoyé préfère que j'aille jouer ailleurs ! Du coup, envoyez-moi chez qui vous voulez en me répondant « va voir utilisateur@domaine » !
[EN] The person I was sent to would like me to play somewhere else! Reply me with “go see user@domain”, and I'll fly to them!''',
'privacy': 'public',
},
'NOTIFY_REQUEST': {
'text': '''@{acct} :caique: Pwii! :caique:
[FR] Entendu ! Je te dirai la prochaine fois que je serai disponible !
[EN] Got it! I'll tell you next time I'm available!''',
'privacy': 'direct',
},
'NOTIFY_IDLE': {
'text': '''@{acct} :caique: Pwii pwii! :caique:
[FR] Tu m'avais demandé de te notifier quand je deviendrai disponible ! Et bien c'est chose faite !
[EN] You requested that I notify you when I become available! I'm available now!''',
'privacy': 'direct',
},
'BELLY_RUB': {
'text': ''':caique: Pwiiiiii <3 :caique:
[FR] J'ai rencontré {nb_users} personnes et j'ai apporté du bonheur {nb_likes} fois ! J'ai bien mérité un petit gratouilli !
[EN] I've met {nb_users} people and I've made them happy {nb_likes} times! I've earned myself a belly rub!''',
'privacy': 'public',
'media': ['BELLY_RUB'],
},
'PAPERTOWEL': {
'text': ''':caique: Pwiii… pwii! :caique:
[FR] Des fois, je comprends vraiment pas mon cousin à tête noire… mais bon, il a l'air de bien s'amuser !
[EN] Sometimes, I really don't understand my black-headed cousin… but he seems to be having a lot of fun!''',
'privacy': 'public',
'media': ['PAPERTOWEL'],
},
'JUMPING': {
'text': ''':caique: Pwii? Pwiii! :caique:
[FR] Mon cousin à tête noir commence à sauter sur place… sinon, j'ai rencontré {nb_users} d'entre vous et je leur ai fait plaisir {nb_likes} fois !
[EN] My black-headed cousin is starting to jump in-place… by the way, I've met {nb_users} of you and I've made them happy {nb_likes} times!''',
'privacy': 'public',
'media': ['JUMPING'],
},
'DANCINGQUEEN': {
'text': ''':caique: Pwii~ pwiii~ pwiii~ 🎶 :caique:
[FR] Je vais bientôt rencontrer tout le fediverse ! {nb_users} personnes ! Pfiou ! Sinon, j'aime bien Queen !
[EN] I've met almost all of the fediverse! {nb_users} of you! Wow! Also, I like Queen!''',
'privacy': 'public',
'media': ['DANCINGQUEEN'],
},
'SHOPPING': {
'text': ''':caique: Pwi pwi pwiiii! :caique:
[FR] J'ai rencontré pas moins de {nb_users} mastonautes ! Et ils ont aimé mes visites {nb_likes} fois ! Mon cousin est parti m'acheter à manger !
[EN] I've met {nb_users} mastonauts! And they liked my visits {nb_likes} times! My cousin went shopping for dinner!''',
'privacy': 'public',
'media': ['SHOPPING'],
},
'MICROWAVE': {
'text': ''':caique: Pwi! Pwi! Pwi! Pwi! Pwi! Pwi! :caique:
[FR] J'ai rencontré {nb_users} personnes formidables ! Ils ont aimé mes visites {nb_likes} fois ! Je vais fêter ça en dinant !
[EN] I've met {nb_users} amazing people! And they liked me {nb_likes} times! I'll eat dinner to celebrate!''',
'privacy': 'public',
'media': ['MICROWAVE'],
},
'DJBIRB': {
'text': ''':caique: Pwiii~ pwiii~ pwiii~ pwiii~ :caique:
[FR] Pwiiiii~ j'ai rencontré {nb_users} personnes qui m'ont aimé {nb_likes} fois, je m'amuse comme un petit fou !
[EN] Pwiiiii~ I've met {nb_users} who liked me {nb_likes} times, I'm having a lot of fun!''',
'privacy': 'public',
'media': ['DJBIRB'],
},
'RIDING': {
'text': ''':caique: Pwii pwii pwii~ :caique:
[FR] Il n'y a pas que mon cousin qui peut s'amuser comme ça ! Regardez-moi ! ~
J'ai rencontré {nb_users} personnes et ils m'ont apprécié {nb_likes} fois !
[EN] My cousin is not the only one that can have fun! Look at me! ~
I've met {nb_users} people so far and they liked me {nb_likes} times!''',
'privacy': 'public',
'media': ['RIDING'],
},
'RIVERDANCE': {
'text': ''':caique: Pwii 🎶 :caique:
[FR] Regardez mon cousin danser !
J'ai rencontré {nb_users} personnes et ils m'ont apprécié {nb_likes} fois !
[EN] Watch my cousin dance!
I've met {nb_users} people so far and they liked me {nb_likes} times!''',
'privacy': 'public',
'media': ['RIVERDANCE'],
},
'BIRBVSPHONE': {
'text': ''':caique: Pwii! :caique:
[FR] Ne plaisante pas avec mon cousin, ok ? Ou sinon, gare à ton téléphone !
[EN] Don't mess with my cousin ok? He'll wreak your phone! Or at least its cover!''',
'privacy': 'public',
'media': ['BIRDVSPHONE'],
},
'KILLERBIRD': {
'text': ''':caique: … :caique:
[FR] Mon cousin me fait peur parfois… j'espère qu'il ne fera jamais rien de tel à aucune des {nb_users} personnes que j'ai rencontré !
[EN] I'm afraid of my cousin sometimes… I hope he won't do anything like this to any of the {nb_users} people I've met so far!''',
'privacy': 'public',
'media': ['KILLERBIRD'],
},
'WITHFRIENDS': {
'text': ''':caique: Pwiii pwii pwii~ :caique:
[FR] En train de passer du temps avec mes amis ! Je leur raconte mes rencontres avec vous tous et les {nb_likes} fois où vous avez apprécié mes visites !
[EN] Hanging out with my friends, telling them about all {nb_users} of you and how you liked my visits {nb_likes} times!''',
'privacy': 'public',
'media': ['WITHFRIENDS'],
}
}
REWARDS = [
{'required_likes': 2,
'required_users': 3,
'msg_id': 'BELLY_RUB'},
{'required_likes': 5,
'required_users': 5,
'msg_id': 'JUMPING'},
{'required_likes': 10,
'required_users': 10,
'msg_id': 'DANCINGQUEEN'},
{'required_likes': 20,
'required_users': 12,
'msg_id': 'SHOPPING'},
{'required_likes': 30,
'required_users': 15,
'msg_id': 'MICROWAVE'},
{'required_likes': 40,
'required_users': 20,
'msg_id': 'DJBIRB'},
{'required_likes': 45,
'required_users': 30,
'msg_id': 'PAPERTOWEL'},
{'required_likes': 46,
'required_users': 35,
'msg_id': 'RIDING'},
{'required_likes': 50,
'required_users': 40,
'msg_id': 'RIVERDANCE'},
{'required_likes': 60,
'required_users': 50,
'msg_id': 'BIRBVSPHONE'},
{'required_likes': 70,
'required_users': 60,
'msg_id': 'KILLERBIRD'},
{'required_likes': 80,
'required_users': 70,
'msg_id': 'WITHFRIENDS'},
]

485
delibird-master/main.py Executable file
View File

@ -0,0 +1,485 @@
#!/usr/bin/env python3
import argparse
import re
import datetime
import random
import json
import itertools
from getpass import getpass
from mastodon import Mastodon, StreamListener
from mastodon.Mastodon import MastodonAPIError, MastodonNotFoundError
from data import MEDIA, MSGS, REWARDS
API_BASE = 'https://social.sitedethib.com'
COMMAND_RE = re.compile(r'(va[s]? voir|vole vers|va, vole vers|rends visite à|go see|go visit|fly to)\s*(.+)', re.IGNORECASE)
FREE_RE = re.compile(r"(va[s]? (te promener|jouer))|rends-toi disponible|repose-toi|va-t'en|go idle|go away|((go|play) somewhere else)|take a break", re.IGNORECASE)
CANCEL_RE = re.compile(r'reviens|arrête|annule|stop|come back|cancel', re.IGNORECASE)
MENTION_RE = re.compile(r'([a-z0-9_]+)(@[a-z0-9\.\-]+[a-z0-9]+)?', re.IGNORECASE)
NOTIFY_RE = re.compile(r'dis[ -]moi|notifie[ -]moi|rappelle[ -]moi|tell me|remind me|notify me', re.IGNORECASE)
LINK_RE = re.compile(r'<a href="([^"]+)"')
STATE_OWNED, STATE_IDLE, STATE_DELIVERY = range(3)
MAX_OWNED = datetime.timedelta(hours=3)
class Error(Exception):
"""Base error class for Delibird-generated errors"""
pass
class InternalError(Error):
"""Generic internal server error"""
def __init__(self, acct):
Error.__init__(self)
self.acct = acct
class InvalidFormatError(Error):
"""Error thrown when a command does not contain a properly-formatted
account"""
pass
class AccountNotFoundError(Error):
"""Error thrown when the requested account cannot be resolved"""
def __init__(self, acct):
Error.__init__(self)
self.acct = acct
class Delibird(StreamListener):
"""Main class for the Delibird bot."""
def __init__(self, mastodon):
StreamListener.__init__(self)
self.mastodon = mastodon
self.state = STATE_IDLE
self.owner = None
self.target = None
self.last_owned = datetime.datetime.now()
self.like_count = 0
self.visited_users = set()
self.to_be_notified = set()
self.reward_level = -1
self.last_idle_toot = None
self.additional_idle_toots = []
self.last_read_notification = None
self.own_acct_id = None
self.last_request_id = None
self.visit_to_request_map = {}
print('Delibird started!')
self.load()
self.resume()
def resume(self):
"""Process missed notifications."""
# Process all missed notifications
last_notification = self.last_read_notification
if last_notification is not None:
for notification in reversed(self.mastodon.notifications(since_id=last_notification)):
self.on_notification(notification)
def save(self, path='state.json'):
"""Save state to a JSON file."""
state = {'like_count': self.like_count,
'visited_users': list(self.visited_users),
'to_be_notified': list(self.to_be_notified),
'reward_level': self.reward_level,
'state': self.state,
'last_owned': self.last_owned.isoformat(),
'additional_idle_toots': self.additional_idle_toots,
'visit_to_request_map': self.visit_to_request_map}
if self.last_request_id is not None:
state['last_request_id'] = self.last_request_id
if self.last_read_notification is not None:
state['last_read_notification'] = self.last_read_notification
if self.last_idle_toot is not None:
state['last_idle_toot'] = self.last_idle_toot.id
if self.owner is not None:
state['owner'] = self.owner.id
if self.target is not None:
state['target'] = self.target.id
if self.own_acct_id is not None:
state['own_acct_id'] = self.own_acct_id
with open(path, 'w') as file:
json.dump(state, file)
def load(self, path='state.json'):
"""Load state from a JSON file.
May perform API requests to retrieve status or account information."""
try:
with open(path, 'r') as file:
state = json.load(file)
self.like_count = state['like_count']
self.visited_users = set(state['visited_users'])
self.to_be_notified = set(state.get('to_be_notified', []))
self.reward_level = state['reward_level']
self.state = state.get('state', STATE_IDLE)
self.last_read_notification = state.get('last_read_notification', None)
self.own_acct_id = state.get('own_acct_id', None)
self.additional_idle_toots = state.get('additional_idle_toots', [])
self.visit_to_request_map = state.get('visit_to_request_map', {})
self.last_request_id = state.get('last_request_id', None)
last_idle_toot = state.get('last_idle_toot', None)
owner = state.get('owner', None)
target = state.get('target', None)
last_owned = state.get('last_owned', None)
if last_owned:
self.last_owned = datetime.datetime.strptime(last_owned, '%Y-%m-%dT%H:%M:%S.%f')
self.last_idle_toot = None if last_idle_toot is None else self.mastodon.status(last_idle_toot)
self.owner = None if owner is None else self.mastodon.account(owner)
self.target = None if target is None else self.mastodon.account(target)
except FileNotFoundError:
pass
def handle_rewards(self):
"""Check rewards conditions and post appropriate toot if conditions for a
new reward are met."""
level = -1
for i, reward in enumerate(REWARDS):
if (self.like_count >= reward['required_likes']
and len(self.visited_users) >= reward['required_users']):
level = i
if level > self.reward_level:
self.reward_level = level
self.send_toot(REWARDS[level]['msg_id'],
nb_likes=self.like_count, nb_users=len(self.visited_users))
self.save()
def upload_media(self, name):
"""Look up a media file description and upload it."""
desc = MEDIA[name]
media = self.mastodon.media_post(desc['file'],
description='Source: %s' % desc['source'])
print('Uploaded %s!' % name)
return media
def send_toot(self, msg_id, in_reply_to_id=None, **kwargs):
"""Look up a toot's description by message id and sends it."""
print('Sending a toot… id: %s' % msg_id)
msg = MSGS[msg_id]
if 'media' in msg:
if isinstance(msg['media'], dict):
grouped_choices = ([key] * count for key, count in msg['media'].items())
choices = list(itertools.chain.from_iterable(grouped_choices))
media_desc = random.choice(choices)
else:
media_desc = msg['media']
media = [self.upload_media(name) for name in media_desc]
else:
media = None
try:
status = self.mastodon.status_post(msg['text'].format(**kwargs),
media_ids=media,
in_reply_to_id=in_reply_to_id,
visibility=msg.get('privacy', ''))
except MastodonNotFoundError:
# Original status deleted
status = self.mastodon.status_post(msg['text'].format(**kwargs),
media_ids=media,
visibility=msg.get('privacy', ''))
self.own_acct_id = status.account.id
self.save()
return status
def resolve_account(self, text_with_user, status):
"""Process command text to resolve an account from account name, mention
or profile URL"""
receiver_acct = None
# First, see if we have a handle we can resolve, without the leading '@'
match = MENTION_RE.match(text_with_user)
if match:
if match.group(2):
# If it's a full handle, use it
receiver_acct = match.group(0)
else:
# If it's *not* a full handle, append the user's domain
receiver_acct = '@'.join([match.group(1)] + status.account.acct.split('@')[1:])
try:
matches = self.mastodon.account_search(receiver_acct)
except MastodonAPIError:
raise InternalError(receiver_acct)
if not matches:
raise AccountNotFoundError(receiver_acct)
return matches[0]
# Maybe it's a link to their profile, or a mention. Switch to link handling.
match = LINK_RE.search(text_with_user)
if match:
url = match.group(1)
try:
matches = self.mastodon.search(url, resolve=True).accounts
except MastodonAPIError:
raise InternalError(url)
if matches:
return matches[0]
raise InvalidFormatError
def handle_unknown_account(self, status, acct):
"""Handle unknown accounts, potentially suggesting other accounts."""
receiver_acct = acct.split('@')
suggested_account = None
if len(receiver_acct) > 1:
username, domain = receiver_acct
try:
matches = self.mastodon.account_search(username, limit=40)
except MastodonAPIError:
pass
else:
for match in matches:
if match.username != username:
continue
if domain in match.acct:
suggested_account = match
break
if suggested_account is None:
self.send_toot('ERROR_UNKNOWN_ACCOUNT', status,
sender_acct=status.account.acct, acct=acct)
else:
self.send_toot('ERROR_UNKNOWN_ACCOUNT2', status,
sender_acct=status.account.acct, acct=acct,
suggested_acct=suggested_account.acct)
def handle_cmd_free(self, status, match=None):
"""Handle the command that allows the bird to go idle prematurely"""
if not self.owner or self.owner.id != status.account.id:
return
if self.state != STATE_OWNED:
return
self.go_idle('IDLE2')
self.save()
def handle_cmd_go_see(self, status, match):
"""Handle the “go see” command requesting the bot to visit a given user"""
text_with_user = match.group(2)
if self.state == STATE_DELIVERY:
if self.owner.id == status.account.id:
self.send_toot('ERROR_DELIVERY2', status, sender_acct=status.account.acct)
elif status.account.acct in self.to_be_notified:
self.send_toot('ERROR_DELIVERY3', status, sender_acct=status.account.acct)
else:
self.send_toot('ERROR_DELIVERY', status, sender_acct=status.account.acct)
return
if self.state == STATE_OWNED and self.owner and self.owner.id != status.account.id:
delta = self.last_owned + MAX_OWNED - datetime.datetime.now()
minutes = delta.seconds // 60
if status.account.acct in self.to_be_notified:
self.send_toot('ERROR_OWNED2', status, sender_acct=status.account.acct,
hours=(minutes // 60), minutes=(minutes % 60))
else:
self.send_toot('ERROR_OWNED', status, sender_acct=status.account.acct,
hours=(minutes // 60), minutes=(minutes % 60))
return
try:
target = self.resolve_account(text_with_user, status)
except AccountNotFoundError as err:
self.handle_unknown_account(status, err.acct)
return
except InvalidFormatError:
self.send_toot('ERROR_INVALID_FORMAT', status,
sender_acct=status.account.acct)
return
except InternalError as err:
self.send_toot('ERROR_INTERNAL', status,
sender_acct=status.account.acct, acct=err.acct)
return
if target.id == self.own_acct_id:
self.send_toot('ERROR_OWN_ACCOUNT', status,
sender_acct=status.account.acct)
return
if target.id == status.account.id:
self.send_toot('ERROR_SAME_ACCOUNT', status,
sender_acct=status.account.acct)
return
if '#nobot' in target.note:
self.send_toot('ERROR_NOBOT', status,
sender_acct=status.account.acct,
receiver_acct=target.acct)
return
self.state = STATE_DELIVERY
self.last_request_id = status.id
self.owner = status.account
self.last_owned = datetime.datetime.now()
self.target = target
if self.last_idle_toot is not None:
try:
self.mastodon.status_delete(self.last_idle_toot)
except MastodonAPIError:
pass
self.last_idle_toot = None
for toot_id in self.additional_idle_toots:
try:
self.mastodon.status_delete(toot_id)
except MastodonAPIError:
pass
self.additional_idle_toots = []
self.send_toot('DELIVERY_START', status, sender_acct=status.account.acct, acct=self.target.acct)
def handle_cmd_cancel(self, status, match=None):
"""Handle the “cancel” command that cancels the last ordered delivery if the
user issuing it is the current owner and the delivery hasn't finished yet."""
if not self.owner or self.owner.id != status.account.id:
return
if self.state != STATE_DELIVERY:
return
self.state = STATE_OWNED
self.send_toot('DELIVERY_CANCELLED', status, sender_acct=status.account.acct,
acct=self.target.acct)
def handle_cmd_notify(self, status, match=None):
"""Handle request to be notified when the Birb goes idle."""
if self.state == STATE_IDLE:
return
self.to_be_notified.add(status.account.acct)
self.send_toot('NOTIFY_REQUEST', status, acct=status.account.acct)
def handle_mention(self, status):
"""Handle toots mentioning Delibird, which may contain commands"""
print('Got a mention!')
# Do not reply if multiple people are mentionned
if len(status.mentions) > 2:
return
# Process commands, in order of priority
cmds = [(COMMAND_RE, self.handle_cmd_go_see),
(CANCEL_RE, self.handle_cmd_cancel),
(FREE_RE, self.handle_cmd_free),
(NOTIFY_RE, self.handle_cmd_notify)]
for regexp, handler in cmds:
match = regexp.search(status.content)
if match:
handler(status, match)
return
def deliver(self):
"""Deliver a message to the target, updating ownership and state in the
process"""
try:
status = self.send_toot('DELIVERED',
sender_acct=self.owner.acct,
sender_acct_short=self.owner.acct,
receiver_acct=self.target.acct,
nb_hours=(MAX_OWNED.seconds // 3600))
except MastodonAPIError:
try:
status = self.send_toot('DELIVERED',
sender_acct=self.owner.acct,
sender_acct_short=self.owner.username,
receiver_acct=self.target.acct,
nb_hours=(MAX_OWNED.seconds // 3600))
except MastodonAPIError:
status = self.send_toot('DELIVERED',
sender_acct=self.owner.username,
sender_acct_short=self.owner.username,
receiver_acct=self.target.acct,
nb_hours=(MAX_OWNED.seconds // 3600))
if status.mentions:
self.visit_to_request_map[status.id] = self.last_request_id
self.owner = self.target
self.last_owned = datetime.datetime.now()
self.state = STATE_OWNED
self.visited_users.add(self.owner.id)
self.to_be_notified.discard(self.owner.acct)
self.save()
else:
# Could not deliver, maybe OStatus-only?
self.state = STATE_OWNED
self.send_toot('ERROR_UNDELIVERABLE',
sender_acct=self.owner.acct,
receiver_acct=self.target.acct)
def go_idle(self, msg_id='IDLE'):
"""Turn idle and announce it with a public toot"""
self.state = STATE_IDLE
self.last_idle_toot = self.send_toot(msg_id)
self.save()
self.additional_idle_toots = [self.send_toot('NOTIFY_IDLE', self.last_idle_toot,
acct=acct).id
for acct in self.to_be_notified]
self.to_be_notified = set()
self.save()
def handle_heartbeat(self):
if self.state == STATE_DELIVERY and self.target is not None and random.random() >= 0.94:
self.deliver()
elif self.state == STATE_DELIVERY:
self.handle_rewards()
elif self.state == STATE_OWNED and datetime.datetime.now() - self.last_owned > MAX_OWNED:
self.go_idle()
def on_notification(self, notification):
self.last_read_notification = notification.id
if notification.type == 'mention':
self.handle_mention(notification.status)
if notification.type == 'favourite' and notification.status.visibility == 'direct':
self.like_count += 1
if notification.status.id in self.visit_to_request_map:
try:
self.mastodon.status_favourite(self.visit_to_request_map[notification.status.id])
except MastodonNotFoundError:
del self.visit_to_request_map[notification.status.id]
except MastodonAPIError:
pass
self.save()
def register(args):
"""Register app on the server"""
Mastodon.create_app('Delibird', api_base_url=args.api_base, to_file='secrets/clientcred.secret')
def login(args):
"""Log in as the given user, generating OAuth credentials"""
mastodon = Mastodon(client_id='secrets/clientcred.secret', api_base_url=args.api_base)
mastodon.log_in(args.user_mail, getpass(), to_file='secrets/usercred.secret')
def run(args):
"""Run the bot"""
mastodon = Mastodon(client_id='secrets/clientcred.secret',
access_token='secrets/usercred.secret',
api_base_url=args.api_base)
delibird = Delibird(mastodon)
print('Starting streaming!')
mastodon.stream_user(delibird)
def main():
parser = argparse.ArgumentParser()
parser.add_argument('command', type=str, choices=['register', 'login', 'run'])
parser.add_argument('-a', '--api-base', type=str, default=API_BASE)
parser.add_argument('-u', '--user-mail', type=str)
args = parser.parse_args()
if args.command == 'register':
register(args)
elif args.command == 'login':
login(args)
elif args.command == 'run':
run(args)
main()