mirror of
https://github.com/MCV-Software/TWBlue.git
synced 2026-07-07 07:51:19 +02:00
Terminando integración
This commit is contained in:
Jesús Pavón Abián
+177
-190
@@ -1,209 +1,196 @@
|
||||
from __future__ import annotations
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Bluesky polling-based update system for TWBlue.
|
||||
|
||||
Since Bluesky's Firehose requires complex CAR/CBOR decoding and filtering
|
||||
of millions of events, we use a polling approach instead of true streaming.
|
||||
This matches the existing start_stream() pattern used by buffers.
|
||||
|
||||
Events are published via pub/sub to maintain consistency with Mastodon's
|
||||
streaming implementation.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any, Callable, Coroutine
|
||||
import threading
|
||||
import time
|
||||
from pubsub import pub
|
||||
|
||||
if TYPE_CHECKING:
|
||||
fromapprove.sessions.blueski.session import Session as BlueskiSession
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Blueski (Bluesky) uses a Firehose model for streaming.
|
||||
# This typically involves connecting to a WebSocket endpoint and receiving events.
|
||||
# The atproto SDK provides tools for this.
|
||||
log = logging.getLogger("sessions.blueski.streaming")
|
||||
|
||||
|
||||
class BlueskiStreaming:
|
||||
def __init__(self, session: BlueskiSession, stream_type: str, params: dict[str, Any] | None = None) -> None:
|
||||
class BlueskyPoller:
|
||||
"""
|
||||
Polling-based update system for Bluesky.
|
||||
|
||||
Periodically checks for new notifications and publishes them via pub/sub.
|
||||
This provides a similar interface to Mastodon's StreamListener but uses
|
||||
polling instead of WebSocket streaming.
|
||||
"""
|
||||
|
||||
def __init__(self, session, session_name, poll_interval=60):
|
||||
"""
|
||||
Initialize the poller.
|
||||
|
||||
Args:
|
||||
session: The Bluesky session instance
|
||||
session_name: Unique identifier for this session (for pub/sub routing)
|
||||
poll_interval: Seconds between API polls (default 60, min 30)
|
||||
"""
|
||||
self.session = session
|
||||
self.stream_type = stream_type # e.g., 'user', 'public', 'hashtag' - will need mapping to Firehose concepts
|
||||
self.params = params or {}
|
||||
self._handler: Callable[[dict[str, Any]], Coroutine[Any, Any, None]] | None = None
|
||||
self._connection_task: asyncio.Task[None] | None = None
|
||||
self._should_stop = False
|
||||
# self._client = None # This would be an instance of atproto.firehose.FirehoseSubscribeReposClient or similar
|
||||
self.session_name = session_name
|
||||
self.poll_interval = max(30, poll_interval) # Minimum 30 seconds to respect rate limits
|
||||
|
||||
# TODO: Map stream_type and params to ATProto Firehose subscription needs.
|
||||
# For example, 'user' might mean subscribing to mentions, replies, follows for the logged-in user.
|
||||
# This would likely involve filtering the general repo firehose for relevant events,
|
||||
# or using a more specific subscription if available for user-level events.
|
||||
self._stop_event = threading.Event()
|
||||
self._thread = None
|
||||
self._last_notification_cursor = None
|
||||
self._last_seen_notification_uri = None
|
||||
|
||||
async def _connect(self) -> None:
|
||||
"""Internal method to connect to the Blueski Firehose."""
|
||||
# from atproto import AsyncClient
|
||||
# from atproto.firehose import FirehoseSubscribeReposClient, parse_subscribe_repos_message
|
||||
# from atproto.xrpc_client.models import get_or_create, ids, models
|
||||
|
||||
logger.info(f"Blueski streaming: Connecting to Firehose for user {self.session.user_id}, stream type {self.stream_type}")
|
||||
self._should_stop = False
|
||||
|
||||
try:
|
||||
# TODO: Replace with actual atproto SDK usage
|
||||
# client = self.session.util.get_client() # Get authenticated client from session utils
|
||||
# if not client or not client.me: # Check if client is authenticated
|
||||
# logger.error("Blueski client not authenticated. Cannot start Firehose.")
|
||||
# return
|
||||
|
||||
# self._firehose_client = FirehoseSubscribeReposClient(params=None, base_uri=self.session.api_base_url) # Adjust base_uri if needed
|
||||
|
||||
# async def on_message_handler(message: models.ComAtprotoSyncSubscribeRepos.Message) -> None:
|
||||
# if self._should_stop:
|
||||
# await self._firehose_client.stop() # Ensure client stops if flag is set
|
||||
# return
|
||||
|
||||
# # This is a simplified example. Real implementation needs to:
|
||||
# # 1. Determine the type of message (commit, handle, info, migrate, tombstone)
|
||||
# # 2. For commits, unpack operations to find posts, likes, reposts, follows, etc.
|
||||
# # 3. Filter these events to be relevant to the user (e.g., mentions, replies to user, new posts from followed users)
|
||||
# # 4. Format the data into a structure that self._handle_event expects.
|
||||
# # This filtering can be complex.
|
||||
|
||||
# # Example: if it's a commit and contains a new post that mentions the user
|
||||
# # if isinstance(message, models.ComAtprotoSyncSubscribeRepos.Commit):
|
||||
# # # This part is highly complex due to CAR CIBOR decoding
|
||||
# # # Operations need to be extracted from the commit block
|
||||
# # # For each op, check if it's a create, and if the record is a post
|
||||
# # # Then, check if the post's text or facets mention the current user.
|
||||
# # # This is a placeholder for that logic.
|
||||
# # logger.debug(f"Firehose commit from {message.repo} at {message.time}")
|
||||
# # # Example of processing ops (pseudo-code, actual decoding is more involved):
|
||||
# # # ops = message.ops
|
||||
# # # for op in ops:
|
||||
# # # if op.action == 'create' and op.path.endswith('/app.bsky.feed.post/...'):
|
||||
# # # record_data = ... # decode op.cid from message.blocks
|
||||
# # # if self.session.util.is_mention_of_me(record_data):
|
||||
# # # event_data = self.session.util.format_post_event(record_data)
|
||||
# # # await self._handle_event("mention", event_data)
|
||||
|
||||
# # For now, we'll just log that a message was received
|
||||
# logger.debug(f"Blueski Firehose message received: {message.__class__.__name__}")
|
||||
|
||||
|
||||
# await self._firehose_client.start(on_message_handler)
|
||||
|
||||
# Placeholder loop to simulate receiving events
|
||||
while not self._should_stop:
|
||||
await asyncio.sleep(1)
|
||||
# In a real implementation, this loop wouldn't exist; it'd be driven by the SDK's event handler.
|
||||
# To simulate an event:
|
||||
# if self._handler:
|
||||
# mock_event = {"type": "placeholder_event", "data": {"text": "Hello from mock stream"}}
|
||||
# await self._handler(mock_event) # Call the registered handler
|
||||
|
||||
logger.info(f"Blueski streaming: Placeholder loop for {self.session.user_id} stopped.")
|
||||
|
||||
|
||||
except asyncio.CancelledError:
|
||||
logger.info(f"Blueski streaming task for user {self.session.user_id} was cancelled.")
|
||||
except Exception as e:
|
||||
logger.error(f"Blueski streaming error for user {self.session.user_id}: {e}", exc_info=True)
|
||||
# Optional: implement retry logic here or in the start_streaming method
|
||||
if not self._should_stop:
|
||||
await asyncio.sleep(30) # Wait before trying to reconnect (if auto-reconnect is desired)
|
||||
if not self._should_stop: # Check again before restarting
|
||||
self._connection_task = asyncio.create_task(self._connect())
|
||||
|
||||
|
||||
finally:
|
||||
# if self._firehose_client:
|
||||
# await self._firehose_client.stop()
|
||||
logger.info(f"Blueski streaming connection closed for user {self.session.user_id}.")
|
||||
|
||||
|
||||
async def _handle_event(self, event_type: str, data: dict[str, Any]) -> None:
|
||||
"""
|
||||
Internal method to process an event from the stream and pass it to the session's handler.
|
||||
"""
|
||||
if self._handler:
|
||||
try:
|
||||
# The data should be transformed into a common format expected by session.handle_streaming_event
|
||||
# This is where Blueski-specific event data is mapped to Approve's internal event structure.
|
||||
# For example, an Blueski 'mention' event needs to be structured similarly to
|
||||
# how a Mastodon 'mention' event would be.
|
||||
await self.session.handle_streaming_event(event_type, data)
|
||||
except Exception as e:
|
||||
logger.error(f"Error handling Blueski streaming event type {event_type}: {e}", exc_info=True)
|
||||
else:
|
||||
logger.warning(f"Blueski streaming: No handler registered for session {self.session.user_id}, event: {event_type}")
|
||||
|
||||
|
||||
def start_streaming(self, handler: Callable[[dict[str, Any]], Coroutine[Any, Any, None]]) -> None:
|
||||
"""Starts the streaming connection."""
|
||||
if self._connection_task and not self._connection_task.done():
|
||||
logger.warning(f"Blueski streaming already active for user {self.session.user_id}.")
|
||||
def start(self):
|
||||
"""Start the polling thread."""
|
||||
if self._thread is not None and self._thread.is_alive():
|
||||
log.warning(f"Bluesky poller for {self.session_name} is already running.")
|
||||
return
|
||||
|
||||
self._handler = handler # This handler is what session.py's handle_streaming_event calls
|
||||
self._should_stop = False
|
||||
logger.info(f"Blueski streaming: Starting for user {self.session.user_id}, type: {self.stream_type}")
|
||||
self._connection_task = asyncio.create_task(self._connect())
|
||||
self._stop_event.clear()
|
||||
self._thread = threading.Thread(
|
||||
target=self._poll_loop,
|
||||
name=f"BlueskyPoller-{self.session_name}",
|
||||
daemon=True
|
||||
)
|
||||
self._thread.start()
|
||||
log.info(f"Bluesky poller started for {self.session_name} (interval: {self.poll_interval}s)")
|
||||
|
||||
def stop(self):
|
||||
"""Stop the polling thread."""
|
||||
if self._thread is None:
|
||||
return
|
||||
|
||||
async def stop_streaming(self) -> None:
|
||||
"""Stops the streaming connection."""
|
||||
logger.info(f"Blueski streaming: Stopping for user {self.session.user_id}")
|
||||
self._should_stop = True
|
||||
# if self._firehose_client: # Assuming the SDK has a stop method
|
||||
# await self._firehose_client.stop()
|
||||
self._stop_event.set()
|
||||
self._thread.join(timeout=5)
|
||||
self._thread = None
|
||||
log.info(f"Bluesky poller stopped for {self.session_name}")
|
||||
|
||||
if self._connection_task:
|
||||
if not self._connection_task.done():
|
||||
self._connection_task.cancel()
|
||||
def is_alive(self):
|
||||
"""Check if the polling thread is running."""
|
||||
return self._thread is not None and self._thread.is_alive()
|
||||
|
||||
def _poll_loop(self):
|
||||
"""Main polling loop running in background thread."""
|
||||
log.debug(f"Polling loop started for {self.session_name}")
|
||||
|
||||
# Initial delay to let the app fully initialize
|
||||
time.sleep(5)
|
||||
|
||||
while not self._stop_event.is_set():
|
||||
try:
|
||||
await self._connection_task
|
||||
except asyncio.CancelledError:
|
||||
logger.info(f"Blueski streaming task successfully cancelled for {self.session.user_id}.")
|
||||
self._connection_task = None
|
||||
self._handler = None
|
||||
logger.info(f"Blueski streaming stopped for user {self.session.user_id}.")
|
||||
self._check_notifications()
|
||||
except Exception as e:
|
||||
log.exception(f"Error in Bluesky polling loop for {self.session_name}: {e}")
|
||||
|
||||
def is_alive(self) -> bool:
|
||||
"""Checks if the streaming connection is currently active."""
|
||||
# return self._connection_task is not None and not self._connection_task.done() and self._firehose_client and self._firehose_client.is_connected
|
||||
return self._connection_task is not None and not self._connection_task.done() # Simplified check
|
||||
# Wait for next poll interval, checking stop event periodically
|
||||
for _ in range(self.poll_interval):
|
||||
if self._stop_event.is_set():
|
||||
break
|
||||
time.sleep(1)
|
||||
|
||||
def get_stream_type(self) -> str:
|
||||
return self.stream_type
|
||||
log.debug(f"Polling loop ended for {self.session_name}")
|
||||
|
||||
def get_params(self) -> dict[str, Any]:
|
||||
return self.params
|
||||
def _check_notifications(self):
|
||||
"""Check for new notifications and publish events."""
|
||||
if not self.session.logged:
|
||||
return
|
||||
|
||||
# TODO: Add methods specific to Blueski streaming if necessary,
|
||||
# e.g., methods to modify subscription details on the fly if the API supports it.
|
||||
# For Bluesky Firehose, this might not be applicable as you usually connect and filter client-side.
|
||||
# However, if there were different Firehose endpoints (e.g., one for public posts, one for user-specific events),
|
||||
# this class might manage multiple connections or re-establish with new parameters.
|
||||
try:
|
||||
api = self.session._ensure_client()
|
||||
if not api:
|
||||
return
|
||||
|
||||
# Example of how events might be processed (highly simplified):
|
||||
# This would be called by the on_message_handler in _connect
|
||||
# async def _process_firehose_message(self, message: models.ComAtprotoSyncSubscribeRepos.Message):
|
||||
# if isinstance(message, models.ComAtprotoSyncSubscribeRepos.Commit):
|
||||
# # Decode CAR file in message.blocks to get ops
|
||||
# # For each op (create, update, delete of a record):
|
||||
# # record = get_record_from_blocks(message.blocks, op.cid)
|
||||
# # if op.path.startswith("app.bsky.feed.post"): # It's a post
|
||||
# # # Check if it's a new post, a reply, a quote, etc.
|
||||
# # # Check for mentions of the current user
|
||||
# # # Example:
|
||||
# # if self.session.util.is_mention_of_me(record):
|
||||
# # formatted_event = self.session.util.format_post_as_notification(record, "mention")
|
||||
# # await self._handle_event("mention", formatted_event)
|
||||
# # elif op.path.startswith("app.bsky.graph.follow"):
|
||||
# # # Check if it's a follow of the current user
|
||||
# # if record.subject == self.session.util.get_my_did(): # Assuming get_my_did() exists
|
||||
# # formatted_event = self.session.util.format_follow_as_notification(record)
|
||||
# # await self._handle_event("follow", formatted_event)
|
||||
# # # Handle likes (app.bsky.feed.like), reposts (app.bsky.feed.repost), etc.
|
||||
# pass
|
||||
# elif isinstance(message, models.ComAtprotoSyncSubscribeRepos.Handle):
|
||||
# # Handle DID to handle mapping updates if necessary
|
||||
# logger.debug(f"Handle update: {message.handle} now points to {message.did} at {message.time}")
|
||||
# elif isinstance(message, models.ComAtprotoSyncSubscribeRepos.Migrate):
|
||||
# logger.info(f"Repo migration: {message.did} migrating from {message.migrateTo} at {message.time}")
|
||||
# elif isinstance(message, models.ComAtprotoSyncSubscribeRepos.Tombstone):
|
||||
# logger.info(f"Repo tombstone: {message.did} at {message.time}")
|
||||
# elif isinstance(message, models.ComAtprotoSyncSubscribeRepos.Info):
|
||||
# logger.info(f"Firehose info: {message.name} - {message.message}")
|
||||
# else:
|
||||
# logger.debug(f"Unknown Firehose message type: {message.__class__.__name__}")
|
||||
# Fetch recent notifications
|
||||
res = api.app.bsky.notification.list_notifications({"limit": 20})
|
||||
notifications = getattr(res, "notifications", [])
|
||||
|
||||
if not notifications:
|
||||
return
|
||||
|
||||
# Track which notifications are new
|
||||
new_notifications = []
|
||||
newest_uri = None
|
||||
|
||||
for notif in notifications:
|
||||
uri = getattr(notif, "uri", None)
|
||||
if not uri:
|
||||
continue
|
||||
|
||||
# First time running - just record the newest and don't flood
|
||||
if self._last_seen_notification_uri is None:
|
||||
newest_uri = uri
|
||||
break
|
||||
|
||||
# Check if we've seen this notification before
|
||||
if uri == self._last_seen_notification_uri:
|
||||
break
|
||||
|
||||
new_notifications.append(notif)
|
||||
if newest_uri is None:
|
||||
newest_uri = uri
|
||||
|
||||
# Update last seen
|
||||
if newest_uri:
|
||||
self._last_seen_notification_uri = newest_uri
|
||||
|
||||
# Publish new notifications (in reverse order so oldest first)
|
||||
for notif in reversed(new_notifications):
|
||||
self._publish_notification(notif)
|
||||
|
||||
except Exception as e:
|
||||
log.debug(f"Error checking notifications for {self.session_name}: {e}")
|
||||
|
||||
def _publish_notification(self, notification):
|
||||
"""Publish a notification event via pub/sub."""
|
||||
try:
|
||||
reason = getattr(notification, "reason", "unknown")
|
||||
log.debug(f"Publishing Bluesky notification: {reason} for {self.session_name}")
|
||||
|
||||
pub.sendMessage(
|
||||
"blueski.notification_received",
|
||||
notification=notification,
|
||||
session_name=self.session_name
|
||||
)
|
||||
|
||||
# Also publish specific events for certain notification types
|
||||
if reason == "mention":
|
||||
pub.sendMessage(
|
||||
"blueski.mention_received",
|
||||
notification=notification,
|
||||
session_name=self.session_name
|
||||
)
|
||||
elif reason == "reply":
|
||||
pub.sendMessage(
|
||||
"blueski.reply_received",
|
||||
notification=notification,
|
||||
session_name=self.session_name
|
||||
)
|
||||
elif reason == "follow":
|
||||
pub.sendMessage(
|
||||
"blueski.follow_received",
|
||||
notification=notification,
|
||||
session_name=self.session_name
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
log.exception(f"Error publishing notification event: {e}")
|
||||
|
||||
|
||||
def create_poller(session, session_name, poll_interval=60):
|
||||
"""
|
||||
Factory function to create a BlueskyPoller instance.
|
||||
|
||||
Args:
|
||||
session: The Bluesky session instance
|
||||
session_name: Unique identifier for this session
|
||||
poll_interval: Seconds between polls (default 60)
|
||||
|
||||
Returns:
|
||||
BlueskyPoller instance
|
||||
"""
|
||||
return BlueskyPoller(session, session_name, poll_interval)
|
||||
|
||||
Reference in New Issue
Block a user