mcvsoftware/twblue
1
0
Fork 0
mirror of https://github.com/MCV-Software/TWBlue.git synced 2026-03-06 09:27:33 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

Feat: Atproto integration. You can see home

Browse Source
This commit is contained in:
Jesús Pavón Abián
2025-08-30 22:48:00 +02:00
parent 8e999e67d4
commit 9124476ce0
9 changed files with 986 additions and 2587 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
src/atprotosocial.defaults Normal file
View File

@@ -0,0 +1,52 @@
[atprotosocial]
handle = string(default="")
app_password = string(default="")
did = string(default="")
session_string = string(default="")
user_name = string(default="")
[general]
relative_times = boolean(default=True)
max_posts_per_call = integer(default=40)
reverse_timelines = boolean(default=False)
persist_size = integer(default=0)
load_cache_in_memory = boolean(default=True)
show_screen_names = boolean(default=False)
hide_emojis = boolean(default=False)
buffer_order = list(default=list('home', 'notifications'))
disable_streaming = boolean(default=True)
[sound]
volume = float(default=1.0)
input_device = string(default="Default")
output_device = string(default="Default")
session_mute = boolean(default=False)
current_soundpack = string(default="FreakyBlue")
indicate_audio = boolean(default=True)
indicate_img = boolean(default=True)
[other_buffers]
timelines = list(default=list())
searches = list(default=list())
muted_buffers = list(default=list())
autoread_buffers = list(default=list(notifications))
[mysc]
spelling_language = string(default="")
save_followers_in_autocompletion_db = boolean(default=False)
save_friends_in_autocompletion_db = boolean(default=False)
ocr_language = string(default="")
[reporting]
braille_reporting = boolean(default=True)
speech_reporting = boolean(default=True)
[templates]
post = string(default="$display_name, $safe_text $date.")
person = string(default="$display_name (@$screen_name). $followers followers, $following following, $posts posts.")
notification = string(default="$display_name $text, $date")
[filters]
[user-aliases]

526
src/controller/atprotosocial/handler.py
View File

@@ -1,484 +1,74 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import TYPE_CHECKING, Any from typing import Any
import languageHandler # Ensure _() injection
fromapprove.controller.base import BaseHandler
fromapprove.sessions import SessionStoreInterface
if TYPE_CHECKING:
fromapprove.config import Config
fromapprove.sessions.atprotosocial.session import Session as ATProtoSocialSession
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class Handler(BaseHandler): class Handler:
SESSION_KIND = "atprotosocial" """Handler for Bluesky integration: creates minimal buffers."""
def __init__(self, session_store: SessionStoreInterface, config: Config) -> None: def __init__(self):
super().__init__(session_store, config) super().__init__()
self.main_controller = wx.GetApp().mainController # Get a reference to the mainController self.menus = dict(
# Define menu labels specific to ATProtoSocial compose="&Post",
self.menus = { )
"menubar_item": _("&Post"), # Top-level menu for posting actions self.item_menu = "&Post"
"compose": _("&New Post"), # New post/skeet
"share": _("&Repost"), # Equivalent of Boost/Retweet
"fav": _("&Like"), # Equivalent of Favorite
"unfav": _("&Unlike"),
# "dm": None, # Disable Direct Message if not applicable
# Add other menu items that need relabeling or enabling/disabling
}
# self.item_menu is another attribute used in mainController.update_menus
# It seems to be the label for the second main menu (originally "&Tweet")
self.item_menu = _("&Post") # Changes the top-level "Tweet" menu label to "Post"
def _get_session(self, user_id: str) -> ATProtoSocialSession: def create_buffers(self, session, createAccounts=True, controller=None):
session = self.session_store.get_session_by_user_id(user_id, self.SESSION_KIND) name = session.get_name()
if not session: controller.accounts.append(name)
# It's possible the session is still being created, especially during initial setup. if createAccounts:
# Try to get it from the global sessions store if it was just added. from pubsub import pub
from sessions import sessions as global_sessions pub.sendMessage("core.create_account", name=name, session_id=session.session_id, logged=True)
if user_id in global_sessions and global_sessions[user_id].KIND == self.SESSION_KIND: root_position = controller.view.search(name, name)
return global_sessions[user_id] # type: ignore[return-value] # Home timeline only for now
raise ValueError(f"No ATProtoSocial session found for user {user_id}") from pubsub import pub
return session # type: ignore[return-value] # We are checking kind pub.sendMessage(
"createBuffer",
def create_buffers(self, user_id: str) -> None: buffer_type="home_timeline",
"""Creates the default set of buffers for an ATProtoSocial session.""" session_type="atprotosocial",
session = self._get_session(user_id) buffer_title=_("Home"),
if not session: parent_tab=root_position,
logger.error(f"Cannot create buffers for ATProtoSocial user {user_id}: session not found.") start=True,
return kwargs=dict(parent=controller.view.nb, name="home_timeline", session=session)
)
logger.info(f"Creating default buffers for ATProtoSocial user {user_id} ({session.label})") # Following-only timeline (reverse-chronological)
pub.sendMessage(
# Home Timeline Buffer "createBuffer",
self.main_controller.add_buffer( buffer_type="following_timeline",
buffer_type="home_timeline", # Generic type, panel will adapt based on session kind session_type="atprotosocial",
user_id=user_id, buffer_title=_("Following"),
name=_("{label} Home").format(label=session.label), parent_tab=root_position,
session_kind=self.SESSION_KIND start=False,
kwargs=dict(parent=controller.view.nb, name="following_timeline", session=session)
) )
# Notifications Buffer def start_buffer(self, controller, buffer):
self.main_controller.add_buffer( """Start a newly created Bluesky buffer."""
buffer_type="notifications", # Generic type
user_id=user_id,
name=_("{label} Notifications").format(label=session.label),
session_kind=self.SESSION_KIND
)
# Own Posts (Profile) Buffer - using "user_posts" which is often generic
# self.main_controller.add_buffer(
# buffer_type="user_posts",
# user_id=user_id, # User whose posts to show (self in this case)
# target_user_id=session.util.get_own_did(), # Pass own DID as target
# name=_("{label} My Posts").format(label=session.label),
# session_kind=self.SESSION_KIND
# )
# Mentions buffer might be part of notifications or a separate stream/filter later.
# Ensure these buffers are shown if it's a new setup
# This part might be handled by mainController.add_buffer or session startup logic
# For now, just creating them. The UI should make them visible.
# --- Action Handlers (called by mainController based on menu interactions) ---
async def repost_item(self, session: ATProtoSocialSession, item_uri: str) -> dict[str, Any]:
"""Handles reposting an item."""
if not session.is_ready():
return {"status": "error", "message": _("Session not ready.")}
try: try:
success = await session.util.repost_post(item_uri) # Assuming repost_post in utils if hasattr(buffer, "start_stream"):
if success: buffer.start_stream(mandatory=True, play_sound=False)
return {"status": "success", "message": _("Post reposted successfully.")} # Enable periodic auto-refresh to simulate real-time updates
else: if hasattr(buffer, "enable_auto_refresh"):
return {"status": "error", "message": _("Failed to repost post.")} buffer.enable_auto_refresh()
except NotificationError as e: finally:
return {"status": "error", "message": e.message} # Ensure we won't try to start it again
except Exception as e: try:
logger.error(f"Error reposting item {item_uri}: {e}", exc_info=True) buffer.needs_init = False
return {"status": "error", "message": _("An unexpected error occurred while reposting.")} except Exception:
pass
async def like_item(self, session: ATProtoSocialSession, item_uri: str) -> dict[str, Any]:
"""Handles liking an item."""
if not session.is_ready():
return {"status": "error", "message": _("Session not ready.")}
try:
like_uri = await session.util.like_post(item_uri) # Assuming like_post in utils
if like_uri:
return {"status": "success", "message": _("Post liked successfully."), "like_uri": like_uri}
else:
return {"status": "error", "message": _("Failed to like post.")}
except NotificationError as e:
return {"status": "error", "message": e.message}
except Exception as e:
logger.error(f"Error liking item {item_uri}: {e}", exc_info=True)
return {"status": "error", "message": _("An unexpected error occurred while liking the post.")}
async def unlike_item(self, session: ATProtoSocialSession, like_uri: str) -> dict[str, Any]: # like_uri is the URI of the like record
"""Handles unliking an item."""
if not session.is_ready():
return {"status": "error", "message": _("Session not ready.")}
try:
# Unlike typically requires the URI of the like record itself, not the post.
# The UI or calling context needs to store this (e.g. from viewer_state of the post).
success = await session.util.delete_like(like_uri) # Assuming delete_like in utils
if success:
return {"status": "success", "message": _("Like removed successfully.")}
else:
return {"status": "error", "message": _("Failed to remove like.")}
except NotificationError as e:
return {"status": "error", "message": e.message}
except Exception as e:
logger.error(f"Error unliking item with like URI {like_uri}: {e}", exc_info=True)
return {"status": "error", "message": _("An unexpected error occurred while unliking.")}
async def handle_action(self, action_name: str, user_id: str, payload: dict[str, Any]) -> dict[str, Any] | None: async def handle_action(self, action_name: str, user_id: str, payload: dict[str, Any]) -> dict[str, Any] | None:
"""Handles actions specific to ATProtoSocial integration.""" logger.debug("handle_action stub: %s %s %s", action_name, user_id, payload)
logger.debug( return None
f"Handling ATProtoSocial action '{action_name}' for user {user_id} with payload: {payload}"
)
# session = self._get_session(user_id)
# TODO: Implement action handlers based on ATProtoSocial's capabilities async def handle_message_command(self, command: str, user_id: str, message_id: str, payload: dict[str, Any]) -> dict[str, Any] | None:
# Example: logger.debug("handle_message_command stub: %s %s %s %s", command, user_id, message_id, payload)
# if action_name == "get_profile_info": return None
# # profile_data = await session.util.get_profile_info(payload.get("handle"))
# # return {"profile": profile_data}
# elif action_name == "follow_user":
# # await session.util.follow_user(payload.get("user_id_to_follow"))
# # return {"status": "success", "message": "User followed"}
# else:
# logger.warning(f"Unknown ATProtoSocial action: {action_name}")
# return {"error": f"Unknown action: {action_name}"}
return None # Placeholder
async def handle_message_command( async def handle_user_command(self, command: str, user_id: str, target_user_id: str, payload: dict[str, Any]) -> dict[str, Any] | None:
self, command: str, user_id: str, message_id: str, payload: dict[str, Any] logger.debug("handle_user_command stub: %s %s %s %s", command, user_id, target_user_id, payload)
) -> dict[str, Any] | None: return None
"""Handles commands related to specific messages for ATProtoSocial."""
logger.debug(
f"Handling ATProtoSocial message command '{command}' for user {user_id}, message {message_id} with payload: {payload}"
)
# session = self._get_session(user_id)
# TODO: Implement message command handlers
# Example:
# if command == "get_post_details":
# # post_details = await session.util.get_post_by_id(message_id)
# # return {"details": post_details}
# elif command == "like_post":
# # await session.util.like_post(message_id)
# # return {"status": "success", "message": "Post liked"}
# else:
# logger.warning(f"Unknown ATProtoSocial message command: {command}")
# return {"error": f"Unknown message command: {command}"}
return None # Placeholder
fromapprove.translation import translate as _ # For user-facing messages
async def handle_user_command(
self, command: str, user_id: str, target_user_id: str, payload: dict[str, Any]
) -> dict[str, Any] | None:
"""Handles commands related to specific users for ATProtoSocial."""
logger.debug(
f"Handling ATProtoSocial user command '{command}' for user {user_id}, target user {target_user_id} with payload: {payload}"
)
session = self._get_session(user_id)
if not session.is_ready():
return {"status": "error", "message": _("ATProtoSocial session is not active or authenticated.")}
# target_user_id is expected to be the DID of the user to act upon.
if not target_user_id:
return {"status": "error", "message": _("Target user DID not provided.")}
success = False
message = _("Action could not be completed.") # Default error message
try:
if command == "follow_user":
success = await session.util.follow_user(target_user_id)
message = _("User followed successfully.") if success else _("Failed to follow user.")
elif command == "unfollow_user":
success = await session.util.unfollow_user(target_user_id)
message = _("User unfollowed successfully.") if success else _("Failed to unfollow user.")
elif command == "mute_user":
success = await session.util.mute_user(target_user_id)
message = _("User muted successfully.") if success else _("Failed to mute user.")
elif command == "unmute_user":
success = await session.util.unmute_user(target_user_id)
message = _("User unmuted successfully.") if success else _("Failed to unmute user.")
elif command == "block_user":
block_uri = await session.util.block_user(target_user_id) # Returns URI or None
success = bool(block_uri)
message = _("User blocked successfully.") if success else _("Failed to block user.")
elif command == "unblock_user":
success = await session.util.unblock_user(target_user_id)
message = _("User unblocked successfully.") if success else _("Failed to unblock user, or user was not blocked.")
else:
logger.warning(f"Unknown ATProtoSocial user command: {command}")
return {"status": "error", "message": _("Unknown action: {command}").format(command=command)}
return {"status": "success" if success else "error", "message": message}
except NotificationError as e: # Catch specific errors raised from utils
logger.error(f"ATProtoSocial user command '{command}' failed for target {target_user_id}: {e.message}")
return {"status": "error", "message": e.message}
except Exception as e:
logger.error(f"Unexpected error during ATProtoSocial user command '{command}' for target {target_user_id}: {e}", exc_info=True)
return {"status": "error", "message": _("An unexpected server error occurred.")}
# --- UI Related Action Handlers (called by mainController) ---
async def user_details(self, buffer: Any) -> None: # buffer is typically a timeline panel
"""Shows user profile details for the selected user in the buffer."""
session = buffer.session
if not session or session.KIND != self.SESSION_KIND or not session.is_ready():
output.speak(_("Active ATProtoSocial session not found or not ready."), True)
return
user_ident = None
if hasattr(buffer, "get_selected_item_author_details"): # Method in panel to get author of selected post
author_details = buffer.get_selected_item_author_details()
if author_details and isinstance(author_details, dict):
user_ident = author_details.get("did") or author_details.get("handle")
if not user_ident:
# Fallback or if no item selected, prompt for user
# For now, just inform user if no selection. A dialog prompt could be added.
output.speak(_("Please select an item or user to view details."), True)
# TODO: Add wx.TextEntryDialog to ask for user handle/DID if none selected
return
try:
profile_data = await session.util.get_user_profile(user_ident)
if profile_data:
# Example: from src.wxUI.dialogs.mastodon.showUserProfile import UserProfileDialog
# For ATProtoSocial, we use the new dialog:
from wxUI.dialogs.atprotosocial.showUserProfile import ShowUserProfileDialog
# Ensure main_controller.view is the correct parent (main frame)
dialog = ShowUserProfileDialog(parent=self.main_controller.view, session=session, user_identifier=user_ident)
dialog.ShowModal() # Show as modal dialog
dialog.Destroy()
else:
output.speak(_("Could not fetch profile for {user_ident}.").format(user_ident=user_ident), True)
except Exception as e:
logger.error(f"Error fetching/displaying profile for {user_ident}: {e}", exc_info=True)
output.speak(_("Error displaying profile: {error}").format(error=str(e)), True)
async def open_user_timeline(self, main_controller: Any, session: ATProtoSocialSession, user_payload: Any | None) -> None:
"""Opens a new buffer for a specific user's posts."""
user_ident = None
if isinstance(user_payload, dict): # Assuming user_payload is a dict from get_selected_item_author_details
user_ident = user_payload.get("did") or user_payload.get("handle")
elif isinstance(user_payload, str): # Direct DID or Handle string
user_ident = user_payload
if not user_ident: # Prompt if not found
dialog = wx.TextEntryDialog(main_controller.view, _("Enter user DID or handle:"), _("View User Timeline"))
if dialog.ShowModal() == wx.ID_OK:
user_ident = dialog.GetValue()
dialog.Destroy()
if not user_ident:
return
# Fetch profile to get canonical handle/DID for buffer name, and to ensure user exists
try:
profile = await session.util.get_user_profile(user_ident)
if not profile:
output.speak(_("User {user_ident} not found.").format(user_ident=user_ident), True)
return
buffer_name = _("{user_handle}'s Posts").format(user_handle=profile.handle)
buffer_id = f"atp_user_feed_{profile.did}" # Unique ID for the buffer
# Check if buffer already exists
# existing_buffer = main_controller.search_buffer_by_id_or_properties(id=buffer_id) # Hypothetical method
# For now, assume it might create duplicates if not handled by add_buffer logic
main_controller.add_buffer(
buffer_type="user_timeline", # This type will need a corresponding panel
user_id=session.uid, # The session user_id
name=buffer_name,
session_kind=self.SESSION_KIND,
target_user_did=profile.did, # Store target DID for the panel to use
target_user_handle=profile.handle
)
except Exception as e:
logger.error(f"Error opening user timeline for {user_ident}: {e}", exc_info=True)
output.speak(_("Failed to open user timeline: {error}").format(error=str(e)), True)
async def open_followers_timeline(self, main_controller: Any, session: ATProtoSocialSession, user_payload: Any | None) -> None:
"""Opens a new buffer for a user's followers."""
user_ident = None
if isinstance(user_payload, dict):
user_ident = user_payload.get("did") or user_payload.get("handle")
elif isinstance(user_payload, str):
user_ident = user_payload
if not user_ident:
dialog = wx.TextEntryDialog(main_controller.view, _("Enter user DID or handle to view followers:"), _("View Followers"))
if dialog.ShowModal() == wx.ID_OK:
user_ident = dialog.GetValue()
dialog.Destroy()
if not user_ident:
return
try:
profile = await session.util.get_user_profile(user_ident) # Ensure user exists, get DID
if not profile:
output.speak(_("User {user_ident} not found.").format(user_ident=user_ident), True)
return
buffer_name = _("Followers of {user_handle}").format(user_handle=profile.handle)
main_controller.add_buffer(
buffer_type="user_list_followers", # Needs specific panel type
user_id=session.uid,
name=buffer_name,
session_kind=self.SESSION_KIND,
target_user_did=profile.did
)
except Exception as e:
logger.error(f"Error opening followers list for {user_ident}: {e}", exc_info=True)
output.speak(_("Failed to open followers list: {error}").format(error=str(e)), True)
async def open_following_timeline(self, main_controller: Any, session: ATProtoSocialSession, user_payload: Any | None) -> None:
"""Opens a new buffer for users a user is following."""
user_ident = None
if isinstance(user_payload, dict):
user_ident = user_payload.get("did") or user_payload.get("handle")
elif isinstance(user_payload, str):
user_ident = user_payload
if not user_ident:
dialog = wx.TextEntryDialog(main_controller.view, _("Enter user DID or handle to view their following list:"), _("View Following"))
if dialog.ShowModal() == wx.ID_OK:
user_ident = dialog.GetValue()
dialog.Destroy()
if not user_ident:
return
try:
profile = await session.util.get_user_profile(user_ident) # Ensure user exists, get DID
if not profile:
output.speak(_("User {user_ident} not found.").format(user_ident=user_ident), True)
return
buffer_name = _("Following by {user_handle}").format(user_handle=profile.handle)
main_controller.add_buffer(
buffer_type="user_list_following", # Needs specific panel type
user_id=session.uid,
name=buffer_name,
session_kind=self.SESSION_KIND,
target_user_did=profile.did
)
except Exception as e:
logger.error(f"Error opening following list for {user_ident}: {e}", exc_info=True)
output.speak(_("Failed to open following list: {error}").format(error=str(e)), True)
async def get_settings_inputs(self, user_id: str | None = None) -> list[dict[str, Any]]:
"""Returns settings inputs for ATProtoSocial, potentially user-specific."""
# This typically delegates to the Session class's method
fromapprove.sessions.atprotosocial.session import Session as ATProtoSocialSession
current_config = {}
if user_id:
# Fetch existing config for the user if available to pre-fill values
# This part depends on how config is stored and accessed.
# For example, if Session class has a method to get its current config:
try:
session = self._get_session(user_id)
current_config = session.get_configurable_values_for_user(user_id)
except ValueError: # No session means no specific config yet for this user
pass
except Exception as e:
logger.error(f"Error fetching current ATProtoSocial config for user {user_id}: {e}")
return ATProtoSocialSession.get_settings_inputs(user_id=user_id, current_config=current_config)
async def update_settings(self, user_id: str, settings_data: dict[str, Any]) -> dict[str, Any]:
"""Updates settings for ATProtoSocial for a given user."""
logger.info(f"Updating ATProtoSocial settings for user {user_id}")
# This is a simplified example. In a real scenario, you'd validate `settings_data`
# and then update the configuration, possibly re-initializing the session or
# informing it of the changes.
# config_manager = self.config.sessions.atprotosocial[user_id]
# for key, value in settings_data.items():
# if hasattr(config_manager, key):
# await config_manager[key].set(value)
# else:
# logger.warning(f"Attempted to set unknown ATProtoSocial setting '{key}' for user {user_id}")
# # Optionally, re-initialize or notify the session if it's active
# try:
# session = self._get_session(user_id)
# await session.stop() # Stop if it might be using old settings
# # Re-fetch config for the session or update it directly
# # session.api_base_url = settings_data.get("api_base_url", session.api_base_url)
# # session.access_token = settings_data.get("access_token", session.access_token)
# if session.active: # Or based on some logic if it should auto-restart
# await session.start()
# logger.info(f"Successfully updated and re-initialized ATProtoSocial session for user {user_id}")
# except ValueError:
# logger.info(f"ATProtoSocial session for user {user_id} not found or not active, settings saved but session not restarted.")
# except Exception as e:
# logger.error(f"Error re-initializing ATProtoSocial session for user {user_id} after settings update: {e}")
# return {"status": "error", "message": f"Settings saved, but failed to restart session: {e}"}
# For now, just a placeholder response
return {"status": "success", "message": "ATProtoSocial settings updated (implementation pending)."}
@classmethod
def get_auth_inputs(cls, user_id: str | None = None) -> list[dict[str, Any]]:
fromapprove.sessions.atprotosocial.session import Session as ATProtoSocialSession
# current_config = {} # fetch if needed
return ATProtoSocialSession.get_auth_inputs(user_id=user_id) # current_config=current_config
@classmethod
async def test_connection(cls, settings: dict[str, Any]) -> tuple[bool, str]:
fromapprove.sessions.atprotosocial.session import Session as ATProtoSocialSession
return await ATProtoSocialSession.test_connection(settings)
@classmethod
def get_user_actions(cls) -> list[dict[str, Any]]:
fromapprove.sessions.atprotosocial.session import Session as ATProtoSocialSession
return ATProtoSocialSession.get_user_actions()
@classmethod
def get_user_list_actions(cls) -> list[dict[str, Any]]:
fromapprove.sessions.atprotosocial.session import Session as ATProtoSocialSession
return ATProtoSocialSession.get_user_list_actions()
@classmethod
def get_config_description(cls) -> str | None:
fromapprove.sessions.atprotosocial.session import Session as ATProtoSocialSession
return ATProtoSocialSession.get_config_description()
@classmethod
def get_auth_type(cls) -> str:
fromapprove.sessions.atprotosocial.session import Session as ATProtoSocialSession
return ATProtoSocialSession.get_auth_type()
@classmethod
def get_logo_path(cls) -> str | None:
fromapprove.sessions.atprotosocial.session import Session as ATProtoSocialSession
return ATProtoSocialSession.get_logo_path()
@classmethod
def get_session_kind(cls) -> str:
return cls.SESSION_KIND
@classmethod
def get_dependencies(cls) -> list[str]:
fromapprove.sessions.atprotosocial.session import Session as ATProtoSocialSession
return ATProtoSocialSession.get_dependencies()

15
src/controller/atprotosocial/messages.py
View File

@@ -1,13 +1,9 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import TYPE_CHECKING, Any from typing import Any
# fromapprove.controller.mastodon import messages as mastodon_messages # Example, if adapting # Translation function is provided globally by TWBlue's language handler (_)
fromapprove.translation import translate as _
if TYPE_CHECKING:
fromapprove.sessions.atprotosocial.session import Session as ATProtoSocialSession # Adjusted import
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -19,14 +15,17 @@ logger = logging.getLogger(__name__)
# Example: If ATProtoSocial develops a standard for "cards" or interactive messages, # Example: If ATProtoSocial develops a standard for "cards" or interactive messages,
# functions to create those would go here. For now, we can imagine placeholders. # functions to create those would go here. For now, we can imagine placeholders.
def format_welcome_message(session: ATProtoSocialSession) -> dict[str, Any]: def format_welcome_message(session: Any) -> dict[str, Any]:
""" """
Generates a welcome message for a new ATProtoSocial session. Generates a welcome message for a new ATProtoSocial session.
This is just a placeholder and example. This is just a placeholder and example.
""" """
# user_profile = session.util.get_own_profile_info() # Assuming this method exists and is async or cached # user_profile = session.util.get_own_profile_info() # Assuming this method exists and is async or cached
# handle = user_profile.get("handle", _("your ATProtoSocial account")) if user_profile else _("your ATProtoSocial account") # handle = user_profile.get("handle", _("your ATProtoSocial account")) if user_profile else _("your ATProtoSocial account")
handle = session.util.get_own_username() or _("your ATProtoSocial account") # Expect session to expose username via db/settings
handle = (getattr(session, "db", {}).get("user_name")
or getattr(getattr(session, "settings", {}), "get", lambda *_: {})("atprotosocial").get("handle")
or _("your Bluesky account"))
return { return {

416
src/controller/mainController.py
View File

@@ -94,6 +94,17 @@ class Controller(object):
[results.append(self.search_buffer(i.name, i.account)) for i in buffers if i.account == account and (i.type != "account")] [results.append(self.search_buffer(i.name, i.account)) for i in buffers if i.account == account and (i.type != "account")]
return results return results
def get_handler(self, type):
"""Return the controller handler for a given session type."""
try:
if type == "mastodon":
return MastodonHandler.Handler()
if type == "atprotosocial":
return ATProtoSocialHandler.Handler()
except Exception:
log.exception("Error creating handler for type %s", type)
return None
def bind_other_events(self): def bind_other_events(self):
""" Binds the local application events with their functions.""" """ Binds the local application events with their functions."""
log.debug("Binding other application events...") log.debug("Binding other application events...")
@@ -193,29 +204,12 @@ class Controller(object):
def get_handler(self, type): def get_handler(self, type):
handler = self.handlers.get(type) handler = self.handlers.get(type)
if handler == None: if handler is None:
if type == "mastodon": if type == "mastodon":
handler = MastodonHandler.Handler() handler = MastodonHandler.Handler()
elif type == "atprotosocial": # Added case for atprotosocial elif type == "atprotosocial":
# Assuming session_store and config_proxy are accessible or passed if needed by Handler constructor handler = ATProtoSocialHandler.Handler()
# For now, let's assume constructor is similar or adapted to not require them, self.handlers[type] = handler
# or that they can be accessed via self if mainController has them.
# Based on atprotosocial.Handler, it needs session_store and config.
# mainController doesn't seem to store these directly for passing.
# This might indicate Handler init needs to be simplified or these need to be plumbed.
# For now, proceeding with a simplified instantiation, assuming it can get what it needs
# or its __init__ will be adapted.
# A common pattern is self.session_store and self.config from a base controller class if mainController inherits one.
# Let's assume for now they are not strictly needed for just getting menu labels or simple actions.
# This part might need refinement based on Handler's actual dependencies for menu updates.
# Looking at atprotosocial/handler.py, it takes session_store and config.
# mainController itself doesn't seem to have these as direct attributes to pass on.
# This implies a potential refactor need or that these handlers are simpler than thought for menu updates.
# For now, let's assume a simplified handler for menu updates or that it gets these elsewhere.
# This needs to be compatible with how MastodonHandler is instantiated and used.
# MastodonHandler() is called without params here.
handler = ATProtoSocialHandler.Handler(session_store=sessions.sessions, config=config.app) # Adjusted: Pass global sessions and config
self.handlers[type]=handler
return handler return handler
def __init__(self): def __init__(self):
@@ -256,14 +250,24 @@ class Controller(object):
for i in sessions.sessions: for i in sessions.sessions:
log.debug("Working on session %s" % (i,)) log.debug("Working on session %s" % (i,))
if sessions.sessions[i].is_logged == False: if sessions.sessions[i].is_logged == False:
self.create_ignored_session_buffer(sessions.sessions[i]) # Try auto-login for ATProtoSocial sessions if credentials exist
continue try:
# Valid types currently are mastodon (Work in progress) if getattr(sessions.sessions[i], "type", None) == "atprotosocial":
# More can be added later. sessions.sessions[i].login()
valid_session_types = ["mastodon"] except Exception:
log.exception("Auto-login attempt failed for session %s", i)
if sessions.sessions[i].is_logged == False:
self.create_ignored_session_buffer(sessions.sessions[i])
continue
# Supported session types
valid_session_types = ["mastodon", "atprotosocial"]
if sessions.sessions[i].type in valid_session_types: if sessions.sessions[i].type in valid_session_types:
handler = self.get_handler(type=sessions.sessions[i].type) try:
handler.create_buffers(sessions.sessions[i], controller=self) handler = self.get_handler(type=sessions.sessions[i].type)
if handler is not None:
handler.create_buffers(sessions.sessions[i], controller=self)
except Exception:
log.exception("Error creating buffers for session %s (%s)", i, sessions.sessions[i].type)
log.debug("Setting updates to buffers every %d seconds..." % (60*config.app["app-settings"]["update_period"],)) log.debug("Setting updates to buffers every %d seconds..." % (60*config.app["app-settings"]["update_period"],))
self.update_buffers_function = RepeatingTimer(60*config.app["app-settings"]["update_period"], self.update_buffers) self.update_buffers_function = RepeatingTimer(60*config.app["app-settings"]["update_period"], self.update_buffers)
self.update_buffers_function.start() self.update_buffers_function.start()
@@ -294,7 +298,10 @@ class Controller(object):
session.login() session.login()
handler = self.get_handler(type=session.type) handler = self.get_handler(type=session.type)
if handler != None and hasattr(handler, "create_buffers"): if handler != None and hasattr(handler, "create_buffers"):
handler.create_buffers(session=session, controller=self, createAccounts=False) try:
handler.create_buffers(session=session, controller=self, createAccounts=False)
except Exception:
log.exception("Error creating buffers after login for session %s (%s)", session.session_id, session.type)
self.start_buffers(session) self.start_buffers(session)
if hasattr(session, "start_streaming"): if hasattr(session, "start_streaming"):
session.start_streaming() session.start_streaming()
@@ -308,102 +315,103 @@ class Controller(object):
self.view.add_buffer(account.buffer , name=name) self.view.add_buffer(account.buffer , name=name)
def create_buffer(self, buffer_type="baseBuffer", session_type="twitter", buffer_title="", parent_tab=None, start=False, kwargs={}): def create_buffer(self, buffer_type="baseBuffer", session_type="twitter", buffer_title="", parent_tab=None, start=False, kwargs={}):
# Copy kwargs to avoid mutating a shared dict across calls
if not isinstance(kwargs, dict):
kwargs = {}
else:
kwargs = dict(kwargs)
log.debug("Creating buffer of type {0} with parent_tab of {2} arguments {1}".format(buffer_type, kwargs, parent_tab)) log.debug("Creating buffer of type {0} with parent_tab of {2} arguments {1}".format(buffer_type, kwargs, parent_tab))
if kwargs.get("parent") == None: if kwargs.get("parent") == None:
kwargs["parent"] = self.view.nb kwargs["parent"] = self.view.nb
if not hasattr(buffers, session_type) and session_type != "atprotosocial": # Allow atprotosocial to be handled separately if not hasattr(buffers, session_type) and session_type != "atprotosocial": # Allow atprotosocial to be handled separately
raise AttributeError("Session type %s does not exist yet." % (session_type)) raise AttributeError("Session type %s does not exist yet." % (session_type))
buffer_panel_class = None try:
if session_type == "atprotosocial": buffer_panel_class = None
from wxUI.buffers.atprotosocial import panels as ATProtoSocialPanels # Import new panels if session_type == "atprotosocial":
if buffer_type == "home_timeline": from wxUI.buffers.atprotosocial import panels as ATProtoSocialPanels # Import new panels
buffer_panel_class = ATProtoSocialPanels.ATProtoSocialHomeTimelinePanel if buffer_type == "home_timeline":
# kwargs for HomeTimelinePanel: parent, name, session buffer_panel_class = ATProtoSocialPanels.ATProtoSocialHomeTimelinePanel
# 'name' is buffer_title, 'parent' is self.view.nb # kwargs for HomeTimelinePanel: parent, name, session
# 'session' needs to be fetched based on user_id in kwargs # 'name' is buffer_title, 'parent' is self.view.nb
if "user_id" in kwargs and "session" not in kwargs: # Ensure session is passed # 'session' needs to be fetched based on user_id in kwargs
kwargs["session"] = sessions.sessions.get(kwargs["user_id"]) if "user_id" in kwargs and "session" not in kwargs: # Ensure session is passed
if "name" not in kwargs: kwargs["name"] = buffer_title kwargs["session"] = sessions.sessions.get(kwargs["user_id"])
# Clean unsupported kwarg for panel ctor
if "user_id" in kwargs:
kwargs.pop("user_id", None)
if "name" not in kwargs: kwargs["name"] = buffer_title
elif buffer_type == "user_timeline": elif buffer_type == "user_timeline":
buffer_panel_class = ATProtoSocialPanels.ATProtoSocialUserTimelinePanel buffer_panel_class = ATProtoSocialPanels.ATProtoSocialUserTimelinePanel
# kwargs for UserTimelinePanel: parent, name, session, target_user_did, target_user_handle # kwargs for UserTimelinePanel: parent, name, session, target_user_did, target_user_handle
if "user_id" in kwargs and "session" not in kwargs: if "user_id" in kwargs and "session" not in kwargs:
kwargs["session"] = sessions.sessions.get(kwargs["user_id"]) kwargs["session"] = sessions.sessions.get(kwargs["user_id"])
if "name" not in kwargs: kwargs["name"] = buffer_title kwargs.pop("user_id", None)
# target_user_did and target_user_handle must be in kwargs from atprotosocial.Handler if "name" not in kwargs: kwargs["name"] = buffer_title
# target_user_did and target_user_handle must be in kwargs from atprotosocial.Handler
elif buffer_type == "notifications": elif buffer_type == "notifications":
buffer_panel_class = ATProtoSocialPanels.ATProtoSocialNotificationPanel buffer_panel_class = ATProtoSocialPanels.ATProtoSocialNotificationPanel
if "user_id" in kwargs and "session" not in kwargs: if "user_id" in kwargs and "session" not in kwargs:
kwargs["session"] = sessions.sessions.get(kwargs["user_id"]) kwargs["session"] = sessions.sessions.get(kwargs["user_id"])
if "name" not in kwargs: kwargs["name"] = buffer_title kwargs.pop("user_id", None)
# target_user_did and target_user_handle must be in kwargs from atprotosocial.Handler if "name" not in kwargs: kwargs["name"] = buffer_title
# target_user_did and target_user_handle must be in kwargs from atprotosocial.Handler
elif buffer_type == "notifications": elif buffer_type == "notifications":
buffer_panel_class = ATProtoSocialPanels.ATProtoSocialNotificationPanel buffer_panel_class = ATProtoSocialPanels.ATProtoSocialNotificationPanel
if "user_id" in kwargs and "session" not in kwargs: if "user_id" in kwargs and "session" not in kwargs:
kwargs["session"] = sessions.sessions.get(kwargs["user_id"]) kwargs["session"] = sessions.sessions.get(kwargs["user_id"])
if "name" not in kwargs: kwargs["name"] = buffer_title kwargs.pop("user_id", None)
elif buffer_type == "user_list_followers" or buffer_type == "user_list_following": if "name" not in kwargs: kwargs["name"] = buffer_title
buffer_panel_class = ATProtoSocialPanels.ATProtoSocialUserListPanel elif buffer_type == "user_list_followers" or buffer_type == "user_list_following":
if "user_id" in kwargs and "session" not in kwargs: buffer_panel_class = ATProtoSocialPanels.ATProtoSocialUserListPanel
kwargs["session"] = sessions.sessions.get(kwargs["user_id"]) elif buffer_type == "following_timeline":
if "name" not in kwargs: kwargs["name"] = buffer_title buffer_panel_class = ATProtoSocialPanels.ATProtoSocialFollowingTimelinePanel
# Ensure 'list_type', 'target_user_did', 'target_user_handle' are in kwargs # Clean stray keys that this panel doesn't accept
if "list_type" not in kwargs: # Set based on buffer_type kwargs.pop("user_id", None)
kwargs["list_type"] = buffer_type.split('_')[-1] # followers or following kwargs.pop("list_type", None)
else: if "name" not in kwargs: kwargs["name"] = buffer_title
log.warning(f"Unsupported ATProtoSocial buffer type: {buffer_type}. Falling back to generic.")
# Fallback to trying to find it in generic buffers or error
# For now, let it try the old way if not found above
available_buffers = getattr(buffers, "base", None) # Or some generic panel module
if available_buffers and hasattr(available_buffers, buffer_type):
buffer_panel_class = getattr(available_buffers, buffer_type)
elif available_buffers and hasattr(available_buffers, "TimelinePanel"): # Example generic
buffer_panel_class = getattr(available_buffers, "TimelinePanel")
else: else:
raise AttributeError(f"ATProtoSocial buffer type {buffer_type} not found in atprotosocial.panels or base panels.") log.warning(f"Unsupported ATProtoSocial buffer type: {buffer_type}. Falling back to generic.")
else: # Existing logic for other session types # Fallback to trying to find it in generic buffers or error
available_buffers = getattr(buffers, session_type) available_buffers = getattr(buffers, "base", None) # Or some generic panel module
if not hasattr(available_buffers, buffer_type): if available_buffers and hasattr(available_buffers, buffer_type):
raise AttributeError("Specified buffer type does not exist: %s" % (buffer_type,)) buffer_panel_class = getattr(available_buffers, buffer_type)
buffer_panel_class = getattr(available_buffers, buffer_type) elif available_buffers and hasattr(available_buffers, "TimelinePanel"): # Example generic
buffer_panel_class = getattr(available_buffers, "TimelinePanel")
else:
raise AttributeError(f"ATProtoSocial buffer type {buffer_type} not found in atprotosocial.panels or base panels.")
else: # Existing logic for other session types
available_buffers = getattr(buffers, session_type)
if not hasattr(available_buffers, buffer_type):
raise AttributeError("Specified buffer type does not exist: %s" % (buffer_type,))
buffer_panel_class = getattr(available_buffers, buffer_type)
# Instantiate the panel # Instantiate the panel
# Ensure 'parent' kwarg is correctly set if not already # Ensure 'parent' kwarg is correctly set if not already
if "parent" not in kwargs: if "parent" not in kwargs:
kwargs["parent"] = self.view.nb # self.view.nb is the wx.Treebook kwargs["parent"] = self.view.nb # self.view.nb is the wx.Treebook
# Clean kwargs that are not meant for panel __init__ directly (like user_id, session_kind if used by add_buffer but not panel) buffer = buffer_panel_class(**kwargs) # This is the wx.Panel instance
# This depends on what add_buffer and panel constructors expect.
# For now, assume kwargs are mostly for the panel.
buffer = buffer_panel_class(**kwargs) # This is the wx.Panel instance if start:
if start: # 'start' usually means load initial data for the buffer
# The panels themselves should handle initial data loading in their __init__ or a separate load method
# For ATProtoSocial panels, this is wx.CallAfter(asyncio.create_task, self.load_initial_posts())
# The old `start_stream` logic might not apply directly.
if hasattr(buffer, "load_initial_data_async"): # A new conventional async method
wx.CallAfter(asyncio.create_task, buffer.load_initial_data_async())
elif hasattr(buffer, "start_stream"): # Legacy way
if kwargs.get("function") == "user_timeline": # This old check might be obsolete
try: try:
buffer.start_stream(play_sound=False) if hasattr(buffer, "start_stream"):
buffer.start_stream(mandatory=True, play_sound=False)
except ValueError: except ValueError:
commonMessageDialogs.unauthorized() commonMessageDialogs.unauthorized()
return return
self.buffers.append(buffer)
if parent_tab == None:
log.debug("Appending buffer {}...".format(buffer,))
self.view.add_buffer(buffer.buffer, buffer_title)
else: else:
call_threaded(buffer.start_stream) self.view.insert_buffer(buffer.buffer, buffer_title, parent_tab)
self.buffers.append(buffer) log.debug("Inserting buffer {0} into control {1}".format(buffer, parent_tab))
if parent_tab == None: except Exception:
log.debug("Appending buffer {}...".format(buffer,)) log.exception("Error creating buffer '%s' for session_type '%s'", buffer_type, session_type)
self.view.add_buffer(buffer.buffer, buffer_title)
else:
self.view.insert_buffer(buffer.buffer, buffer_title, parent_tab)
log.debug("Inserting buffer {0} into control {1}".format(buffer, parent_tab))
def set_buffer_positions(self, session): def set_buffer_positions(self, session):
"Sets positions for buffers if values exist in the database." "Sets positions for buffers if values exist in the database."
@@ -589,6 +597,53 @@ class Controller(object):
return return
session = buffer.session session = buffer.session
# Compose for Bluesky (ATProto): dialog with attachments/CW/language
if getattr(session, "type", "") == "atprotosocial":
# In invisible interface, prefer a quick, minimal compose to avoid complex UI
if self.showing == False:
# Parent=None so it shows even if main window is hidden
dlg = wx.TextEntryDialog(None, _("Write your post:"), _("Compose"))
if dlg.ShowModal() == wx.ID_OK:
text = dlg.GetValue().strip()
dlg.Destroy()
if not text:
return
try:
uri = session.send_message(text)
if uri:
output.speak(_("Post sent successfully!"), True)
else:
output.speak(_("Failed to send post."), True)
except Exception:
log.exception("Error sending Bluesky post from invisible compose")
output.speak(_("An error occurred while posting to Bluesky."), True)
else:
dlg.Destroy()
return
from wxUI.dialogs.atprotosocial.postDialogs import Post as ATPostDialog
dlg = ATPostDialog()
if dlg.ShowModal() == wx.ID_OK:
text, files, cw_text, langs = dlg.get_payload()
dlg.Destroy()
if not text and not files:
return
try:
uri = session.send_message(text, files=files, cw_text=cw_text, is_sensitive=bool(cw_text), languages=langs)
if uri:
output.speak(_("Post sent successfully!"), True)
try:
if hasattr(buffer, "start_stream"):
buffer.start_stream(mandatory=False, play_sound=False)
except Exception:
pass
else:
output.speak(_("Failed to send post."), True)
except Exception:
log.exception("Error sending Bluesky post from compose dialog")
output.speak(_("An error occurred while posting to Bluesky."), True)
else:
dlg.Destroy()
return
# For a new post, reply_to_uri and quote_uri are None. # For a new post, reply_to_uri and quote_uri are None.
# Import the new dialog # Import the new dialog
from wxUI.dialogs.composeDialog import ComposeDialog from wxUI.dialogs.composeDialog import ComposeDialog
@@ -644,28 +699,67 @@ class Controller(object):
def post_reply(self, *args, **kwargs): def post_reply(self, *args, **kwargs):
buffer = self.get_current_buffer() # This is the panel instance buffer = self.get_current_buffer()
if not buffer or not buffer.session: if not buffer or not buffer.session:
output.speak(_("No active session to reply."), True) output.speak(_("No active session to reply."), True)
return return
selected_item_uri = buffer.get_selected_item_id() # URI of the post to reply to selected_item_uri = None
if hasattr(buffer, "get_selected_item_id"):
selected_item_uri = buffer.get_selected_item_id()
if not selected_item_uri: if not selected_item_uri:
output.speak(_("No item selected to reply to."), True) output.speak(_("No item selected to reply to."), True)
return return
# Optionally, get initial text for reply (e.g., mentioning users) session = buffer.session
# initial_text = buffer.session.compose_panel.get_reply_text(selected_item_uri, author_handle_of_selected_post) if getattr(session, "type", "") == "atprotosocial":
# For now, simple empty initial text for reply. if self.showing == False:
initial_text = "" dlg = wx.TextEntryDialog(None, _("Write your reply:"), _("Reply"))
# Get author handle for reply text (if needed by compose_panel.get_reply_text) if dlg.ShowModal() == wx.ID_OK:
# author_handle = buffer.get_selected_item_author_handle() # Panel needs this method text = dlg.GetValue().strip()
# if author_handle: dlg.Destroy()
# initial_text = f"@{author_handle} " if not text:
return
try:
uri = session.send_message(text, reply_to=selected_item_uri)
if uri:
output.speak(_("Reply sent."), True)
else:
output.speak(_("Failed to send reply."), True)
except Exception:
log.exception("Error sending Bluesky reply (invisible)")
output.speak(_("An error occurred while replying."), True)
else:
dlg.Destroy()
return
from wxUI.dialogs.atprotosocial.postDialogs import Post as ATPostDialog
dlg = ATPostDialog(caption=_("Reply"))
if dlg.ShowModal() == wx.ID_OK:
text, files, cw_text, langs = dlg.get_payload()
dlg.Destroy()
if not text and not files:
return
try:
uri = session.send_message(text, files=files, cw_text=cw_text, is_sensitive=bool(cw_text), languages=langs, reply_to=selected_item_uri)
if uri:
output.speak(_("Reply sent."), True)
try:
if hasattr(buffer, "start_stream"):
buffer.start_stream(mandatory=False, play_sound=False)
except Exception:
pass
else:
output.speak(_("Failed to send reply."), True)
except Exception:
log.exception("Error sending Bluesky reply (dialog)")
output.speak(_("An error occurred while replying."), True)
else:
dlg.Destroy()
return
from wxUI.dialogs.composeDialog import ComposeDialog from wxUI.dialogs.composeDialog import ComposeDialog
dialog = ComposeDialog(parent=self.view, session=buffer.session, reply_to_uri=selected_item_uri, initial_text=initial_text) dialog = ComposeDialog(parent=self.view, session=buffer.session, reply_to_uri=selected_item_uri, initial_text="")
dialog.Show() # Or ShowModal, depending on how pubsub message for send is handled for dialog lifecycle dialog.Show()
def send_dm(self, *args, **kwargs): def send_dm(self, *args, **kwargs):
@@ -675,40 +769,58 @@ class Controller(object):
def post_retweet(self, *args, **kwargs): def post_retweet(self, *args, **kwargs):
buffer = self.get_current_buffer() buffer = self.get_current_buffer()
if hasattr(buffer, "share_item"): # Generic buffer method if hasattr(buffer, "share_item"):
return buffer.share_item() # This likely calls back to a session/handler method return buffer.share_item()
# If direct handling is needed for ATProtoSocial: session = getattr(buffer, "session", None)
elif buffer.session and buffer.session.KIND == "atprotosocial": if not session:
item_uri = buffer.get_selected_item_id() # URI of the post to potentially quote or repost return
if getattr(session, "type", "") == "atprotosocial":
item_uri = None
if hasattr(buffer, "get_selected_item_id"):
item_uri = buffer.get_selected_item_id()
if not item_uri: if not item_uri:
output.speak(_("No item selected."), True) output.speak(_("No item selected."), True)
return return
session = buffer.session if self.showing == False:
# For ATProtoSocial, the "Share" menu item (which maps to post_retweet) dlg = wx.TextEntryDialog(None, _("Write your quote (optional):"), _("Quote"))
# will now open the ComposeDialog for quoting. if dlg.ShowModal() == wx.ID_OK:
# A direct/quick repost action could be added as a separate menu item if desired. text = dlg.GetValue().strip()
dlg.Destroy()
try:
uri = session.send_message(text, quote_uri=item_uri)
if uri:
output.speak(_("Quote posted."), True)
else:
output.speak(_("Failed to send quote."), True)
except Exception:
log.exception("Error sending Bluesky quote (invisible)")
output.speak(_("An error occurred while posting the quote."), True)
else:
dlg.Destroy()
return
initial_text = "" from wxUI.dialogs.atprotosocial.postDialogs import Post as ATPostDialog
# Attempt to get context from the selected item for the quote's initial text dlg = ATPostDialog(caption=_("Quote post"))
# The buffer panel needs a method like get_selected_item_details_for_quote() if dlg.ShowModal() == wx.ID_OK:
# which might return author handle and text snippet. text, files, cw_text, langs = dlg.get_payload()
if hasattr(buffer, "get_selected_item_summary_for_quote"): dlg.Destroy()
# This method should return a string like "QT @author_handle: text_snippet..." try:
# or just the text snippet. uri = session.send_message(text, files=files, cw_text=cw_text, is_sensitive=bool(cw_text), languages=langs, quote_uri=item_uri)
quote_context_text = buffer.get_selected_item_summary_for_quote() if uri:
if quote_context_text: output.speak(_("Quote posted."), True)
initial_text = quote_context_text + "\n\n" # Add space for user's own text try:
else: # Fallback if panel doesn't provide detailed quote summary if hasattr(buffer, "start_stream"):
item_web_url = "" # Ideally, get the web URL of the post buffer.start_stream(mandatory=False, play_sound=False)
if hasattr(buffer, "get_selected_item_web_url"): except Exception:
item_web_url = buffer.get_selected_item_web_url() or "" pass
initial_text = f"Quoting {item_web_url}\n\n" else:
output.speak(_("Failed to send quote."), True)
except Exception:
from wxUI.dialogs.composeDialog import ComposeDialog log.exception("Error sending Bluesky quote (dialog)")
dialog = ComposeDialog(parent=self.view, session=session, quote_uri=item_uri, initial_text=initial_text) output.speak(_("An error occurred while posting the quote."), True)
dialog.Show() # Non-modal, send is handled via pubsub else:
dlg.Destroy()
return return
def add_to_favourites(self, *args, **kwargs): def add_to_favourites(self, *args, **kwargs):
@@ -1001,11 +1113,11 @@ class Controller(object):
self.current_account = account self.current_account = account
buffer_object = self.get_first_buffer(account) buffer_object = self.get_first_buffer(account)
if buffer_object == None: if buffer_object == None:
output.speak(_(u"{0}: This account is not logged into Twitter.").format(account), True) output.speak(_(u"{0}: This account is not logged in.").format(account), True)
return return
buff = self.view.search(buffer_object.name, account) buff = self.view.search(buffer_object.name, account)
if buff == None: if buff == None:
output.speak(_(u"{0}: This account is not logged into Twitter.").format(account), True) output.speak(_(u"{0}: This account is not logged in.").format(account), True)
return return
self.view.change_buffer(buff) self.view.change_buffer(buff)
buffer = self.get_current_buffer() buffer = self.get_current_buffer()
@@ -1029,11 +1141,11 @@ class Controller(object):
self.current_account = account self.current_account = account
buffer_object = self.get_first_buffer(account) buffer_object = self.get_first_buffer(account)
if buffer_object == None: if buffer_object == None:
output.speak(_(u"{0}: This account is not logged into Twitter.").format(account), True) output.speak(_(u"{0}: This account is not logged in.").format(account), True)
return return
buff = self.view.search(buffer_object.name, account) buff = self.view.search(buffer_object.name, account)
if buff == None: if buff == None:
output.speak(_(u"{0}: This account is not logged into twitter.").format(account), True) output.speak(_(u"{0}: This account is not logged in.").format(account), True)
return return
self.view.change_buffer(buff) self.view.change_buffer(buff)
buffer = self.get_current_buffer() buffer = self.get_current_buffer()

18
src/sessionmanager/sessionManager.py
View File

@@ -12,6 +12,7 @@ import config_utils
import config import config
import application import application
import asyncio # For async event handling import asyncio # For async event handling
import wx
from pubsub import pub from pubsub import pub
from controller import settings from controller import settings
from sessions.mastodon import session as MastodonSession from sessions.mastodon import session as MastodonSession
@@ -37,8 +38,8 @@ class sessionManagerController(object):
# Initialize the manager, responsible for storing session objects. # Initialize the manager, responsible for storing session objects.
manager.setup() manager.setup()
self.view = view.sessionManagerWindow() self.view = view.sessionManagerWindow()
# Using CallAfter to handle async method from pubsub # Handle new account synchronously on the UI thread
pub.subscribe(lambda type: wx.CallAfter(asyncio.create_task, self.manage_new_account(type)), "sessionmanager.new_account") pub.subscribe(self.manage_new_account, "sessionmanager.new_account")
pub.subscribe(self.remove_account, "sessionmanager.remove_account") pub.subscribe(self.remove_account, "sessionmanager.remove_account")
if self.started == False: if self.started == False:
pub.subscribe(self.configuration, "sessionmanager.configuration") pub.subscribe(self.configuration, "sessionmanager.configuration")
@@ -122,7 +123,7 @@ class sessionManagerController(object):
log.warning(f"Unknown session type '{i.get('type')}' for ID {i.get('id')}. Skipping.") log.warning(f"Unknown session type '{i.get('type')}' for ID {i.get('id')}. Skipping.")
continue continue
s.get_configuration() # Assumes get_configuration() exists and is useful for all session types s.get_configuration() # Load per-session configuration
# For ATProtoSocial, this loads from its specific config file. # For ATProtoSocial, this loads from its specific config file.
# Login is now primarily handled by session.start() via mainController, # Login is now primarily handled by session.start() via mainController,
@@ -138,6 +139,13 @@ class sessionManagerController(object):
# except Exception as e: # except Exception as e:
# log.exception(f"Exception during pre-emptive login check for session {s.uid} ({s.kind}).") # log.exception(f"Exception during pre-emptive login check for session {s.uid} ({s.kind}).")
# continue # continue
# Try to auto-login for ATProtoSocial so the app starts with buffers ready
try:
if i.get("type") == "atprotosocial":
s.login()
except Exception:
log.exception("Auto-login failed for ATProtoSocial session %s", i.get("id"))
sessions.sessions[i.get("id")] = s # Add to global session store sessions.sessions[i.get("id")] = s # Add to global session store
self.new_sessions[i.get("id")] = s # Track as a new session for this manager instance self.new_sessions[i.get("id")] = s # Track as a new session for this manager instance
# self.view.destroy() # self.view.destroy()
@@ -145,7 +153,7 @@ class sessionManagerController(object):
def show_auth_error(self): def show_auth_error(self):
error = view.auth_error() # This seems to be a generic auth error display error = view.auth_error() # This seems to be a generic auth error display
async def manage_new_account(self, type): # Made async def manage_new_account(self, type):
# Generic settings for all account types. # Generic settings for all account types.
location = (str(time.time())[-6:]) # Unique ID for the session config directory location = (str(time.time())[-6:]) # Unique ID for the session config directory
log.debug("Creating %s session in the %s path" % (type, location)) log.debug("Creating %s session in the %s path" % (type, location))
@@ -166,7 +174,7 @@ class sessionManagerController(object):
return return
try: try:
result = await s.authorise() # Call the (now potentially async) authorise method result = s.authorise()
if result == True: if result == True:
# Session config (handle, did for atproto) should be saved by authorise/login. # Session config (handle, did for atproto) should be saved by authorise/login.
# Here we just update the session manager's internal list and UI. # Here we just update the session manager's internal list and UI.

1551
src/sessions/atprotosocial/session.py
View File

File diff suppressed because it is too large Load Diff

885
src/wxUI/buffers/atprotosocial/panels.py
View File

@@ -1,693 +1,244 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import wx import wx
import asyncio import languageHandler # Ensure _() is available
import logging import logging
from pubsub import pub import wx
import config
from mysc.repeating_timer import RepeatingTimer
from datetime import datetime
from approve.translation import translate as _ from multiplatform_widgets import widgets
from approve.notifications import NotificationError
from multiplatform_widgets import widgets # Assuming this provides a generic list control
logger = logging.getLogger(__name__) log = logging.getLogger("wxUI.buffers.atprotosocial.panels")
# Attempt to import a base panel if available, otherwise wx.Panel
try:
from ..mastodon.base import basePanel as BaseTimelinePanel # If a suitable base exists
except ImportError:
logger.warning("Mastodon basePanel not found, using wx.Panel as base for ATProtoSocial panels.")
class BaseTimelinePanel(wx.Panel): # Minimal fallback
def __init__(self, parent, name=""):
super().__init__(parent, name=name)
# Derived classes should create self.list (widgets.list)
self.list = None # Must be initialized by subclass
self.session = None # Must be set by subclass or via a method
self.account = "" # Must be set
self.name = name # Buffer name/type
self.viewer_states = {} # For like/repost URIs
def get_selected_item_id(self):
if self.list and self.list.get_selected_count() > 0:
idx = self.list.get_selected()
# Assuming item data (URI) is stored using SetItemData or similar
# This needs to be robust based on how items are actually added.
# For now, let's assume we might store URI in a parallel list or directly.
# This was a placeholder. Correct implementation relies on GetItemData if SetItemData was used.
# If item_uris list is maintained parallel to the list control items:
# if hasattr(self, "item_uris") and self.item_uris and idx < len(self.item_uris):
# return self.item_uris[idx]
# However, using GetItemData is generally cleaner if URIs are stored there.
# This method is overridden in ATProtoSocialUserTimelinePanel to use GetItemData.
pass # Base implementation might not be suitable if not overridden.
return None
def get_selected_item_author_details(self):
"""Retrieves author details for the selected item from the message cache."""
selected_item_uri = self.get_selected_item_id() # Relies on overridden get_selected_item_id
if selected_item_uri and self.session and hasattr(self.session, "message_cache"):
item_data = self.session.message_cache.get(selected_item_uri)
# if item_data and isinstance(item_data, dict):
author_dict = item_data.get("author")
if isinstance(author_dict, dict):
return author_dict
logger.debug(f"BaseTimelinePanel: Could not get author details for {selected_item_uri}. Cache entry: {self.session.message_cache.get(selected_item_uri) if self.session and hasattr(self.session, 'message_cache') else 'N/A'}")
return None
def get_selected_item_summary_for_quote(self):
"""Generates a summary string for quoting the selected post."""
selected_item_uri = self.get_selected_item_id()
if selected_item_uri and self.session and hasattr(self.session, "message_cache"):
item_data = self.session.message_cache.get(selected_item_uri)
if item_data and isinstance(item_data, dict):
record = item_data.get("record") # This is the Main post record dict/object
author_info = item_data.get("author", {})
author_handle = author_info.get("handle", "user")
text_content = getattr(record, 'text', '') if not isinstance(record, dict) else record.get('text', '')
text_snippet = (text_content[:70] + "...") if len(text_content) > 73 else text_content
# Try to get web URL for context as well
web_url = self.get_selected_item_web_url() or selected_item_uri
return f"QT @{author_handle}: \"{text_snippet}\"\n({web_url})"
return _("Quoting post...") # Fallback
def get_selected_item_web_url(self):
# This method should be overridden by specific panel types (like ATProtoSocialUserTimelinePanel)
# as URL structure is platform-dependent.
item_uri = self.get_selected_item_id()
if item_uri:
return f"Web URL for: {item_uri}" # Generic placeholder
return ""
def store_item_viewer_state(self, item_uri: str, key: str, value: Any):
if item_uri not in self.viewer_states:
self.viewer_states[item_uri] = {}
self.viewer_states[item_uri][key] = value
def get_item_viewer_state(self, item_uri: str, key: str) -> Any | None:
return self.viewer_states.get(item_uri, {}).get(key)
def set_focus_in_list(self):
if self.list:
self.list.list.SetFocus()
class ATProtoSocialUserTimelinePanel(BaseTimelinePanel): class ATProtoSocialHomeTimelinePanel(object):
def __init__(self, parent, name: str, session, target_user_did: str, target_user_handle: str): """Minimal Home timeline buffer for Bluesky.
super().__init__(parent, name=name)
self.session = session
self.account = session.label # Or session.uid / session.get_name()
self.target_user_did = target_user_did
self.target_user_handle = target_user_handle
self.type = "user_timeline" # Buffer type identifier
self.item_uris = [] # To store AT URIs of posts, parallel to list items Exposes a .buffer wx.Panel with a List control and provides
self.cursor = None # For pagination to load older posts start_stream()/get_more_items() to fetch items from atproto.
self.newest_item_timestamp = None # For fetching newer posts (not directly used by Bluesky cursor pagination for "new") """
self._setup_ui()
# Initial load is now typically triggered by mainController after buffer creation
# wx.CallAfter(asyncio.create_task, self.load_initial_posts())
def _setup_ui(self):
self.list = widgets.list(self, _("Author"), _("Post Content"), _("Date"), style=wx.LC_REPORT | wx.LC_SINGLE_SEL | wx.LC_VIRTUAL)
# Set column widths as appropriate
self.list.set_windows_size(0, 120) # Author
self.list.set_windows_size(1, 350) # Post Content (main part)
self.list.set_windows_size(2, 150) # Date
self.list.set_size()
# Bind list events if needed (e.g., item selection, activation)
# self.list.list.Bind(wx.EVT_LIST_ITEM_ACTIVATED, self.on_item_activated)
sizer = wx.BoxSizer(wx.VERTICAL)
sizer.Add(self.list.list, 1, wx.EXPAND | wx.ALL, 0) # List takes most space
self.SetSizer(sizer)
self.Layout()
async def load_initial_posts(self, limit: int = 20):
"""Loads the initial set of posts for the user's timeline."""
logger.info(f"ATProtoSocialUserTimelinePanel: Loading initial posts for {self.target_user_handle} ({self.target_user_did})")
if not self.session or not self.session.is_ready():
logger.warning("Session not ready, cannot load posts.")
# Optionally display a message in the panel
return
try:
# filter_type="posts_no_replies" or "posts_with_replies" or "posts_and_author_threads"
# "posts_and_author_threads" is good for profile view to see everything
fetched_data = await self.session.fetch_user_timeline(
user_did=self.target_user_did,
limit=limit,
new_only=True, # To get newest first
filter_type="posts_and_author_threads"
)
# fetch_user_timeline returns (processed_ids, next_cursor)
# The processed_ids are already in message_cache.
# We need to update the list control.
if fetched_data:
post_uris, self.cursor = fetched_data
self.item_uris = post_uris # Store URIs for get_selected_item_id
self.update_list_ctrl()
else:
self.list.list.DeleteAllItems() # Clear if no data
self.list.list.InsertItem(0, _("No posts found."))
except NotificationError as e:
logger.error(f"NotificationError loading posts for {self.target_user_handle}: {e.message}")
self.list.list.InsertItem(0, _("Error: ") + e.message)
except Exception as e:
logger.error(f"Error loading posts for {self.target_user_handle}: {e}", exc_info=True)
self.list.list.InsertItem(0, _("An unexpected error occurred loading posts."))
async def load_more_posts(self, limit: int = 20):
"""Loads older posts for the user's timeline using the current cursor."""
logger.info(f"ATProtoSocialUserTimelinePanel: Loading more posts for {self.target_user_handle}, cursor: {self.cursor}")
if not self.session or not self.session.is_ready() or not self.cursor:
logger.warning(f"Session not ready or no cursor, cannot load more posts. Cursor: {self.cursor}")
if not self.cursor:
self.list.list.InsertItem(self.list.list.GetItemCount(), _("No more posts to load."))
return
try:
fetched_data = await self.session.fetch_user_timeline(
user_did=self.target_user_did,
limit=limit,
cursor=self.cursor,
new_only=False, # Fetching older items
filter_type="posts_and_author_threads"
)
if fetched_data:
new_post_uris, self.cursor = fetched_data
if new_post_uris:
self.item_uris.extend(new_post_uris) # Add to existing URIs
self.update_list_ctrl(append=True) # Append new items
else:
self.list.list.InsertItem(self.list.list.GetItemCount(), _("No more posts found."))
self.cursor = None # No more items to load
else:
self.list.list.InsertItem(self.list.list.GetItemCount(), _("Failed to load more posts or no more posts."))
self.cursor = None # Stop further attempts if API returns no data structure
except NotificationError as e:
logger.error(f"NotificationError loading more posts for {self.target_user_handle}: {e.message}")
self.list.list.InsertItem(self.list.list.GetItemCount(), _("Error: ") + e.message)
except Exception as e:
logger.error(f"Error loading more posts for {self.target_user_handle}: {e}", exc_info=True)
self.list.list.InsertItem(self.list.list.GetItemCount(), _("An unexpected error occurred."))
def update_list_ctrl(self, append: bool = False):
"""Populates or updates the list control with cached post data."""
if not append:
self.list.list.DeleteAllItems()
current_uris_to_display = self.item_uris
else: # Appending, so only add new URIs
# This assumes self.item_uris has already been extended with new URIs
# And we need to find which ones are truly new to the list control items
# A simpler append strategy is just to add all from the new batch.
# For now, if append is true, this method isn't directly called with new_only=True logic from session.
# This method is mostly for full refresh or initial population.
# The `order_buffer` in session.py handles adding to `self.item_uris`.
# This method should just render what's in self.item_uris.
# Let's simplify: this method always redraws based on self.item_uris.
# If appending, the caller (load_more_posts) should have extended self.item_uris.
pass # No, if appending, we add items, don't delete all. This logic needs care.
if not append:
self.list.list.DeleteAllItems()
start_index = 0
if append:
start_index = self.list.list.GetItemCount() # Add after existing items
for i, post_uri in enumerate(self.item_uris[start_index:] if append else self.item_uris):
post_data = self.session.message_cache.get(post_uri)
if post_data and isinstance(post_data, dict):
display_string = self.session.compose_panel.compose_post_for_display(post_data)
# Split display string for columns (simplified)
lines = display_string.split('\n', 2)
author_line = lines[0]
content_line = lines[1] if len(lines) > 1 else ""
# Date is part of author_line, this is a simplification.
# A proper list control might need custom rendering or more structured data.
# For a virtual list, we'd use self.list.list.SetItemCount(len(self.item_uris))
# and implement OnGetItemText. For now, direct insertion:
actual_index = start_index + i
self.list.list.InsertItem(actual_index, author_line) # Column 0: Author + Timestamp
self.list.list.SetItem(actual_index, 1, content_line) # Column 1: Main content
self.list.list.SetItem(actual_index, 2, "") # Column 2: Date (already in header)
self.list.list.SetItemData(actual_index, post_uri) # Store URI for retrieval
else:
logger.warning(f"Post data for URI {post_uri} not found in cache or invalid format.")
self.list.list.InsertItem(start_index + i, post_uri)
self.list.list.SetItem(start_index + i, 1, _("Error: Post data missing."))
if not self.item_uris and not append:
self.list.list.InsertItem(0, _("No posts to display."))
# --- Item Interaction Methods ---
# These are now part of BaseTimelinePanel and inherited
# get_selected_item_id() -> Returns item URI from self.item_uris
# get_selected_item_author_details() -> Returns author dict from message_cache
# get_selected_item_summary_for_quote() -> Returns "QT @author: snippet..." from message_cache
# get_selected_item_web_url() -> Constructs bsky.app URL for the post
# store_item_viewer_state(item_uri, key, value) -> Stores in self.viewer_states
# get_item_viewer_state(item_uri, key) -> Retrieves from self.viewer_states
# Overriding from BaseTimelinePanel to use SetItemData for URI storage directly
def get_selected_item_id(self):
if self.list and self.list.get_selected_count() > 0:
idx = self.list.get_selected()
return self.list.list.GetItemData(idx) # Assumes URI was stored with SetItemData
return None
def get_selected_item_web_url(self):
item_uri = self.get_selected_item_id()
if item_uri and self.session:
# Attempt to get handle from cached author data if available, otherwise use DID from URI
post_data = self.session.message_cache.get(item_uri)
author_handle_or_did = item_uri.split('/')[2] # Extract DID from at://<did>/...
if post_data and isinstance(post_data, dict):
author_info = post_data.get("author")
if author_info and isinstance(author_info, dict) and author_info.get("handle"):
author_handle_or_did = author_info.get("handle")
rkey = item_uri.split('/')[-1]
return f"https://bsky.app/profile/{author_handle_or_did}/post/{rkey}"
return ""
class ATProtoSocialHomeTimelinePanel(ATProtoSocialUserTimelinePanel):
def __init__(self, parent, name: str, session): def __init__(self, parent, name: str, session):
super().__init__(parent, name, session, super().__init__()
target_user_did=session.util.get_own_did() or "N/A", self.session = session
target_user_handle=session.util.get_own_username() or "N/A") self.account = session.get_name()
self.name = name
self.type = "home_timeline" self.type = "home_timeline"
self.invisible = True
async def load_initial_posts(self, limit: int = 20): self.needs_init = True
"""Loads the initial set of posts for the home timeline.""" self.buffer = _HomePanel(parent, name)
logger.info(f"ATProtoSocialHomeTimelinePanel: Loading initial posts for home timeline for {self.session.label}") self.buffer.session = session
if not self.session or not self.session.is_ready(): self.buffer.name = name
logger.warning("Session not ready for home timeline.") # Ensure controller can resolve current account from the GUI panel
return self.buffer.account = self.account
try: self.items = [] # list of dicts: {uri, author, text, indexed_at}
# The session's fetch_home_timeline updates self.session.home_timeline_buffer and self.session.home_timeline_cursor
# It returns (processed_ids, next_cursor)
processed_ids, _ = await self.session.fetch_home_timeline(limit=limit, new_only=True)
if processed_ids:
self.item_uris = list(self.session.home_timeline_buffer) # Reflect the session buffer
self.update_list_ctrl()
else:
self.list.list.DeleteAllItems()
self.list.list.InsertItem(0, _("Home timeline is empty or failed to load."))
except Exception as e:
logger.error(f"Error loading home timeline: {e}", exc_info=True)
if self.list.list: self.list.list.InsertItem(0, _("Error loading home timeline."))
async def load_more_posts(self, limit: int = 20):
"""Loads older posts for the home timeline using the session's cursor."""
logger.info(f"ATProtoSocialHomeTimelinePanel: Loading more posts, cursor: {self.session.home_timeline_cursor}")
if not self.session or not self.session.is_ready():
logger.warning("Session not ready, cannot load more posts for home timeline.")
return
if not self.session.home_timeline_cursor:
if self.list.list: self.list.list.InsertItem(self.list.list.GetItemCount(), _("No more posts."))
return
try:
new_post_uris, _ = await self.session.fetch_home_timeline(
cursor=self.session.home_timeline_cursor,
limit=limit,
new_only=False
)
if new_post_uris:
# self.item_uris is now just a reflection of session.home_timeline_buffer
self.item_uris = list(self.session.home_timeline_buffer)
self.update_list_ctrl() # Redraw the list with the full buffer
else:
if self.list.list: self.list.list.InsertItem(self.list.list.GetItemCount(), _("No more posts found."))
self.session.home_timeline_cursor = None
except Exception as e:
logger.error(f"Error loading more for home timeline: {e}", exc_info=True)
if self.list.list: self.list.list.InsertItem(self.list.list.GetItemCount(), _("Error loading more posts."))
class ATProtoSocialNotificationPanel(BaseTimelinePanel):
def __init__(self, parent, name: str, session):
super().__init__(parent, name=name)
self.session = session
self.account = session.label
self.type = "notifications"
self.item_uris = [] # Stores notification URIs or unique IDs
self.cursor = None self.cursor = None
self._setup_ui() self._auto_timer = None
# Initial load handled by session.fetch_notifications -> send_notification_to_channel
# This panel should listen to pubsub or have a method to add notifications.
# For now, it's a static list that needs manual refresh.
pub.subscribe(self.on_new_notification_processed, f"approve.notification_processed.{self.session.uid}")
def start_stream(self, mandatory=False, play_sound=True):
def _setup_ui(self): """Fetch newest items and render them."""
# Simplified list for notifications: Author, Action, Snippet/Link, Date
self.list = widgets.list(self, _("Author"), _("Action"), _("Details"), _("Date"), style=wx.LC_REPORT | wx.LC_SINGLE_SEL | wx.LC_VIRTUAL)
self.list.set_windows_size(0, 100)
self.list.set_windows_size(1, 250)
self.list.set_windows_size(2, 150)
self.list.set_windows_size(3, 120)
self.list.set_size()
sizer = wx.BoxSizer(wx.VERTICAL)
sizer.Add(self.list.list, 1, wx.EXPAND | wx.ALL, 0)
self.SetSizer(sizer)
self.Layout()
wx.CallAfter(asyncio.create_task, self.load_initial_notifications())
async def load_initial_notifications(self, limit: int = 30):
logger.info(f"ATProtoSocialNotificationPanel: Loading initial notifications for {self.session.label}")
if not self.session or not self.session.is_ready(): return
try: try:
# fetch_notifications in session.py handles sending to channel, not directly populating a list here. count = self.session.settings["general"]["max_posts_per_call"] or 40
# This panel needs to be populated by notifications received by send_notification_to_channel. except Exception:
# For a poll-based refresh: count = 40
self.cursor = await self.session.fetch_notifications(limit=limit, cursor=None) # Returns next cursor
# The actual display items are added via pubsub from session's notification handlers
# So, this load_initial_notifications mainly serves to trigger the fetch.
# The list will be populated by on_new_notification_processed.
# If no items appear, it means they were all read or no new ones.
if not self.list.list.GetItemCount():
# If fetch_notifications itself doesn't add to list (only via pubsub),
# and no pubsub messages came through for unread items, this will be shown.
# If fetch_notifications is expected to return items directly for initial load,
# this logic would be different. For now, assuming pubsub populates.
self.list.list.InsertItem(0, _("No new unread notifications found or failed to load initial set."))
elif self.list.list.GetItemText(0).startswith(_("No new unread notifications")): # If only placeholder is there
pass # Keep placeholder until real notif comes via pubsub
except Exception as e:
logger.error(f"Error in NotificationPanel load_initial_notifications/refresh: {e}", exc_info=True)
if self.list.list and self.list.list.GetItemCount() == 0:
self.list.list.InsertItem(0, _("Error loading notifications."))
async def load_more_notifications(self, limit: int = 20):
"""Fetches older notifications using the current cursor."""
logger.info(f"ATProtoSocialNotificationPanel: Loading more notifications for {self.session.label}, cursor: {self.cursor}")
if not self.session or not self.session.is_ready():
logger.warning("Session not ready, cannot load more notifications.")
return
if not self.cursor:
logger.info("No older notifications cursor available.")
if self.list.list: self.list.list.InsertItem(self.list.list.GetItemCount(), _("No more older notifications."))
return
try: try:
# This fetch will send items via pubsub if they are "new" in the context of this fetch. api = self.session._ensure_client()
# The panel's on_new_notification_processed will then add them. # The atproto SDK expects params, not raw kwargs
# We need to ensure that fetch_notifications correctly handles pagination for older items.
# The session's fetch_notifications should ideally return the list of processed items too for direct handling here.
# For now, we rely on it sending via pubsub and updating self.cursor.
# Make 'fetch_notifications' return the items directly for "load more" scenarios
# to avoid complex pubsub interaction for prepending vs appending.
# This requires a change in session.fetch_notifications or a new method.
# Let's assume session.fetch_notifications can be used for this for now and it returns items.
# Conceptual: if session.fetch_notifications returned items directly:
# items, next_cursor = await self.session.fetch_notifications(cursor=self.cursor, limit=limit, fetch_mode="older")
# if items:
# for notif_obj in reversed(items): # If fetch_notifications returns newest first from the page
# self._add_notification_to_list(notif_obj, prepend=False) # Append older items
# self.cursor = next_cursor
# else:
# self.list.list.InsertItem(self.list.list.GetItemCount(), _("No more older notifications found."))
# self.cursor = None
# Current session.fetch_notifications sends via pubsub. This is not ideal for "load more".
# For now, "load more" on notifications will just re-trigger a general refresh.
# A proper "load older" requires session.fetch_notifications to support fetching older pages
# and this panel to append them.
output.speak(_("Refreshing recent notifications. True 'load older' for notifications is not yet fully implemented."), True)
await self.refresh_notifications(limit=limit)
except Exception as e:
logger.error(f"Error loading more notifications: {e}", exc_info=True)
if self.list.list: self.list.list.InsertItem(self.list.list.GetItemCount(), _("Error loading more notifications."))
def on_new_notification_processed(self, notification_obj: Any):
"""Handles new notification object from pubsub, adds to list control."""
# This needs to be called via wx.CallAfter if pubsub is from another thread
# For now, assuming it's called on main thread or handled by pubsub config.
# Convert Notification object to a dictionary suitable for compose_notification_for_display
# This assumes notification_obj is an instance of approve.notifications.Notification
notif_dict_for_display = {
"title": notification_obj.title,
"body": notification_obj.body,
"author_name": notification_obj.author_name,
"timestamp_dt": datetime.fromtimestamp(notification_obj.timestamp) if notification_obj.timestamp else None,
"kind": notification_obj.kind.value # Pass the string value of the enum
# Add any other fields that compose_notification_for_display might use
}
display_string = self.session.compose_panel.compose_notification_for_display(notif_dict_for_display)
# For a simple list, we might just display the string.
# If the list has columns, we need to parse `display_string` or have `compose_notification_for_display` return parts.
# For now, let's assume a single main column for the formatted string, and author for the first.
# This panel's list setup: _("Author"), _("Action"), _("Details"), _("Date")
author_display = notification_obj.author_name or _("System")
# The `display_string` from `compose_notification_for_display` usually has timestamp and title.
# We need to adapt how this is split into columns or simplify the columns.
# Let's try putting the main part of the composed string in "Action" and snippet in "Details".
parts = display_string.split('\n', 1) # Split by first newline if any
main_action_line = parts[0]
details_line = parts[1] if len(parts) > 1 else (notification_obj.body or "")
timestamp_str = ""
if notification_obj.timestamp:
timestamp_str = datetime.fromtimestamp(notification_obj.timestamp).strftime("%I:%M %p %b %d")
# Prepend to list
# Columns: Author, Action (title from compose), Details (body snippet from compose), Date
idx = self.list.list.InsertItem(0, author_display)
self.list.list.SetItem(idx, 1, main_action_line)
self.list.list.SetItem(idx, 2, (details_line[:75] + "...") if len(details_line) > 78 else details_line)
self.list.list.SetItem(idx, 3, timestamp_str) # Date string from notification object
# Store a unique ID for the notification if available (e.g., its URI or a generated one)
# This helps if we need to interact with it (e.g., mark as read, navigate to source)
unique_id = notification_obj.message_id or notification_obj.url or str(notification_obj.timestamp) # Fallback ID
self.list.list.SetItemData(idx, unique_id)
if self.list.list.GetItemCount() > 0:
# Remove placeholder "No unread notifications..." if it exists and isn't the item we just added
# This check needs to be more robust if the placeholder is generic.
first_item_text = self.list.list.GetItemText(0) if self.list.list.GetItemCount() == 1 else self.list.list.GetItemText(1) # Check previous first item if count > 1
if first_item_text.startswith(_("No unread notifications")) and self.list.list.GetItemCount() > 1:
# Find and delete the placeholder; it's safer to check by a specific marker or ensure it's always at index 0 when list is empty
for i in range(self.list.list.GetItemCount()):
if self.list.list.GetItemText(i).startswith(_("No unread notifications")):
self.list.list.DeleteItem(i)
break
elif self.list.list.GetItemText(0).startswith(_("No unread notifications")) and self.list.list.GetItemCount() == 1 and unique_id != self.list.list.GetItemData(0):
# This case can happen if the placeholder was the only item, and we added a new one.
# However, the InsertItem(0,...) already shifted it. This logic is tricky.
# A better way: if list was empty and had placeholder, clear it BEFORE inserting new.
pass
def UnbindPubSub(self): # Call this on panel close
if hasattr(self, 'session') and self.session: # Ensure session exists before trying to get uid
pub.unsubscribe(self.on_new_notification_processed, f"approve.notification_processed.{self.session.uid}")
super().Destroy()
def get_selected_item_id(self): # Returns Notification URI or URL stored with item
if self.list and self.list.get_selected_count() > 0:
idx = self.list.get_selected()
return self.list.list.GetItemData(idx)
return None
def get_selected_item_web_url(self): # Attempt to return a web URL if stored, or construct one
item_identifier = self.get_selected_item_id() # This might be a post URI, like URI, or follow URI
if item_identifier and item_identifier.startswith("at://"):
# This is a generic AT URI, try to make a bsky.app link if it's a post.
# More specific handling might be needed depending on what ID is stored.
try: try:
# Example: at://did:plc:xyz/app.bsky.feed.post/3k අඩුk අඩුj අඩු from atproto import models as at_models # type: ignore
parts = item_identifier.replace("at://", "").split("/") params = at_models.AppBskyFeedGetTimeline.Params(limit=count)
if len(parts) == 3 and parts[1] == "app.bsky.feed.post": res = api.app.bsky.feed.get_timeline(params)
did_or_handle = parts[0] except Exception:
rkey = parts[2] # Fallback to plain dict params if typed models unavailable
# Try to resolve DID to handle for a nicer URL if possible (complex here) res = api.app.bsky.feed.get_timeline({"limit": count})
return f"https://bsky.app/profile/{did_or_handle}/post/{rkey}" feed = getattr(res, "feed", [])
elif len(parts) == 3 and parts[1] == "app.bsky.actor.profile": # Link to profile self.cursor = getattr(res, "cursor", None)
did_or_handle = parts[0] self.items = []
return f"https://bsky.app/profile/{did_or_handle}" for it in feed:
except Exception as e: post = getattr(it, "post", None)
logger.debug(f"Could not parse AT URI {item_identifier} for web URL: {e}") if not post:
elif item_identifier and item_identifier.startswith("http"): # Already a web URL continue
return item_identifier record = getattr(post, "record", None)
return item_identifier # Fallback to returning the ID itself author = getattr(post, "author", None)
text = getattr(record, "text", "") if record else ""
handle = getattr(author, "handle", "") if author else ""
indexed_at = getattr(post, "indexed_at", None)
self.items.append({
"uri": getattr(post, "uri", ""),
"author": handle,
"text": text,
"indexed_at": indexed_at,
})
self._render_list(replace=True)
return len(self.items)
except Exception:
log.exception("Failed to load Bluesky home timeline")
self.buffer.list.clear()
self.buffer.list.insert_item(False, _("Error"), _("Could not load timeline."), "")
return 0
def get_more_items(self):
if not self.cursor:
return 0
try:
api = self.session._ensure_client()
try:
from atproto import models as at_models # type: ignore
params = at_models.AppBskyFeedGetTimeline.Params(limit=40, cursor=self.cursor)
res = api.app.bsky.feed.get_timeline(params)
except Exception:
res = api.app.bsky.feed.get_timeline({"limit": 40, "cursor": self.cursor})
feed = getattr(res, "feed", [])
self.cursor = getattr(res, "cursor", None)
new_items = []
for it in feed:
post = getattr(it, "post", None)
if not post:
continue
record = getattr(post, "record", None)
author = getattr(post, "author", None)
text = getattr(record, "text", "") if record else ""
handle = getattr(author, "handle", "") if author else ""
indexed_at = getattr(post, "indexed_at", None)
new_items.append({
"uri": getattr(post, "uri", ""),
"author": handle,
"text": text,
"indexed_at": indexed_at,
})
if not new_items:
return 0
self.items.extend(new_items)
self._render_list(replace=False, start=len(self.items) - len(new_items))
return len(new_items)
except Exception:
log.exception("Failed to load more Bluesky timeline items")
return 0
def _render_list(self, replace: bool, start: int = 0):
if replace:
self.buffer.list.clear()
for i in range(start, len(self.items)):
it = self.items[i]
dt = ""
if it.get("indexed_at"):
try:
# indexed_at is ISO format; show HH:MM or date
dt = str(it["indexed_at"])[:16].replace("T", " ")
except Exception:
dt = ""
text = it.get("text", "").replace("\n", " ")
if len(text) > 200:
text = text[:197] + "..."
self.buffer.list.insert_item(False, it.get("author", ""), text, dt)
# For compatibility with controller expectations
def save_positions(self):
try:
pos = self.buffer.list.get_selected()
self.session.db[self.name + "_pos"] = pos
except Exception:
pass
# Support actions that need a selected item identifier (e.g., reply)
def get_selected_item_id(self):
try:
idx = self.buffer.list.get_selected()
if idx is None or idx < 0:
return None
return self.items[idx].get("uri")
except Exception:
return None
# Auto-refresh support (polling) to simulate near real-time updates
def _periodic_refresh(self):
try:
# Ensure UI updates happen on the main thread
wx.CallAfter(self.start_stream, False, False)
except Exception:
pass
def enable_auto_refresh(self, seconds: int | None = None):
try:
if self._auto_timer:
return
if seconds is None:
# Use global update_period (minutes) → seconds; minimum 15s
minutes = config.app["app-settings"].get("update_period", 2)
seconds = max(15, int(minutes * 60))
self._auto_timer = RepeatingTimer(seconds, self._periodic_refresh)
self._auto_timer.start()
except Exception:
log.exception("Failed to enable auto refresh for Bluesky panel %s", self.name)
def disable_auto_refresh(self):
try:
if self._auto_timer:
self._auto_timer.stop()
self._auto_timer = None
except Exception:
pass
class ATProtoSocialUserListPanel(BaseTimelinePanel): class _HomePanel(wx.Panel):
def __init__(self, parent, name: str, session, list_type: str, target_user_did: str, target_user_handle: str | None = None): def __init__(self, parent, name):
super().__init__(parent, name=name) super().__init__(parent, name=name)
self.session = session self.name = name
self.account = session.label self.type = "home_timeline"
self.list_type = list_type
self.target_user_did = target_user_did
self.target_user_handle = target_user_handle or target_user_did
self.type = f"user_list_{list_type}"
self.user_list_data = []
self.cursor = None
self._setup_ui()
wx.CallAfter(asyncio.create_task, self.load_initial_users())
def _setup_ui(self):
self.list = widgets.list(self, _("Display Name"), _("Handle"), _("Bio"), style=wx.LC_REPORT | wx.LC_SINGLE_SEL | wx.LC_VIRTUAL)
self.list.set_windows_size(0, 150)
self.list.set_windows_size(1, 150)
self.list.set_windows_size(2, 300)
self.list.set_size()
sizer = wx.BoxSizer(wx.VERTICAL) sizer = wx.BoxSizer(wx.VERTICAL)
self.list = widgets.list(self, _("Author"), _("Post"), _("Date"), style=wx.LC_REPORT | wx.LC_SINGLE_SEL)
self.list.set_windows_size(0, 120)
self.list.set_windows_size(1, 360)
self.list.set_windows_size(2, 150)
self.list.set_size()
sizer.Add(self.list.list, 1, wx.EXPAND | wx.ALL, 0) sizer.Add(self.list.list, 1, wx.EXPAND | wx.ALL, 0)
self.SetSizer(sizer) self.SetSizer(sizer)
self.Layout()
async def load_initial_users(self, limit: int = 30):
logger.info(f"ATProtoSocialUserListPanel: Loading initial users for {self.list_type} of {self.target_user_handle or self.target_user_did}") class ATProtoSocialFollowingTimelinePanel(ATProtoSocialHomeTimelinePanel):
if not self.session or not self.session.is_ready(): """Following-only timeline (reverse-chronological)."""
logger.warning(f"Session not ready, cannot load {self.list_type}.")
return def __init__(self, parent, name: str, session):
super().__init__(parent, name, session)
self.type = "following_timeline"
# Make sure the underlying wx panel also reflects this type
try: try:
# Using the controller.userList function for paginated fetching directly self.buffer.type = "following_timeline"
# This requires access to mainController or passing it down. except Exception:
# For simplicity, let's assume a helper on session that calls the controller.userList function. pass
# Or, we can make this panel call a new session method that wraps this.
# For now, let's assume session has a method like `get_paginated_user_list`.
# This method needs to exist on the session:
# async def get_paginated_user_list(self, list_type, identifier, limit, cursor) -> tuple[list, str|None]:
# from controller.atprotosocial import userList as atpUserListCtrl # Keep import local
# return await atpUserListCtrl.get_user_list_paginated(self, list_type, identifier, limit, cursor)
# Always call the session method now
users, self.cursor = await self.session.get_paginated_user_list(
list_type=self.list_type,
identifier=self.target_user_did,
limit=limit,
cursor=None
)
if users:
self.user_list_data = users
self.update_list_ctrl()
else:
self.list.list.DeleteAllItems()
self.list.list.InsertItem(0, _("No users found in this list."))
except Exception as e:
logger.error(f"Error loading {self.list_type} for {self.target_user_handle}: {e}", exc_info=True)
self.list.list.InsertItem(0, _("Error loading user list."))
async def load_more_users(self, limit: int = 30):
logger.info(f"ATProtoSocialUserListPanel: Loading more users for {self.list_type} of {self.target_user_handle or self.target_user_did}, cursor: {self.cursor}")
if not self.session or not self.session.is_ready():
logger.warning(f"Session not ready, cannot load more {self.list_type}.")
return
if not self.cursor: # No cursor means no more pages or initial load failed to get one
logger.info(f"No cursor available for {self.list_type} of {self.target_user_handle or self.target_user_did}, assuming no more items.")
# self.list.list.InsertItem(self.list.list.GetItemCount(), _("No more users to load.")) # Avoid duplicate messages if already shown
return
def start_stream(self, mandatory=False, play_sound=True):
try: try:
new_users, next_cursor = await self.session.get_paginated_user_list( count = self.session.settings["general"]["max_posts_per_call"] or 40
list_type=self.list_type, except Exception:
identifier=self.target_user_did, count = 40
limit=limit, try:
cursor=self.cursor api = self.session._ensure_client()
) # Use plain dict params to ensure algorithm is passed regardless of SDK models version
res = api.app.bsky.feed.get_timeline({"limit": count, "algorithm": "reverse-chronological"})
self.cursor = next_cursor # Update cursor regardless of whether new_users were found feed = getattr(res, "feed", [])
self.cursor = getattr(res, "cursor", None)
if new_users: self.items = []
self.user_list_data.extend(new_users) for it in feed:
self.update_list_ctrl(append=True) post = getattr(it, "post", None)
logger.info(f"Loaded {len(new_users)} more users for {self.list_type} of {self.target_user_handle or self.target_user_did}.") if not post:
else: continue
logger.info(f"No more users found for {self.list_type} of {self.target_user_handle or self.target_user_did} with cursor {self.cursor}.") record = getattr(post, "record", None)
# self.list.list.InsertItem(self.list.list.GetItemCount(), _("No more users found.")) # Message can be optional author = getattr(post, "author", None)
except NotificationError as e: # Catch errors from session.get_paginated_user_list text = getattr(record, "text", "") if record else ""
logger.error(f"NotificationError loading more {self.list_type} for {self.target_user_handle}: {e.message}", exc_info=True) handle = getattr(author, "handle", "") if author else ""
if self.list.list: self.list.list.InsertItem(self.list.list.GetItemCount(), _("Error: ") + e.message) indexed_at = getattr(post, "indexed_at", None)
except Exception as e: self.items.append({
logger.error(f"Error loading more {self.list_type} for {self.target_user_handle}: {e}", exc_info=True) "uri": getattr(post, "uri", ""),
if self.list.list: self.list.list.InsertItem(self.list.list.GetItemCount(), _("An unexpected error occurred while loading more users.")) "author": handle,
"text": text,
def update_list_ctrl(self, append: bool = False): "indexed_at": indexed_at,
"""Populates or updates the list control with user data.""" })
if not append: self._render_list(replace=True)
self.list.list.DeleteAllItems() return len(self.items)
except Exception:
start_index = 0 log.exception("Failed to load Bluesky following timeline")
if append: self.buffer.list.clear()
start_index = self.list.list.GetItemCount() self.buffer.list.insert_item(False, _("Error"), _("Could not load timeline."), "")
items_to_add = self.user_list_data[start_index:] return 0
else:
items_to_add = self.user_list_data
for i, user_data in enumerate(items_to_add):
if not isinstance(user_data, dict): continue # Should be formatted dicts
display_name = user_data.get("displayName", "")
handle = user_data.get("handle", "")
description = user_data.get("description", "")
actual_index = start_index + i
self.list.list.InsertItem(actual_index, display_name)
self.list.list.SetItem(actual_index, 1, f"@{handle}")
self.list.list.SetItem(actual_index, 2, description.replace("\n", " ")) # Show bio on one line
self.list.list.SetItemData(actual_index, user_data.get("did")) # Store DID for actions
if not self.user_list_data and not append:
self.list.list.InsertItem(0, _("This list is empty."))
# Override item interaction methods if the data stored/retrieved needs different handling
def get_selected_item_id(self): # Returns DID for users
if self.list and self.list.get_selected_count() > 0:
idx = self.list.get_selected()
return self.list.list.GetItemData(idx) # DID was stored here
return None
def get_selected_item_author_details(self): # For a user list, the "author" is the user item itself
selected_did = self.get_selected_item_id()
if selected_did:
# Find the user_data dict in self.user_list_data
for user_data_item in self.user_list_data:
if user_data_item.get("did") == selected_did:
return user_data_item # Return the whole dict, mainController.user_details can use it
return None
def get_selected_item_summary_for_quote(self): # Not applicable for a list of users
return ""
def get_selected_item_web_url(self): # Construct profile URL
selected_did = self.get_selected_item_id()
if selected_did:
# Find handle from self.user_list_data
for user_data_item in self.user_list_data:
if user_data_item.get("did") == selected_did:
handle = user_data_item.get("handle")
if handle: return f"https://bsky.app/profile/{handle}"
return f"https://bsky.app/profile/{selected_did}" # Fallback to DID
return ""

5
src/wxUI/commonMessageDialogs.py
View File

@@ -59,3 +59,8 @@ def remove_filter():
return dlg.ShowModal() return dlg.ShowModal()
def error_removing_filters(): def error_removing_filters():
return wx.MessageDialog(None, _("TWBlue was unable to remove the filter you specified. Please try again."), _("Error"), wx.ICON_ERROR).ShowModal() return wx.MessageDialog(None, _("TWBlue was unable to remove the filter you specified. Please try again."), _("Error"), wx.ICON_ERROR).ShowModal()
def common_error(message):
"""Show a generic error dialog with the provided message."""
dlg = wx.MessageDialog(None, message, _("Error"), wx.OK | wx.ICON_ERROR)
return dlg.ShowModal()

103
src/wxUI/dialogs/atprotosocial/postDialogs.py Normal file
View File

@@ -0,0 +1,103 @@
# -*- coding: utf-8 -*-
import wx
class Post(wx.Dialog):
def __init__(self, caption=_("Post"), text="", *args, **kwds):
super(Post, self).__init__(parent=None, id=wx.ID_ANY, *args, **kwds)
self.SetTitle(caption)
main_sizer = wx.BoxSizer(wx.VERTICAL)
# Text
self.text = wx.TextCtrl(self, wx.ID_ANY, text, style=wx.TE_MULTILINE)
self.text.SetMinSize((400, 160))
main_sizer.Add(self.text, 1, wx.EXPAND | wx.ALL, 6)
# Sensitive + CW
cw_box = wx.BoxSizer(wx.HORIZONTAL)
self.sensitive = wx.CheckBox(self, wx.ID_ANY, _("Sensitive content (CW)"))
self.spoiler = wx.TextCtrl(self, wx.ID_ANY)
self.spoiler.Enable(False)
self.sensitive.Bind(wx.EVT_CHECKBOX, lambda evt: self.spoiler.Enable(self.sensitive.GetValue()))
cw_box.Add(self.sensitive, 0, wx.ALIGN_CENTER_VERTICAL | wx.ALL, 4)
cw_box.Add(self.spoiler, 1, wx.ALIGN_CENTER_VERTICAL | wx.ALL, 4)
main_sizer.Add(cw_box, 0, wx.EXPAND | wx.LEFT | wx.RIGHT, 2)
# Attachments (images only)
attach_box = wx.StaticBoxSizer(wx.VERTICAL, self, _("Attachments (images)"))
self.attach_list = wx.ListCtrl(self, style=wx.LC_REPORT | wx.LC_SINGLE_SEL)
self.attach_list.InsertColumn(0, _("File"))
self.attach_list.InsertColumn(1, _("Alt"))
attach_box.Add(self.attach_list, 1, wx.EXPAND | wx.ALL, 5)
btn_row = wx.BoxSizer(wx.HORIZONTAL)
self.btn_add = wx.Button(self, wx.ID_ADD, _("Add image..."))
self.btn_remove = wx.Button(self, wx.ID_REMOVE, _("Remove"))
self.btn_remove.Enable(False)
btn_row.Add(self.btn_add, 0, wx.ALL, 2)
btn_row.Add(self.btn_remove, 0, wx.ALL, 2)
attach_box.Add(btn_row, 0, wx.ALIGN_LEFT)
main_sizer.Add(attach_box, 0, wx.EXPAND | wx.LEFT | wx.RIGHT | wx.BOTTOM, 6)
# Language (single optional)
lang_row = wx.BoxSizer(wx.HORIZONTAL)
lang_row.Add(wx.StaticText(self, label=_("Language")), 0, wx.ALIGN_CENTER_VERTICAL | wx.RIGHT, 4)
self.lang_choice = wx.ComboBox(self, wx.ID_ANY, choices=["", "en", "es", "fr", "de", "ja", "pt", "ru", "zh"], style=wx.CB_DROPDOWN | wx.CB_READONLY)
self.lang_choice.SetSelection(0)
lang_row.Add(self.lang_choice, 0, wx.ALIGN_CENTER_VERTICAL)
main_sizer.Add(lang_row, 0, wx.LEFT | wx.RIGHT | wx.BOTTOM, 6)
# Buttons
btn_sizer = wx.StdDialogButtonSizer()
self.send = wx.Button(self, wx.ID_OK, _("Send"))
self.send.SetDefault()
btn_sizer.AddButton(self.send)
cancel = wx.Button(self, wx.ID_CANCEL, _("Cancel"))
btn_sizer.AddButton(cancel)
btn_sizer.Realize()
main_sizer.Add(btn_sizer, 0, wx.ALIGN_RIGHT | wx.ALL, 4)
self.SetSizer(main_sizer)
main_sizer.Fit(self)
self.Layout()
# Bindings
self.btn_add.Bind(wx.EVT_BUTTON, self.on_add)
self.btn_remove.Bind(wx.EVT_BUTTON, self.on_remove)
self.attach_list.Bind(wx.EVT_LIST_ITEM_SELECTED, lambda evt: self.btn_remove.Enable(True))
self.attach_list.Bind(wx.EVT_LIST_ITEM_DESELECTED, lambda evt: self.btn_remove.Enable(False))
def on_add(self, evt):
if self.attach_list.GetItemCount() >= 4:
wx.MessageBox(_("You can attach up to 4 images."), _("Attachment limit"), wx.ICON_INFORMATION)
return
fd = wx.FileDialog(self, _("Select image"), wildcard=_("Image files (*.png;*.jpg;*.jpeg;*.gif)|*.png;*.jpg;*.jpeg;*.gif"), style=wx.FD_OPEN | wx.FD_FILE_MUST_EXIST)
if fd.ShowModal() != wx.ID_OK:
fd.Destroy()
return
path = fd.GetPath()
fd.Destroy()
alt_dlg = wx.TextEntryDialog(self, _("Alternative text (optional)"), _("Description"))
alt = ""
if alt_dlg.ShowModal() == wx.ID_OK:
alt = alt_dlg.GetValue()
alt_dlg.Destroy()
idx = self.attach_list.InsertItem(self.attach_list.GetItemCount(), path)
self.attach_list.SetItem(idx, 1, alt)
def on_remove(self, evt):
sel = self.attach_list.GetFirstSelected()
if sel != -1:
self.attach_list.DeleteItem(sel)
def get_payload(self):
text = self.text.GetValue().strip()
cw_text = self.spoiler.GetValue().strip() if self.sensitive.GetValue() else None
lang = self.lang_choice.GetValue().strip() or None
files = []
for i in range(self.attach_list.GetItemCount()):
files.append({
"path": self.attach_list.GetItemText(i, 0),
"alt": self.attach_list.GetItemText(i, 1),
})
return text, files, cw_text, (lang and [lang] or [])