libervia-desktop-kivy: libervia/desktop_kivy/core/cagou

comparison libervia/desktop_kivy/core/cagou_main.py @ 514:d78728d7fd6a

plugin wid calls, core: implements WebRTC DataChannel file transfer: - Add a new "file" icon in call UI to send a file via WebRTC. - Handle new preflight mechanism, and WebRTC file transfer. - Native file chooser handling has been moved to new `core.file_chooser` module, and now supports "save" and "dir" modes (based on `plyer`). rel 442

author	Goffi <goffi@goffi.org>
date	Sat, 06 Apr 2024 13:37:27 +0200
parents	bbef1a413515
children	2ff26b4273df

comparison

equal deleted inserted replaced

-:0fdf3e59aaad
+:d78728d7fd6a
 import asyncio
 import os.path
 import glob
 import sys
 from pathlib import Path
+from typing import Callable
 from urllib import parse as urlparse
 from functools import partial
 from libervia.backend.core.i18n import _
+# `do_hack` msut be run before any Kivy import!
 from . import kivy_hack
 kivy_hack.do_hack()
 from .constants import Const as C
 from libervia.backend.core import log as logging
 from libervia.backend.core import exceptions
+from libervia.desktop_kivy.core.file_chooser import FileChooser
 from libervia.frontends.quick_frontend.quick_app import QuickApp
 from libervia.frontends.quick_frontend import quick_widgets
 from libervia.frontends.quick_frontend import quick_chat
 from libervia.frontends.quick_frontend import quick_utils
-from libervia.frontends.tools import jid
+from libervia.frontends.tools import aio, jid
 from libervia.backend.tools import utils as libervia_utils
 from libervia.backend.tools import config
 from libervia.backend.tools.common import data_format
 from libervia.backend.tools.common import dynamic_import
 from libervia.backend.tools.common import files_utils
 return self.switch_widget(None, wid)
 ## bridge handlers ##
+async def _on_webrtc_file(
+self,
+action_data: dict,
+action_id: str|None,
+profile: str
+) -> None:
+assert action_id is not None
+try:
+xmlui_data = action_data["xmlui"]
+except KeyError:
+raise exceptions.InternalError("Missing XMLUI in file action.")
+try:
+from_jid = jid.JID(action_data["from_jid"])
+except KeyError:
+raise exceptions.InternalError(
+f"Missing 'from_jid' key: {action_data!r}"
+)
+from libervia.frontends.tools.webrtc_file import WebRTCFileReceiver
+confirm_msg = WebRTCFileReceiver.format_confirm_msg(action_data, from_jid)
+file_accepted = action_data.get("file_accepted", False)
+if file_accepted:
+accepted = True
+else:
+accepted = await self.ask_confirmation(confirm_msg, _("File Request"))
+xmlui_data = {"answer": C.bool_const(accepted)}
+if accepted:
+file_data = action_data.get("file_data") or {}
+file_name = file_data.get("name", "received_file")
+dest_path_s = await FileChooser.a_open(
+mode="save",
+title=_("Please select the destination for file {file_name!r}.").format(
+file_name=file_name
+),
+# FIXME: It doesn't seem to be a way to specify destination file name,
+# ``path`` doesn't work this way, at least on Linux/KDE.
+default_path=f"./{file_name}"
+)
+if dest_path_s is None:
+accepted = False
+else:
+dest_path = Path(dest_path_s)
+try:
+session_id = action_data["session_id"]
+except KeyError:
+raise exceptions.InternalError("'session_id' is missing.")
+file_receiver = WebRTCFileReceiver(
+self.a_bridge,
+profile,
+on_close_cb=lambda: self.add_note(
+_("File Received"),
+_('The file "{file_name}" has been successfuly received.').format(
+file_name = file_data.get("name", "")
+)
+)
+)
+await file_receiver.receive_file_webrtc(
+from_jid,
+session_id,
+dest_path,
+file_data
+)
+await self.a_bridge.action_launch(
+action_id, data_format.serialise(xmlui_data), profile_key=profile
+)
+def action_manager(
+self,
+action_data: dict,
+callback: Callable|None = None,
+ui_show_cb: Callable|None = None,
+user_action: bool = True,
+action_id: str|None = None,
+progress_cb: Callable|None = None,
+progress_eb: Callable|None = None,
+profile: str = C.PROF_KEY_NONE
+) -> None:
+if (
+action_data.get("type") == C.META_TYPE_FILE
+and action_data.get("webrtc", False)
+):
+aio.run_async(self._on_webrtc_file(action_data, action_id, profile))
+else:
+super().action_manager(
+action_data, callback, ui_show_cb, user_action, action_id, progress_cb,
+progress_eb, profile
+)
 def otr_state_handler(self, state, dest_jid, profile):
 """OTR state has changed for on destinee"""
 # XXX: this method could be in QuickApp but it's here as
 #      it's only used by LiberviaDesktopKivy so far
 dest_jid = jid.JID(dest_jid)
 def callback():
 self.close_ui()
 cb(*args, **kwargs)
 return callback
-def show_dialog(self, message, title, type="info", answer_cb=None, answer_data=None):
+def show_dialog(
+self,
+message: str,
+title: str,
+type: str = "info",
+answer_cb: Callable|None = None,
+answer_data: dict|None = None
+):
+"""Show a dialog to the user.
+@param message: The main text of the dialog.
+@param title: Title of the dialog.
+@param type: Type of dialog (info, warning, error, yes/no).
+@param answer_cb: A callback that will be called when the user answers to the dialog.
+You can pass an asynchronous function as well.
+@param answer_data: Additional data for the dialog.
+"""
 if type in ('info', 'warning', 'error'):
 self.add_note(title, message, type)
 elif type == "yes/no":
 wid = dialog.ConfirmDialog(title=title, message=message,
 yes_cb=self._dialog_cb(answer_cb,
 )
 self.add_notif_widget(wid)
 return wid
 else:
 log.warning(_("unknown dialog type: {dialog_type}").format(dialog_type=type))
+async def a_show_dialog(
+self,
+message: str,
+title: str,
+type: str = "info",
+answer_data: dict|None = None
+) -> bool|None:
+"""Shows a dialog asynchronously and returns the user's response for 'yes/no' dialogs.
+This method wraps the synchronous ``show_dialog`` method to work in an
+asynchronous context.
+It is specifically useful for 'yes/no' type dialogs, returning True for 'yes' and
+False for 'no'. For other types, it returns None immediately after showing the
+dialog.
+See [show_dialog] for params.
+@return: True if the user clicked 'yes', False if 'no', and None for other dialog types.
+"""
+future = asyncio.Future()
+def answer_cb(answer: bool, data: dict):
+if not future.done():
+future.set_result(answer)
+if type == "yes/no":
+self.show_dialog(message, title, type, answer_cb, answer_data)
+return await future
+else:
+self.show_dialog(message, title, type)
+return None
+async def ask_confirmation(
+self,
+message: str,
+title: str,
+answer_data: dict|None = None
+) -> bool:
+ret = await self.a_show_dialog(message, title, "yes/no", answer_data)
+assert ret is bool
+return ret
 def share(self, media_type, data):
 share_wid = ShareWidget(media_type=media_type, data=data)
 try:
 self.show_extra_ui(share_wid)

Mercurial > libervia-desktop-kivy

comparison libervia/desktop_kivy/core/cagou_main.py @ 514:d78728d7fd6a