385 líneas
10 KiB
Python
385 líneas
10 KiB
Python
from mastodon import Mastodon
|
|
from mastodon.Mastodon import MastodonMalformedEventError, MastodonNetworkError, MastodonReadTimeout, MastodonAPIError, MastodonIllegalArgumentError
|
|
import getpass
|
|
import unidecode
|
|
import fileinput,re
|
|
import os
|
|
import sys
|
|
import os.path
|
|
|
|
class Mastobot:
|
|
|
|
name = 'Mastobot'
|
|
|
|
def __init__(self, mastodon=None, mastodon_hostname=None):
|
|
|
|
file_path = "secrets/secrets.txt"
|
|
|
|
is_setup = self.check_setup(file_path)
|
|
|
|
if is_setup:
|
|
|
|
self.uc_client_id = self.get_parameter("uc_client_id", file_path)
|
|
self.uc_client_secret = self.get_parameter("uc_client_secret", file_path)
|
|
self.uc_access_token = self.get_parameter("uc_access_token", file_path)
|
|
|
|
self.mastodon, self.mastodon_hostname = self.log_in(self)
|
|
|
|
else:
|
|
|
|
while(True):
|
|
|
|
logged_in, self.mastodon, self.mastodon_hostname = self.setup()
|
|
|
|
if not logged_in:
|
|
|
|
print("\nLog in failed! Try again.\n")
|
|
|
|
else:
|
|
|
|
break
|
|
|
|
def get_data(self, notif):
|
|
|
|
notification_id = notif.id
|
|
|
|
if notif.type != 'mention':
|
|
|
|
print(f'dismissing notification {notification_id}')
|
|
|
|
self.mastodon.notifications_dismiss(notification_id)
|
|
|
|
return
|
|
|
|
account_id = notif.account.id
|
|
|
|
username = notif.account.acct
|
|
|
|
status_id = notif.status.id
|
|
|
|
text = notif.status.content
|
|
|
|
visibility = notif.status.visibility
|
|
|
|
reply, question = self.get_question(self, text)
|
|
|
|
if reply:
|
|
|
|
mau = self.mastodon.instance_nodeinfo().usage.users.activeMonth
|
|
|
|
mau = '{:,}'.format(mau).replace(',','.')
|
|
|
|
registers = self.mastodon.instance().stats.user_count
|
|
|
|
registers = '{:,}'.format(registers).replace(',','.')
|
|
|
|
posts = self.mastodon.instance().stats.status_count
|
|
|
|
posts = '{:,}'.format(posts).replace(',','.')
|
|
|
|
peers = self.mastodon.instance().stats.domain_count
|
|
|
|
peers = '{:,}'.format(peers).replace(',','.')
|
|
|
|
version = self.mastodon.instance().version
|
|
|
|
reg_open = self.mastodon.instance_nodeinfo().openRegistrations
|
|
|
|
if reg_open:
|
|
|
|
opened = 'obert'
|
|
|
|
else:
|
|
|
|
opened = 'tancat'
|
|
|
|
post_text = f"@{username}, dades de {self.mastodon_hostname}:\n\n"
|
|
|
|
post_text += f"Usuaris registrats: {registers}\n"
|
|
|
|
post_text += f"Usuaris actius (mes): {mau}\n"
|
|
|
|
post_text += f"Apunts: {posts}\n"
|
|
|
|
post_text += f"Servidors federats: {peers}\n\n"
|
|
|
|
post_text += f"Versió Mastodon: v{version}\n"
|
|
|
|
post_text += f"Registre: {opened}"
|
|
|
|
post_text = (post_text[:400] + '... ') if len(post_text) > 400 else post_text
|
|
|
|
self.mastodon.status_post(post_text, in_reply_to_id=status_id,visibility=visibility)
|
|
|
|
print(f'Replied notification {notification_id}')
|
|
|
|
self.mastodon.notifications_dismiss(notification_id)
|
|
|
|
else:
|
|
|
|
print(f'dismissing notification {notification_id}')
|
|
|
|
self.mastodon.notifications_dismiss(notification_id)
|
|
|
|
return
|
|
|
|
@staticmethod
|
|
def get_question(self, text):
|
|
|
|
reply = False
|
|
|
|
keyword = ''
|
|
|
|
content = self.cleanhtml(self, text)
|
|
|
|
content = self.unescape(self, content)
|
|
|
|
try:
|
|
|
|
start = content.index("@")
|
|
end = content.index(" ")
|
|
if len(content) > end:
|
|
|
|
content = content[0: start:] + content[end +1::]
|
|
|
|
cleanit = content.count('@')
|
|
|
|
i = 0
|
|
while i < cleanit :
|
|
|
|
start = content.rfind("@")
|
|
end = len(content)
|
|
content = content[0: start:] + content[end +1::]
|
|
i += 1
|
|
|
|
content = content.lower()
|
|
question = content
|
|
|
|
keyword_length = 4
|
|
keyword = 'info'
|
|
|
|
if unidecode.unidecode(question)[0:keyword_length] == keyword:
|
|
|
|
reply = True
|
|
|
|
except:
|
|
|
|
pass
|
|
|
|
return (reply, question)
|
|
|
|
@staticmethod
|
|
def log_in(self):
|
|
|
|
file_path = "secrets/secrets.txt"
|
|
uc_client_id = self.get_parameter("uc_client_id", file_path)
|
|
uc_client_secret = self.get_parameter("uc_client_secret", file_path)
|
|
uc_access_token = self.get_parameter("uc_access_token", file_path)
|
|
|
|
file_path = "config/config.txt"
|
|
self.mastodon_hostname = self.get_parameter("mastodon_hostname", file_path)
|
|
|
|
self.mastodon = Mastodon(
|
|
client_id = uc_client_id,
|
|
client_secret = uc_client_secret,
|
|
access_token = uc_access_token,
|
|
api_base_url = 'https://' + self.mastodon_hostname,
|
|
)
|
|
|
|
headers={ 'Authorization': 'Bearer %s'%uc_access_token }
|
|
|
|
return (self.mastodon, self.mastodon_hostname)
|
|
|
|
@staticmethod
|
|
def check_setup(file_path):
|
|
|
|
is_setup = False
|
|
|
|
if not os.path.isfile(file_path):
|
|
print(f"File {file_path} not found, running setup.")
|
|
return
|
|
else:
|
|
is_setup = True
|
|
return is_setup
|
|
|
|
@staticmethod
|
|
def get_parameter(parameter, file_path ):
|
|
|
|
with open( file_path ) as f:
|
|
for line in f:
|
|
if line.startswith( parameter ):
|
|
return line.replace(parameter + ":", "").strip()
|
|
|
|
print(f'{file_path} Missing parameter {parameter}')
|
|
sys.exit(0)
|
|
|
|
@staticmethod
|
|
def cleanhtml(self, raw_html):
|
|
|
|
cleanr = re.compile('<.*?>')
|
|
cleantext = re.sub(cleanr, '', raw_html)
|
|
return cleantext
|
|
|
|
@staticmethod
|
|
def unescape(self, s):
|
|
|
|
s = s.replace("'", "'")
|
|
return s
|
|
|
|
@staticmethod
|
|
def modify_file(self, file_name, pattern,value=""):
|
|
|
|
fh=fileinput.input(file_name,inplace=True)
|
|
|
|
for line in fh:
|
|
|
|
replacement=pattern + value
|
|
line=re.sub(pattern,replacement,line)
|
|
sys.stdout.write(line)
|
|
|
|
fh.close()
|
|
|
|
@staticmethod
|
|
def create_config(self):
|
|
|
|
if not os.path.exists('config'):
|
|
|
|
os.makedirs('config')
|
|
|
|
if not os.path.exists(self.config_filepath):
|
|
|
|
print(self.config_filepath + " created!")
|
|
with open('config/config.txt', 'w'): pass
|
|
|
|
@staticmethod
|
|
def write_config(self):
|
|
|
|
with open(self.config_filepath, 'a') as the_file:
|
|
|
|
the_file.write('mastodon_hostname: \n')
|
|
print(f"adding parameter 'mastodon_hostname' to {self.config_filepath}")
|
|
|
|
@staticmethod
|
|
def read_config_line(self):
|
|
|
|
with open(self.config_filepath) as fp:
|
|
|
|
line = fp.readline()
|
|
self.modify_file(self, self.config_filepath, "mastodon_hostname: ", value=self.mastodon_hostname)
|
|
|
|
def setup(self):
|
|
|
|
logged_in = False
|
|
|
|
try:
|
|
|
|
self.mastodon_hostname = input("Enter Mastodon hostname (or 'q' to exit): ")
|
|
|
|
if self.mastodon_hostname == 'q':
|
|
|
|
sys.exit("Bye")
|
|
|
|
user_name = input("User email, ex. user@" + self.mastodon_hostname +"? ")
|
|
user_password = getpass.getpass("User password? ")
|
|
app_name = input("App name? ")
|
|
|
|
Mastodon.create_app(app_name, scopes=["read","write"],
|
|
|
|
to_file="app_clientcred.txt", api_base_url=self.mastodon_hostname)
|
|
|
|
mastodon = Mastodon(client_id = "app_clientcred.txt", api_base_url = self.mastodon_hostname)
|
|
|
|
mastodon.log_in(
|
|
user_name,
|
|
user_password,
|
|
scopes = ["read", "write"],
|
|
to_file = "app_usercred.txt"
|
|
)
|
|
|
|
if os.path.isfile("app_usercred.txt"):
|
|
|
|
print(f"Log in succesful!")
|
|
logged_in = True
|
|
|
|
if not os.path.exists('secrets'):
|
|
os.makedirs('secrets')
|
|
|
|
secrets_filepath = 'secrets/secrets.txt'
|
|
|
|
if not os.path.exists(secrets_filepath):
|
|
with open(secrets_filepath, 'w'): pass
|
|
print(f"{secrets_filepath} created!")
|
|
|
|
with open(secrets_filepath, 'a') as the_file:
|
|
print("Writing secrets parameter names to " + secrets_filepath)
|
|
the_file.write('uc_client_id: \n'+'uc_client_secret: \n'+'uc_access_token: \n')
|
|
|
|
client_path = 'app_clientcred.txt'
|
|
|
|
with open(client_path) as fp:
|
|
|
|
line = fp.readline()
|
|
cnt = 1
|
|
|
|
while line:
|
|
|
|
if cnt == 1:
|
|
|
|
print("Writing client id to " + secrets_filepath)
|
|
self.modify_file(self, secrets_filepath, "uc_client_id: ", value=line.rstrip())
|
|
|
|
elif cnt == 2:
|
|
|
|
print("Writing client secret to " + secrets_filepath)
|
|
self.modify_file(self, secrets_filepath, "uc_client_secret: ", value=line.rstrip())
|
|
|
|
line = fp.readline()
|
|
cnt += 1
|
|
|
|
token_path = 'app_usercred.txt'
|
|
|
|
with open(token_path) as fp:
|
|
|
|
line = fp.readline()
|
|
print("Writing access token to " + secrets_filepath)
|
|
self.modify_file(self, secrets_filepath, "uc_access_token: ", value=line.rstrip())
|
|
|
|
if os.path.exists("app_clientcred.txt"):
|
|
|
|
print("Removing app_clientcred.txt temp file..")
|
|
os.remove("app_clientcred.txt")
|
|
|
|
if os.path.exists("app_usercred.txt"):
|
|
|
|
print("Removing app_usercred.txt temp file..")
|
|
os.remove("app_usercred.txt")
|
|
|
|
self.config_filepath = 'config/config.txt'
|
|
|
|
self.create_config(self)
|
|
self.write_config(self)
|
|
self.read_config_line(self)
|
|
|
|
print("Secrets setup done!\n")
|
|
|
|
except MastodonIllegalArgumentError as i_error:
|
|
|
|
sys.stdout.write(f'\n{str(i_error)}\n')
|
|
|
|
except MastodonNetworkError as n_error:
|
|
|
|
sys.stdout.write(f'\n{str(n_error)}\n')
|
|
|
|
except MastodonReadTimeout as r_error:
|
|
|
|
sys.stdout.write(f'\n{str(r_error)}\n')
|
|
|
|
except MastodonAPIError as a_error:
|
|
|
|
sys.stdout.write(f'\n{str(a_error)}\n')
|
|
|
|
return (logged_in, mastodon, self.mastodon_hostname)
|
|
|
|
|
|
|