Liderahenk/Ahenk
1
0
Fork 0
mirror of https://github.com/Pardus-LiderAhenk/ahenk synced 2024-11-28 02:12:25 +03:00
Code Issues Projects Releases Packages Wiki Activity

conky and message written with dbus

Browse source
This commit is contained in:
3bru 2024-03-06 16:27:18 +03:00
parent 5901c8a185
commit 6614129f20
1 changed files with 57 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

92
src/plugins/conky/execute_xmessage.py
View file

@ -7,7 +7,7 @@ from base.model.enum.content_type import ContentType
import json import json
from base.plugin.abstract_plugin import AbstractPlugin from base.plugin.abstract_plugin import AbstractPlugin
import threading import threading
import dbus
class RunXMessageCommand(AbstractPlugin): class RunXMessageCommand(AbstractPlugin):
def __init__(self, data, context): def __init__(self, data, context):
@ -24,48 +24,70 @@ class RunXMessageCommand(AbstractPlugin):
# command for ltsp # command for ltsp
self.custom_message_command_ltsp = "su {0} -c 'export DISPLAY={1} && export XAUTHORITY=~{2}/.Xauthority && python3 /usr/share/ahenk/plugins/conky/ask.py \"LİDER AHENK\\\ BİLDİRİ \" \"{3}\" ' " self.custom_message_command_ltsp = "su {0} -c 'export DISPLAY={1} && export XAUTHORITY=~{2}/.Xauthority && python3 /usr/share/ahenk/plugins/conky/ask.py \"LİDER AHENK\\\ BİLDİRİ \" \"{3}\" ' "
# def execute_xmessage(self, message):
# users = self.Sessions.user_name();
# self.logger.debug('[XMessage] users : ' + str(users))
# for user in users:
# user_display = self.Sessions.display(user)
# user_ip = self.Sessions.userip(user)
# if user_display is None:
# self.logger.debug('[XMessage] executing for display none for user ' + str(user))
# else:
# self.logger.debug('[XMessage] user display ' + str(user_display) + ' user ' + str(user))
# if user_ip is None:
# t = threading.Thread(
# target=self.execute(self.custom_message_command.format(user, user_display, user, message)))
# t.start()
# else:
# # message format for ltsp
# self.logger.debug('user_ip: ' + str(user_ip) + ' user_display: ' + str(user_display))
# message_list = []
# message_parser = message.split(" ")
# self.logger.debug('running parser:--->> ' + str(message_parser))
# for msg in message_parser:
# message = '\\\ ' + str(msg)
# message_list.append(message)
# self.logger.debug('message_list:--->> ' + str(message_list))
# message = ''.join(str(x) for x in message_list)
# self.logger.debug('message: ' + str(message))
# t = threading.Thread(
# target=self.execute(self.custom_message_command_ltsp.format(user, user_display, user, message),
# ip=user_ip))
# t.start()
# self.context.create_response(code=self.message_code.TASK_PROCESSED.value,
# message='İşlem başarıyla gerçekleştirildi.',
# data=json.dumps({'Result': message}),
# content_type=ContentType.APPLICATION_JSON.value)
def execute_xmessage(self, message): def execute_xmessage(self, message):
bus = dbus.SessionBus()
users = self.Sessions.user_name(); service_name = "org.freedesktop.Notifications"
self.logger.debug('[XMessage] users : ' + str(users)) object_path = "/org/freedesktop/Notifications"
for user in users: obj = bus.get_object(service_name, object_path)
user_display = self.Sessions.display(user) interface = dbus.Interface(obj, dbus_interface="org.freedesktop.Notifications")
user_ip = self.Sessions.userip(user)
if user_display is None:
self.logger.debug('[XMessage] executing for display none for user ' + str(user))
else:
self.logger.debug('[XMessage] user display ' + str(user_display) + ' user ' + str(user))
if user_ip is None:
t = threading.Thread(
target=self.execute(self.custom_message_command.format(user, user_display, user, message)))
t.start()
else:
# message format for ltsp
self.logger.debug('user_ip: ' + str(user_ip) + ' user_display: ' + str(user_display))
message_list = []
message_parser = message.split(" ")
self.logger.debug('running parser:--->> ' + str(message_parser))
for msg in message_parser:
message = '\\\ ' + str(msg)
message_list.append(message)
self.logger.debug('message_list:--->> ' + str(message_list))
message = ''.join(str(x) for x in message_list)
self.logger.debug('message: ' + str(message))
t = threading.Thread(
target=self.execute(self.custom_message_command_ltsp.format(user, user_display, user, message),
ip=user_ip))
t.start()
app_name = "Liderahenk"
replaces_id = 0
app_icon = ""
summary = "LİDER AHENK BİLDİRİM "
body = message
notification_id = interface.Notify(app_name, replaces_id, app_icon, summary, body, [], {}, -1)
self.logger.info("Notification sent with ID: {0}".format(notification_id))
self.context.create_response(code=self.message_code.TASK_PROCESSED.value, self.context.create_response(code=self.message_code.TASK_PROCESSED.value,
message='İşlem başarıyla gerçekleştirildi.', message='İşlem başarıyla gerçekleştirildi.',
data=json.dumps({'Result': message}), data=json.dumps({'Result': body}),
content_type=ContentType.APPLICATION_JSON.value) content_type=ContentType.APPLICATION_JSON.value)
def execute_user_message(self, selected_user, message): def execute_user_message(self, selected_user, message):
users = self.Sessions.user_name(); users = self.Sessions.user_name();