spla/xmpp
1
0
Fork 0
Codi Incidències Sol·licituds Pull repo.project Llançaments Paquets Wiki Activitat

Hide non public methods and new feature: status!

Explorar la Font
This commit is contained in:
spla 2022-08-09 14:34:29 +02:00
pare e603fc6d79
commit 36860a55ec
S'han modificat 3 arxius amb 115 adicions i 87 eliminacions
Mostrar estadístiques Descarregar el fitxer de pedaç Descarregar el fitxer de diferències Expand all files Collapse all files

111
akkomabot.py
Veure arxiu

@ -31,21 +31,21 @@ class Akkomabot:
file_path = "secrets/secrets.txt" file_path = "secrets/secrets.txt"
is_setup = self.check_setup(file_path) is_setup = self.__check_setup(file_path)
if is_setup: if is_setup:
self.uc_client_id = self.get_parameter("uc_client_id", file_path) 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_client_secret = self.__get_parameter("uc_client_secret", file_path)
self.uc_access_token = self.get_parameter("uc_access_token", file_path) self.__uc_access_token = self.__get_parameter("uc_access_token", file_path)
self.akkoma, self.akkoma_hostname = self.log_in(self) self.akkoma, self.akkoma_hostname = self.__log_in(self)
bot_file_path = "config/config.txt" bot_file_path = "config/config.txt"
self.bot_lang = self.get_parameter("bot_lang" , bot_file_path) self.__bot_lang = self.__get_parameter("bot_lang" , bot_file_path)
self.load_strings(self) self.__load_strings(self)
else: else:
@ -61,22 +61,22 @@ class Akkomabot:
bot_file_path = "config/config.txt" bot_file_path = "config/config.txt"
self.bot_lang = self.get_parameter("bot_lang" , bot_file_path) self.__bot_lang = self.__get_parameter("bot_lang" , bot_file_path)
self.load_strings(self) self.__load_strings(self)
break break
@staticmethod @staticmethod
def log_in(self): def __log_in(self):
file_path = "secrets/secrets.txt" file_path = "secrets/secrets.txt"
uc_client_id = self.get_parameter("uc_client_id", file_path) uc_client_id = self.__get_parameter("uc_client_id", file_path)
uc_client_secret = self.get_parameter("uc_client_secret", file_path) uc_client_secret = self.__get_parameter("uc_client_secret", file_path)
uc_access_token = self.get_parameter("uc_access_token", file_path) uc_access_token = self.__get_parameter("uc_access_token", file_path)
file_path = "config/config.txt" file_path = "config/config.txt"
self.akkoma_hostname = self.get_parameter("akkoma_hostname", file_path) self.akkoma_hostname = self.__get_parameter("akkoma_hostname", file_path)
self.akkoma = Akkoma( self.akkoma = Akkoma(
client_id = uc_client_id, client_id = uc_client_id,
@ -90,7 +90,7 @@ class Akkomabot:
return (self.akkoma, self.akkoma_hostname) return (self.akkoma, self.akkoma_hostname)
@staticmethod @staticmethod
def check_setup(file_path): def __check_setup(file_path):
is_setup = False is_setup = False
@ -116,7 +116,7 @@ class Akkomabot:
user_name = input("Bot user name, ex. john? ") user_name = input("Bot user name, ex. john? ")
user_password = getpass.getpass("Bot password? ") user_password = getpass.getpass("Bot password? ")
app_name = input("Bot App name? ") app_name = input("Bot App name? ")
self.bot_lang = input("Bot replies lang (ca or en)? ") self.__bot_lang = input("Bot replies lang (ca or en)? ")
client_id, client_secret = Akkoma.create_app( client_id, client_secret = Akkoma.create_app(
app_name, app_name,
@ -128,7 +128,7 @@ class Akkomabot:
grant_type = 'password' grant_type = 'password'
akkoma.log_in( akkoma.__log_in(
client_id, client_id,
client_secret, client_secret,
grant_type, grant_type,
@ -168,12 +168,12 @@ class Akkomabot:
if cnt == 1: if cnt == 1:
print("Writing client id to " + secrets_filepath) print("Writing client id to " + secrets_filepath)
self.modify_file(self, secrets_filepath, "uc_client_id: ", value=line.rstrip()) self.__modify_file(self, secrets_filepath, "uc_client_id: ", value=line.rstrip())
elif cnt == 2: elif cnt == 2:
print("Writing client secret to " + secrets_filepath) print("Writing client secret to " + secrets_filepath)
self.modify_file(self, secrets_filepath, "uc_client_secret: ", value=line.rstrip()) self.__modify_file(self, secrets_filepath, "uc_client_secret: ", value=line.rstrip())
line = fp.readline() line = fp.readline()
cnt += 1 cnt += 1
@ -184,7 +184,7 @@ class Akkomabot:
line = fp.readline() line = fp.readline()
print("Writing access token to " + secrets_filepath) print("Writing access token to " + secrets_filepath)
self.modify_file(self, secrets_filepath, "uc_access_token: ", value=line.rstrip()) self.__modify_file(self, secrets_filepath, "uc_access_token: ", value=line.rstrip())
if os.path.exists("app_clientcred.txt"): if os.path.exists("app_clientcred.txt"):
@ -198,9 +198,9 @@ class Akkomabot:
self.config_filepath = 'config/config.txt' self.config_filepath = 'config/config.txt'
self.create_config(self) self.__create_config(self)
self.write_config(self) self.__write_config(self)
self.read_config_line(self) self.__read_config_line(self)
print("Secrets setup done!\n") print("Secrets setup done!\n")
@ -223,7 +223,7 @@ class Akkomabot:
return (logged_in, akkoma, self.akkoma_hostname) return (logged_in, akkoma, self.akkoma_hostname)
@staticmethod @staticmethod
def get_parameter(parameter, file_path ): def __get_parameter(parameter, file_path ):
with open( file_path ) as f: with open( file_path ) as f:
for line in f: for line in f:
@ -234,29 +234,30 @@ class Akkomabot:
sys.exit(0) sys.exit(0)
@staticmethod @staticmethod
def load_strings(self): def __load_strings(self):
lang_file_path = f"app/locales/{self.bot_lang}.txt" lang_file_path = f"app/locales/{self.__bot_lang}.txt"
self.register_str = self.get_parameter("register_str", lang_file_path) self.register_str = self.__get_parameter("register_str", lang_file_path)
self.unregister_str = self.get_parameter("unregister_str", lang_file_path) self.unregister_str = self.__get_parameter("unregister_str", lang_file_path)
self.stats_str = self.get_parameter("stats_str", lang_file_path) self.stats_str = self.__get_parameter("stats_str", lang_file_path)
self.registerok_str = self.get_parameter("registerok_str", lang_file_path) self.status_str = self.__get_parameter("status_str", lang_file_path)
self.user_str = self.get_parameter("user_str", lang_file_path) self.registerok_str = self.__get_parameter("registerok_str", lang_file_path)
self.password_str = self.get_parameter("password_str", lang_file_path) self.user_str = self.__get_parameter("user_str", lang_file_path)
self.server_str = self.get_parameter("server_str", lang_file_path) self.password_str = self.__get_parameter("password_str", lang_file_path)
self.xmpp_account_str = self.get_parameter("xmpp_account_str", lang_file_path) self.server_str = self.__get_parameter("server_str", lang_file_path)
self.notdeleteadmin_str = self.get_parameter("notdeleteadmin_str", lang_file_path) self.xmpp_account_str = self.__get_parameter("xmpp_account_str", lang_file_path)
self.deleted_str = self.get_parameter("deleted_str", lang_file_path) self.notdeleteadmin_str = self.__get_parameter("notdeleteadmin_str", lang_file_path)
self.stats_title_str = self.get_parameter("stats_title_str", lang_file_path) self.deleted_str = self.__get_parameter("deleted_str", lang_file_path)
self.registered_users_str = self.get_parameter("registered_users_str", lang_file_path) self.stats_title_str = self.__get_parameter("stats_title_str", lang_file_path)
self.users_online_str = self.get_parameter("users_online_str", lang_file_path) self.registered_users_str = self.__get_parameter("registered_users_str", lang_file_path)
self.node_users_str = self.get_parameter("node_users_str", lang_file_path) self.users_online_str = self.__get_parameter("users_online_str", lang_file_path)
self.uptime_str = self.get_parameter("uptime_str", lang_file_path) self.node_users_str = self.__get_parameter("node_users_str", lang_file_path)
self.processes_str = self.get_parameter("processes_str", lang_file_path) self.uptime_str = self.__get_parameter("uptime_str", lang_file_path)
self.processes_str = self.__get_parameter("processes_str", lang_file_path)
@staticmethod @staticmethod
def modify_file(self, file_name, pattern,value=""): def __modify_file(self, file_name, pattern,value=""):
fh=fileinput.input(file_name,inplace=True) fh=fileinput.input(file_name,inplace=True)
@ -269,7 +270,7 @@ class Akkomabot:
fh.close() fh.close()
@staticmethod @staticmethod
def create_config(self): def __create_config(self):
if not os.path.exists('config'): if not os.path.exists('config'):
@ -281,7 +282,7 @@ class Akkomabot:
with open('config/config.txt', 'w'): pass with open('config/config.txt', 'w'): pass
@staticmethod @staticmethod
def write_config(self): def __write_config(self):
with open(self.config_filepath, 'a') as the_file: with open(self.config_filepath, 'a') as the_file:
@ -290,13 +291,13 @@ class Akkomabot:
print(f"adding parameter 'bot_lang' to {self.config_filepath}") print(f"adding parameter 'bot_lang' to {self.config_filepath}")
@staticmethod @staticmethod
def read_config_line(self): def __read_config_line(self):
with open(self.config_filepath) as fp: with open(self.config_filepath) as fp:
line = fp.readline() line = fp.readline()
self.modify_file(self, self.config_filepath, "akkoma_hostname: ", value=self.akkoma_hostname) self.__modify_file(self, self.config_filepath, "akkoma_hostname: ", value=self.akkoma_hostname)
self.modify_file(self, self.config_filepath, "bot_lang: ", value=self.bot_lang) self.__modify_file(self, self.config_filepath, "bot_lang: ", value=self.__bot_lang)
def get_data(self, notif): def get_data(self, notif):
@ -312,7 +313,7 @@ class Akkomabot:
visibility = notif.status.visibility visibility = notif.status.visibility
reply, question = self.get_question(self, text) reply, question = self.__get_question(self, text)
mention_dict = {'reply': reply, 'question': question, 'id': id, 'account_id': account_id, 'acct': acct, 'status_id': status_id, 'text': text, 'visibility': visibility} mention_dict = {'reply': reply, 'question': question, 'id': id, 'account_id': account_id, 'acct': acct, 'status_id': status_id, 'text': text, 'visibility': visibility}
@ -321,15 +322,15 @@ class Akkomabot:
return mention return mention
@staticmethod @staticmethod
def get_question(self, text): def __get_question(self, text):
reply = False reply = False
keyword = '' keyword = ''
content = self.cleanhtml(self, text) content = self.__cleanhtml(self, text)
content = self.unescape(self, content) content = self.__unescape(self, content)
try: try:
@ -355,7 +356,7 @@ class Akkomabot:
#keyword_length = 8 #keyword_length = 8
keyword = question keyword = question
if keyword == self.register_str or keyword == self.unregister_str or keyword == self.stats_str: if keyword == self.register_str or keyword == self.unregister_str or keyword == self.stats_str or self.status_str:
keyword_length = len(keyword) keyword_length = len(keyword)
@ -370,14 +371,14 @@ class Akkomabot:
return (reply, question) return (reply, question)
@staticmethod @staticmethod
def cleanhtml(self, raw_html): def __cleanhtml(self, raw_html):
cleanr = re.compile('<.*?>') cleanr = re.compile('<.*?>')
cleantext = re.sub(cleanr, '', raw_html) cleantext = re.sub(cleanr, '', raw_html)
return cleantext return cleantext
@staticmethod @staticmethod
def unescape(self, s): def __unescape(self, s):
s = s.replace("&apos;", "'") s = s.replace("&apos;", "'")
return s return s

75
ejabberdapi.py
Veure arxiu

@ -28,20 +28,20 @@ class Ejabberd:
def __init__(self, api_base_url=None, local_vhost=None, admin_account=None, admin_pass=None): def __init__(self, api_base_url=None, local_vhost=None, admin_account=None, admin_pass=None):
self.ejabberd_config_path = "secrets/ejabberd_secrets.txt" self.__ejabberd_config_path = "secrets/ejabberd_secrets.txt"
is_setup = self.check_setup(self) is_setup = self.__check_setup(self)
if is_setup: if is_setup:
self.api_base_url = self.get_parameter("api_base_url", self.ejabberd_config_path) self.__api_base_url = self.__get_parameter("api_base_url", self.__ejabberd_config_path)
self.local_vhost = self.get_parameter("local_vhost", self.ejabberd_config_path) self.__local_vhost = self.__get_parameter("local_vhost", self.__ejabberd_config_path)
self.admin_account = self.get_parameter("admin_account", self.ejabberd_config_path) self.__admin_account = self.__get_parameter("admin_account", self.__ejabberd_config_path)
self.admin_pass = self.get_parameter("admin_pass", self.ejabberd_config_path) self.__admin_pass = self.__get_parameter("admin_pass", self.__ejabberd_config_path)
else: else:
self.api_base_url, self.local_vhost, self.admin_account, self.admin_pass = self.setup(self) self.__api_base_url, self.__local_vhost, self.__admin_account, self.__admin_pass = self.setup(self)
def generate_pass(self): def generate_pass(self):
@ -61,13 +61,13 @@ class Ejabberd:
def register(self, username, host, user_password): def register(self, username, host, user_password):
data = {'user':username, data = {'user':username,
'host':self.local_vhost, 'host':self.__local_vhost,
'password':user_password, 'password':user_password,
} }
API_ENDPOINT = self.api_base_url + '/api/register?' API_ENDPOINT = self.__api_base_url + '/api/register?'
response = requests.post(url = API_ENDPOINT, json = data, auth=(self.admin_account, self.admin_pass)) response = requests.post(url = API_ENDPOINT, json = data, auth=(self.__admin_account, self.__admin_pass))
is_registered = response.ok is_registered = response.ok
@ -87,19 +87,19 @@ class Ejabberd:
is_admin = False is_admin = False
if username+'@'+host == self.admin_account: if username+'@'+host == self.__admin_account:
is_admin = True is_admin = True
return (is_unregistered, is_admin) return (is_unregistered, is_admin)
data = {'user':username, data = {'user':username,
'host':self.local_vhost, 'host':self.__local_vhost,
} }
API_ENDPOINT = self.api_base_url + '/api/unregister?' API_ENDPOINT = self.__api_base_url + '/api/unregister?'
response = requests.post(url = API_ENDPOINT, json = data, auth=(self.admin_account, self.admin_pass)) response = requests.post(url = API_ENDPOINT, json = data, auth=(self.__admin_account, self.__admin_pass))
is_unregistered = response.ok is_unregistered = response.ok
@ -119,9 +119,9 @@ class Ejabberd:
"name": name "name": name
} }
API_ENDPOINT = self.api_base_url + '/api/stats?' API_ENDPOINT = self.__api_base_url + '/api/stats?'
response = requests.post(url = API_ENDPOINT, json = data, auth=(self.admin_account, self.admin_pass)) response = requests.post(url = API_ENDPOINT, json = data, auth=(self.__admin_account, self.__admin_pass))
result = response.json()['stat'] result = response.json()['stat']
@ -131,13 +131,26 @@ class Ejabberd:
return stats return stats
def status(self):
data = {
}
API_ENDPOINT = self.__api_base_url + '/api/status?'
response = requests.post(url = API_ENDPOINT, json = data, auth=(self.__admin_account, self.__admin_pass))
result = response.json()
return result
@staticmethod @staticmethod
def check_setup(self): def __check_setup(self):
is_setup = False is_setup = False
if not os.path.isfile(self.ejabberd_config_path): if not os.path.isfile(self.__ejabberd_config_path):
print(f"File {self.ejabberd_config_path} not found, running setup.") print(f"File {self.__ejabberd_config_path} not found, running setup.")
else: else:
is_setup = True is_setup = True
@ -149,23 +162,23 @@ class Ejabberd:
if not os.path.exists('secrets'): if not os.path.exists('secrets'):
os.makedirs('secrets') os.makedirs('secrets')
self.api_base_url = input("api_base_url, in ex. 'http://127.0.0.1:5280': ") self.__api_base_url = input("api_base_url, in ex. 'http://127.0.0.1:5280': ")
self.local_vhost = input("local_vhost, in ex. 'ejabberd.server': ") self.__local_vhost = input("local_vhost, in ex. 'ejabberd.server': ")
self.admin_account = input("admin_account, in ex. 'admin@ejabberd.server': ") self.__admin_account = input("admin_account, in ex. 'admin@ejabberd.server': ")
self.admin_pass = getpass.getpass("admin_pass, in ex. 'my_very_hard_secret_pass': ") self.__admin_pass = getpass.getpass("admin_pass, in ex. 'my_very_hard_secret_pass': ")
if not os.path.exists(self.ejabberd_config_path): if not os.path.exists(self.__ejabberd_config_path):
with open(self.ejabberd_config_path, 'w'): pass with open(self.__ejabberd_config_path, 'w'): pass
print(f"{self.ejabberd_config_path} created!") print(f"{self.__ejabberd_config_path} created!")
with open(self.ejabberd_config_path, 'a') as the_file: with open(self.__ejabberd_config_path, 'a') as the_file:
print("Writing ejabberd secrets parameters to " + self.ejabberd_config_path) print("Writing ejabberd secrets parameters to " + self.__ejabberd_config_path)
the_file.write(f'api_base_url: {self.api_base_url}\n'+f'local_vhost: {self.local_vhost}\n'+f'admin_account: {self.admin_account}\n'+f'admin_pass: {self.admin_pass}\n') the_file.write(f'api_base_url: {self.__api_base_url}\n'+f'local_vhost: {self.__local_vhost}\n'+f'admin_account: {self.__admin_account}\n'+f'admin_pass: {self.__admin_pass}\n')
return (self.api_base_url, self.local_vhost, self.admin_account, self.admin_pass) return (self.__api_base_url, self.__local_vhost, self.__admin_account, self.__admin_pass)
@staticmethod @staticmethod
def get_parameter(parameter, file_path ): def __get_parameter(parameter, file_path ):
with open( file_path ) as f: with open( file_path ) as f:
for line in f: for line in f:

14
xmpp.py
Veure arxiu

@ -9,8 +9,14 @@ if __name__ == '__main__':
ejabberd = Ejabberd() ejabberd = Ejabberd()
try:
notifications = bot.akkoma.notifications() notifications = bot.akkoma.notifications()
except:
pass
for notif in notifications: for notif in notifications:
if notif.type != 'mention': if notif.type != 'mention':
@ -75,6 +81,14 @@ if __name__ == '__main__':
bot.akkoma.status_post(post, in_reply_to_id=mention.status_id, visibility=mention.visibility) bot.akkoma.status_post(post, in_reply_to_id=mention.status_id, visibility=mention.visibility)
elif mention.question == bot.status_str:
status = ejabberd.status()
post = f'@{mention.acct}, {status}\n'
bot.akkoma.status_post(post, in_reply_to_id=mention.status_id, visibility=mention.visibility)
print(f"Dismissing notification id {mention.id}") print(f"Dismissing notification id {mention.id}")
bot.akkoma.notifications_dismiss(mention.id) bot.akkoma.notifications_dismiss(mention.id)