From 3d5ace7a79789d98f69e348aaa97458bcb506d28 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Nov 2025 00:11:29 +0000 Subject: [PATCH] Add Mastodon/ActivityPub connector for social media data indexing Implement a read-only connector for Mastodon-compatible instances (Mastodon, Pixelfed, BookWyrm) that indexes user's statuses, favourites, and bookmarks. Features: - MastodonConnector API client with pagination support - HTML content cleaning and markdown formatting - Incremental indexing using since_id tracking - Authentication route for adding/testing connectors - Search integration in researcher agent nodes - Celery background task for non-blocking indexing --- .../app/agents/researcher/nodes.py | 9 + .../app/agents/researcher/utils.py | 1 + .../app/connectors/mastodon_connector.py | 405 ++++++++++++++++++ surfsense_backend/app/db.py | 2 + surfsense_backend/app/routes/__init__.py | 2 + .../routes/mastodon_add_connector_route.py | 178 ++++++++ .../routes/search_source_connectors_routes.py | 55 +++ .../app/services/connector_service.py | 90 ++++ .../app/tasks/celery_tasks/connector_tasks.py | 43 ++ .../app/tasks/connector_indexers/__init__.py | 6 + .../connector_indexers/mastodon_indexer.py | 320 ++++++++++++++ 11 files changed, 1111 insertions(+) create mode 100644 surfsense_backend/app/connectors/mastodon_connector.py create mode 100644 surfsense_backend/app/routes/mastodon_add_connector_route.py create mode 100644 surfsense_backend/app/tasks/connector_indexers/mastodon_indexer.py diff --git a/surfsense_backend/app/agents/researcher/nodes.py b/surfsense_backend/app/agents/researcher/nodes.py index d1d7272ab..e76d9447b 100644 --- a/surfsense_backend/app/agents/researcher/nodes.py +++ b/surfsense_backend/app/agents/researcher/nodes.py @@ -216,6 +216,14 @@ async def search_single_connector( top_k=top_k, search_mode=search_mode, ) + elif connector == "MASTODON_CONNECTOR": + source_object, chunks = await connector_service.search_mastodon( + user_query=reformulated_query, + user_id=user_id, + search_space_id=search_space_id, + top_k=top_k, + search_mode=search_mode, + ) except Exception as e: logging.error("Error searching connector %s: %s", connector, traceback.format_exc()) return (connector, None, [], str(e)) @@ -692,6 +700,7 @@ async def fetch_documents_by_ids( "AIRTABLE_CONNECTOR": "Airtable (Selected)", "LUMA_CONNECTOR": "Luma Events (Selected)", "HOME_ASSISTANT_CONNECTOR": "Home Assistant (Selected)", + "MASTODON_CONNECTOR": "Mastodon (Selected)", } source_object = { diff --git a/surfsense_backend/app/agents/researcher/utils.py b/surfsense_backend/app/agents/researcher/utils.py index 4b74866a0..2a14c6f06 100644 --- a/surfsense_backend/app/agents/researcher/utils.py +++ b/surfsense_backend/app/agents/researcher/utils.py @@ -47,6 +47,7 @@ CONNECTOR_METADATA: dict[str, ConnectorMetadata] = { "LUMA_CONNECTOR": ConnectorMetadata("🎯", "Luma events", "Luma"), "ELASTICSEARCH_CONNECTOR": ConnectorMetadata("🔎", "Elasticsearch chunks", "Elasticsearch"), "HOME_ASSISTANT_CONNECTOR": ConnectorMetadata("🏠", "Home Assistant items", "Home Assistant"), + "MASTODON_CONNECTOR": ConnectorMetadata("🐘", "Mastodon posts", "Mastodon"), } # Default metadata for unknown connectors diff --git a/surfsense_backend/app/connectors/mastodon_connector.py b/surfsense_backend/app/connectors/mastodon_connector.py new file mode 100644 index 000000000..75603323f --- /dev/null +++ b/surfsense_backend/app/connectors/mastodon_connector.py @@ -0,0 +1,405 @@ +""" +Mastodon/ActivityPub connector for fetching social media data. + +This connector interfaces with the Mastodon REST API to retrieve: +- User's own posts (statuses) +- Favorited posts +- Bookmarked posts +- Notifications + +Works with Mastodon, Pixelfed, and other Mastodon-compatible instances. +""" + +import logging +from datetime import datetime +from typing import Any + +import aiohttp + +logger = logging.getLogger(__name__) + + +class MastodonConnector: + """Connector for Mastodon and compatible ActivityPub platforms (Pixelfed, etc.).""" + + def __init__(self, instance_url: str, access_token: str): + """ + Initialize the Mastodon connector. + + Args: + instance_url: Base URL of Mastodon instance (e.g., https://mastodon.social) + access_token: User access token for authentication + """ + self.instance_url = instance_url.rstrip("/") + self.access_token = access_token + self.headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json", + } + + async def _make_request( + self, endpoint: str, method: str = "GET", params: dict | None = None + ) -> tuple[Any | None, str | None]: + """ + Make an authenticated request to Mastodon API. + + Args: + endpoint: API endpoint (e.g., /api/v1/accounts/verify_credentials) + method: HTTP method + params: Optional query parameters + + Returns: + Tuple of (response_data, error_message) + """ + url = f"{self.instance_url}{endpoint}" + + try: + async with aiohttp.ClientSession() as session: + async with session.request( + method, url, headers=self.headers, params=params, timeout=30 + ) as response: + if response.status == 200: + return await response.json(), None + elif response.status == 401: + return None, "Invalid or expired access token" + elif response.status == 403: + return None, "Access forbidden - check token scopes" + elif response.status == 404: + return None, f"Endpoint not found: {endpoint}" + elif response.status == 429: + return None, "Rate limit exceeded - try again later" + else: + error_text = await response.text() + return None, f"API error {response.status}: {error_text}" + except aiohttp.ClientConnectorError as e: + return None, f"Connection error: Unable to reach {self.instance_url}. {e!s}" + except TimeoutError: + return None, f"Request timeout connecting to {self.instance_url}" + except Exception as e: + return None, f"Unexpected error: {e!s}" + + async def _paginate_request( + self, endpoint: str, params: dict | None = None, max_items: int = 200 + ) -> tuple[list[dict], str | None]: + """ + Make paginated requests to fetch all items. + + Args: + endpoint: API endpoint + params: Optional query parameters + max_items: Maximum number of items to fetch + + Returns: + Tuple of (items_list, error_message) + """ + all_items = [] + params = params or {} + params["limit"] = min(40, max_items) # Mastodon default max per page + + while len(all_items) < max_items: + result, error = await self._make_request(endpoint, params=params) + if error: + if all_items: + # Return what we have so far + return all_items, f"Partial fetch: {error}" + return [], error + + if not result: + break + + all_items.extend(result) + + if len(result) < params["limit"]: + # No more items + break + + # Get the last item's ID for pagination + if result: + params["max_id"] = result[-1]["id"] + + return all_items[:max_items], None + + async def verify_credentials(self) -> tuple[dict | None, str | None]: + """ + Verify the access token and get account info. + + Returns: + Tuple of (account_info, error_message) + """ + return await self._make_request("/api/v1/accounts/verify_credentials") + + async def test_connection(self) -> tuple[bool, str | None]: + """ + Test the connection to the Mastodon instance. + + Returns: + Tuple of (success, error_message) + """ + result, error = await self.verify_credentials() + if error: + return False, error + return True, None + + async def get_account_statuses( + self, account_id: str, max_items: int = 100, since_id: str | None = None + ) -> tuple[list[dict], str | None]: + """ + Get statuses (posts) from a specific account. + + Args: + account_id: The account ID + max_items: Maximum number of statuses to fetch + since_id: Only fetch statuses newer than this ID + + Returns: + Tuple of (statuses_list, error_message) + """ + params = {"exclude_replies": "false", "exclude_reblogs": "false"} + if since_id: + params["since_id"] = since_id + + return await self._paginate_request( + f"/api/v1/accounts/{account_id}/statuses", params=params, max_items=max_items + ) + + async def get_own_statuses( + self, max_items: int = 100, since_id: str | None = None + ) -> tuple[list[dict], str | None]: + """ + Get the authenticated user's own statuses. + + Args: + max_items: Maximum number of statuses to fetch + since_id: Only fetch statuses newer than this ID + + Returns: + Tuple of (statuses_list, error_message) + """ + # First get account info + account, error = await self.verify_credentials() + if error: + return [], error + + return await self.get_account_statuses( + account["id"], max_items=max_items, since_id=since_id + ) + + async def get_favourites( + self, max_items: int = 100 + ) -> tuple[list[dict], str | None]: + """ + Get the user's favorited statuses. + + Returns: + Tuple of (favourites_list, error_message) + """ + return await self._paginate_request( + "/api/v1/favourites", max_items=max_items + ) + + async def get_bookmarks( + self, max_items: int = 100 + ) -> tuple[list[dict], str | None]: + """ + Get the user's bookmarked statuses. + + Returns: + Tuple of (bookmarks_list, error_message) + """ + return await self._paginate_request( + "/api/v1/bookmarks", max_items=max_items + ) + + async def get_notifications( + self, max_items: int = 50, types: list[str] | None = None + ) -> tuple[list[dict], str | None]: + """ + Get the user's notifications. + + Args: + max_items: Maximum number of notifications to fetch + types: Filter by notification types (mention, favourite, reblog, follow, poll, etc.) + + Returns: + Tuple of (notifications_list, error_message) + """ + params = {} + if types: + params["types[]"] = types + + return await self._paginate_request( + "/api/v1/notifications", params=params, max_items=max_items + ) + + async def get_instance_info(self) -> tuple[dict | None, str | None]: + """ + Get information about the Mastodon instance. + + Returns: + Tuple of (instance_info, error_message) + """ + # Try v2 first, fall back to v1 + result, error = await self._make_request("/api/v2/instance") + if error: + result, error = await self._make_request("/api/v1/instance") + return result, error + + async def get_all_indexable_data( + self, + max_statuses: int = 100, + max_favourites: int = 50, + max_bookmarks: int = 50, + since_id: str | None = None, + ) -> tuple[list[dict], str | None]: + """ + Get all data suitable for indexing. + + This includes: + - User's own statuses + - Favorited statuses + - Bookmarked statuses + + Returns: + Tuple of (indexable_items, error_message) + """ + items = [] + errors = [] + + # Get user's own statuses + statuses, error = await self.get_own_statuses( + max_items=max_statuses, since_id=since_id + ) + if error: + errors.append(f"Statuses: {error}") + else: + for status in statuses: + items.append({ + "type": "status", + "source": "own", + "data": status, + }) + + # Get favourites + favourites, error = await self.get_favourites(max_items=max_favourites) + if error: + errors.append(f"Favourites: {error}") + else: + for status in favourites: + items.append({ + "type": "status", + "source": "favourite", + "data": status, + }) + + # Get bookmarks + bookmarks, error = await self.get_bookmarks(max_items=max_bookmarks) + if error: + errors.append(f"Bookmarks: {error}") + else: + for status in bookmarks: + items.append({ + "type": "status", + "source": "bookmark", + "data": status, + }) + + error_msg = "; ".join(errors) if errors else None + return items, error_msg + + def format_status_to_markdown(self, item: dict) -> str: + """Format a status item to markdown.""" + source = item.get("source", "unknown") + status = item.get("data", {}) + + # Extract status info + account = status.get("account", {}) + username = account.get("acct", "unknown") + display_name = account.get("display_name", username) + content = status.get("content", "") + created_at = status.get("created_at", "") + url = status.get("url", "") + visibility = status.get("visibility", "public") + + # Clean HTML content (basic cleaning) + import re + content_text = re.sub(r"<[^>]+>", "", content) + content_text = content_text.replace("&", "&") + content_text = content_text.replace("<", "<") + content_text = content_text.replace(">", ">") + content_text = content_text.replace(""", '"') + content_text = content_text.replace("'", "'") + + # Build markdown + source_label = { + "own": "My Post", + "favourite": "Favorited Post", + "bookmark": "Bookmarked Post", + }.get(source, "Post") + + md = f"# {source_label}\n\n" + md += f"**Author:** {display_name} (@{username})\n" + md += f"**Date:** {created_at}\n" + md += f"**Visibility:** {visibility}\n" + if url: + md += f"**URL:** {url}\n" + + # Engagement stats + reblogs = status.get("reblogs_count", 0) + favourites = status.get("favourites_count", 0) + replies = status.get("replies_count", 0) + if reblogs or favourites or replies: + md += f"**Engagement:** {reblogs} boosts, {favourites} favourites, {replies} replies\n" + + md += f"\n## Content\n\n{content_text}\n" + + # Media attachments + media = status.get("media_attachments", []) + if media: + md += "\n## Media\n\n" + for i, attachment in enumerate(media, 1): + media_type = attachment.get("type", "unknown") + media_url = attachment.get("url", "") + description = attachment.get("description", "No description") + md += f"{i}. [{media_type}]({media_url})" + if description: + md += f" - {description}" + md += "\n" + + # Tags/hashtags + tags = status.get("tags", []) + if tags: + tag_names = [f"#{tag.get('name', '')}" for tag in tags] + md += f"\n**Tags:** {' '.join(tag_names)}\n" + + # Mentions + mentions = status.get("mentions", []) + if mentions: + mention_names = [f"@{m.get('acct', '')}" for m in mentions] + md += f"**Mentions:** {' '.join(mention_names)}\n" + + # Poll if present + poll = status.get("poll") + if poll: + md += "\n## Poll\n\n" + for option in poll.get("options", []): + title = option.get("title", "") + votes = option.get("votes_count", 0) + md += f"- {title}: {votes} votes\n" + + # Reblog info + reblog = status.get("reblog") + if reblog: + reblog_account = reblog.get("account", {}) + md += f"\n*Boosted from @{reblog_account.get('acct', 'unknown')}*\n" + + return md + + def format_item_to_markdown(self, item: dict) -> str: + """Format any item to markdown based on its type.""" + item_type = item.get("type", "") + + if item_type == "status": + return self.format_status_to_markdown(item) + else: + # Generic formatting + return f"# {item.get('type', 'Unknown')}\n\n{item}" diff --git a/surfsense_backend/app/db.py b/surfsense_backend/app/db.py index f376afb17..7a93edc6b 100644 --- a/surfsense_backend/app/db.py +++ b/surfsense_backend/app/db.py @@ -51,6 +51,7 @@ class DocumentType(str, Enum): LUMA_CONNECTOR = "LUMA_CONNECTOR" ELASTICSEARCH_CONNECTOR = "ELASTICSEARCH_CONNECTOR" HOME_ASSISTANT_CONNECTOR = "HOME_ASSISTANT_CONNECTOR" + MASTODON_CONNECTOR = "MASTODON_CONNECTOR" class SearchSourceConnectorType(str, Enum): @@ -73,6 +74,7 @@ class SearchSourceConnectorType(str, Enum): LUMA_CONNECTOR = "LUMA_CONNECTOR" ELASTICSEARCH_CONNECTOR = "ELASTICSEARCH_CONNECTOR" HOME_ASSISTANT_CONNECTOR = "HOME_ASSISTANT_CONNECTOR" + MASTODON_CONNECTOR = "MASTODON_CONNECTOR" class ChatType(str, Enum): diff --git a/surfsense_backend/app/routes/__init__.py b/surfsense_backend/app/routes/__init__.py index b329491ee..fb9c36a21 100644 --- a/surfsense_backend/app/routes/__init__.py +++ b/surfsense_backend/app/routes/__init__.py @@ -15,6 +15,7 @@ from .home_assistant_add_connector_route import ( router as home_assistant_add_connector_router, ) from .llm_config_routes import router as llm_config_router +from .mastodon_add_connector_route import router as mastodon_add_connector_router from .logs_routes import router as logs_router from .luma_add_connector_route import router as luma_add_connector_router from .podcasts_routes import router as podcasts_router @@ -35,6 +36,7 @@ router.include_router(google_gmail_add_connector_router) router.include_router(airtable_add_connector_router) router.include_router(luma_add_connector_router) router.include_router(home_assistant_add_connector_router) +router.include_router(mastodon_add_connector_router) router.include_router(llm_config_router) router.include_router(logs_router) router.include_router(site_configuration_router) diff --git a/surfsense_backend/app/routes/mastodon_add_connector_route.py b/surfsense_backend/app/routes/mastodon_add_connector_route.py new file mode 100644 index 000000000..4ce0e0c18 --- /dev/null +++ b/surfsense_backend/app/routes/mastodon_add_connector_route.py @@ -0,0 +1,178 @@ +""" +Mastodon/ActivityPub connector authentication route. + +This route allows users to add a Mastodon connector by providing +their instance URL and access token. + +Works with Mastodon, Pixelfed, and other Mastodon-compatible instances. +""" + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel +from sqlalchemy.ext.asyncio import AsyncSession + +from app.connectors.mastodon_connector import MastodonConnector +from app.db import SearchSourceConnector, SearchSourceConnectorType, User, get_async_session +from app.users import current_active_user + +router = APIRouter() + + +class MastodonConnectorRequest(BaseModel): + """Request model for adding a Mastodon connector.""" + + search_space_id: int + instance_url: str # e.g., https://mastodon.social + access_token: str + name: str = "Mastodon" + + +class MastodonConnectorResponse(BaseModel): + """Response model for Mastodon connector creation.""" + + id: int + name: str + connector_type: str + message: str + username: str + + +@router.post("/auth/mastodon/add", response_model=MastodonConnectorResponse) +async def add_mastodon_connector( + request: MastodonConnectorRequest, + user: User = Depends(current_active_user), + session: AsyncSession = Depends(get_async_session), +): + """ + Add a Mastodon/ActivityPub connector. + + This endpoint allows users to connect their Mastodon, Pixelfed, or other + ActivityPub-compatible account by providing the instance URL and access token. + + To create an access token: + 1. Go to your Mastodon instance Settings → Development → New Application + 2. Give it a name (e.g., "SurfSense") + 3. Select read scopes: read:accounts, read:bookmarks, read:favourites, read:statuses + 4. Create the application and copy the access token + + Args: + request: The connector configuration + user: Current authenticated user + session: Database session + + Returns: + MastodonConnectorResponse with connector details + """ + # Validate and normalize URL + instance_url = request.instance_url.rstrip("/") + if not instance_url.startswith(("http://", "https://")): + instance_url = f"https://{instance_url}" + + # Test connection to Mastodon + mastodon_client = MastodonConnector( + instance_url=instance_url, + access_token=request.access_token, + ) + + # Verify credentials and get account info + account, error = await mastodon_client.verify_credentials() + if error: + raise HTTPException( + status_code=400, + detail=f"Failed to connect to Mastodon: {error}", + ) + + # Get username and display name + username = account.get("acct", "unknown") + display_name = account.get("display_name", username) + + # Create connector config + connector_config = { + "INSTANCE_URL": instance_url, + "ACCESS_TOKEN": request.access_token, + "ACCOUNT_ID": account.get("id"), + "USERNAME": username, + } + + # Use display name or provided name + connector_name = request.name if request.name != "Mastodon" else f"Mastodon (@{username})" + + # Create the connector in the database + db_connector = SearchSourceConnector( + name=connector_name, + connector_type=SearchSourceConnectorType.MASTODON_CONNECTOR, + config=connector_config, + search_space_id=request.search_space_id, + user_id=str(user.id), + is_indexable=True, + ) + + session.add(db_connector) + await session.commit() + await session.refresh(db_connector) + + return MastodonConnectorResponse( + id=db_connector.id, + name=db_connector.name, + connector_type=db_connector.connector_type.value, + message=f"Successfully connected to {instance_url} as @{username}", + username=username, + ) + + +@router.post("/auth/mastodon/test") +async def test_mastodon_connection( + instance_url: str, + access_token: str, + user: User = Depends(current_active_user), +): + """ + Test connection to a Mastodon instance. + + This endpoint allows users to verify their credentials before + creating a connector. + + Args: + instance_url: Mastodon instance URL + access_token: User access token + user: Current authenticated user + + Returns: + Connection status and account info + """ + # Normalize URL + instance_url = instance_url.rstrip("/") + if not instance_url.startswith(("http://", "https://")): + instance_url = f"https://{instance_url}" + + # Test connection + mastodon_client = MastodonConnector( + instance_url=instance_url, + access_token=access_token, + ) + + account, error = await mastodon_client.verify_credentials() + if error: + raise HTTPException( + status_code=400, + detail=f"Connection failed: {error}", + ) + + # Get instance info + instance_info, _ = await mastodon_client.get_instance_info() + + return { + "status": "success", + "message": "Successfully connected to Mastodon", + "account_info": { + "username": account.get("acct", "unknown"), + "display_name": account.get("display_name", ""), + "followers_count": account.get("followers_count", 0), + "following_count": account.get("following_count", 0), + "statuses_count": account.get("statuses_count", 0), + }, + "instance_info": { + "title": instance_info.get("title", "Unknown") if instance_info else "Unknown", + "version": instance_info.get("version", "Unknown") if instance_info else "Unknown", + }, + } diff --git a/surfsense_backend/app/routes/search_source_connectors_routes.py b/surfsense_backend/app/routes/search_source_connectors_routes.py index 5d2ea3387..4a5fd4fdb 100644 --- a/surfsense_backend/app/routes/search_source_connectors_routes.py +++ b/surfsense_backend/app/routes/search_source_connectors_routes.py @@ -48,6 +48,7 @@ from app.tasks.connector_indexers import ( index_jira_issues, index_linear_issues, index_luma_events, + index_mastodon_data, index_notion_pages, index_slack_messages, ) @@ -705,6 +706,22 @@ async def index_connector_content( ) response_message = "Home Assistant indexing started in the background." + elif ( + connector.connector_type + == SearchSourceConnectorType.MASTODON_CONNECTOR + ): + from app.tasks.celery_tasks.connector_tasks import ( + index_mastodon_data_task, + ) + + logger.info( + f"Triggering Mastodon indexing for connector {connector_id} into search space {search_space_id}" + ) + index_mastodon_data_task.delay( + connector_id, search_space_id, str(user.id), indexing_from, indexing_to + ) + response_message = "Mastodon indexing started in the background." + else: raise HTTPException( status_code=400, @@ -1578,3 +1595,41 @@ async def run_home_assistant_indexing( f"Critical error in run_home_assistant_indexing for connector {connector_id}: {e}", exc_info=True, ) + + +async def run_mastodon_indexing( + session: AsyncSession, + connector_id: int, + search_space_id: int, + user_id: str, + start_date: str, + end_date: str, +): + """Runs the Mastodon indexing task and updates the timestamp.""" + try: + indexed_count, error_message = await index_mastodon_data( + session, + connector_id, + search_space_id, + user_id, + start_date, + end_date, + update_last_indexed=False, + ) + if error_message: + logger.error( + f"Mastodon indexing failed for connector {connector_id}: {error_message}" + ) + else: + logger.info( + f"Mastodon indexing successful for connector {connector_id}. Indexed {indexed_count} items." + ) + # Update the last indexed timestamp only on success + await update_connector_last_indexed(session, connector_id) + await session.commit() + except Exception as e: + await session.rollback() + logger.error( + f"Critical error in run_mastodon_indexing for connector {connector_id}: {e}", + exc_info=True, + ) diff --git a/surfsense_backend/app/services/connector_service.py b/surfsense_backend/app/services/connector_service.py index ef1ca1758..1c05e393d 100644 --- a/surfsense_backend/app/services/connector_service.py +++ b/surfsense_backend/app/services/connector_service.py @@ -2645,3 +2645,93 @@ class ConnectorService: } return result_object, ha_chunks + + async def search_mastodon( + self, + user_query: str, + user_id: str, + search_space_id: int, + top_k: int = 20, + search_mode: SearchMode = SearchMode.CHUNKS, + ) -> tuple: + """ + Search for Mastodon documents and return both the source information and langchain documents + + Args: + user_query: The user's query + user_id: The user's ID + search_space_id: The search space ID to search in + top_k: Maximum number of results to return + search_mode: Search mode (CHUNKS or DOCUMENTS) + + Returns: + tuple: (sources_info, langchain_documents) + """ + if search_mode == SearchMode.CHUNKS: + mastodon_chunks = await self.chunk_retriever.hybrid_search( + query_text=user_query, + top_k=top_k, + user_id=user_id, + search_space_id=search_space_id, + document_type="MASTODON_CONNECTOR", + ) + elif search_mode == SearchMode.DOCUMENTS: + mastodon_chunks = await self.document_retriever.hybrid_search( + query_text=user_query, + top_k=top_k, + user_id=user_id, + search_space_id=search_space_id, + document_type="MASTODON_CONNECTOR", + ) + mastodon_chunks = self._transform_document_results(mastodon_chunks) + + if not mastodon_chunks: + return { + "id": 36, + "name": "Mastodon", + "type": "MASTODON_CONNECTOR", + "sources": [], + }, [] + + sources_list = [] + async with self.counter_lock: + for _i, chunk in enumerate(mastodon_chunks): + document = chunk.get("document", {}) + metadata = document.get("metadata", {}) + + status_id = metadata.get("status_id", "") + source = metadata.get("source", "") + username = metadata.get("username", "") + status_url = metadata.get("url", "") + + title = document.get("title", "Mastodon Post") + description = chunk.get("content", "")[:150] + if len(description) == 150: + description += "..." + + favourites = metadata.get("favourites_count", 0) + reblogs = metadata.get("reblogs_count", 0) + if favourites or reblogs: + description = f"[{favourites}❤️ {reblogs}🔁] {description}" + + source_entry = { + "id": chunk.get("chunk_id", self.source_id_counter), + "title": title, + "description": description, + "url": status_url, + "status_id": status_id, + "source": source, + "username": username, + } + + self.source_id_counter += 1 + sources_list.append(source_entry) + + result_object = { + "id": 36, + "name": "Mastodon", + "type": "MASTODON_CONNECTOR", + "sources": sources_list, + } + + return result_object, mastodon_chunks diff --git a/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py b/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py index 072e1ce1c..098437e09 100644 --- a/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py +++ b/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py @@ -643,3 +643,46 @@ async def _index_home_assistant_data( await run_home_assistant_indexing( session, connector_id, search_space_id, user_id, start_date, end_date ) + + +@celery_app.task(name="index_mastodon_data", bind=True) +def index_mastodon_data_task( + self, + connector_id: int, + search_space_id: int, + user_id: str, + start_date: str, + end_date: str, +): + """Celery task to index Mastodon data.""" + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + try: + loop.run_until_complete( + _index_mastodon_data( + connector_id, search_space_id, user_id, start_date, end_date + ) + ) + finally: + loop.close() + + +async def _index_mastodon_data( + connector_id: int, + search_space_id: int, + user_id: str, + start_date: str, + end_date: str, +): + """Index Mastodon data with new session.""" + from app.routes.search_source_connectors_routes import ( + run_mastodon_indexing, + ) + + async with get_celery_session_maker()() as session: + await run_mastodon_indexing( + session, connector_id, search_space_id, user_id, start_date, end_date + ) diff --git a/surfsense_backend/app/tasks/connector_indexers/__init__.py b/surfsense_backend/app/tasks/connector_indexers/__init__.py index f202b4262..51b87d383 100644 --- a/surfsense_backend/app/tasks/connector_indexers/__init__.py +++ b/surfsense_backend/app/tasks/connector_indexers/__init__.py @@ -19,6 +19,7 @@ Available indexers: - Luma: Index events from Luma - Elasticsearch: Index documents from Elasticsearch instances - Home Assistant: Index automations, scripts, and events from Home Assistant +- Mastodon: Index posts, favourites, and bookmarks from Mastodon/Pixelfed """ # Communication platforms @@ -40,6 +41,9 @@ from .jira_indexer import index_jira_issues from .linear_indexer import index_linear_issues from .luma_indexer import index_luma_events +# Social media +from .mastodon_indexer import index_mastodon_data + # Documentation and knowledge management from .notion_indexer import index_notion_pages from .slack_indexer import index_slack_messages @@ -65,4 +69,6 @@ __all__ = [ # noqa: RUF022 "index_google_gmail_messages", # Smart home "index_home_assistant_data", + # Social media + "index_mastodon_data", ] diff --git a/surfsense_backend/app/tasks/connector_indexers/mastodon_indexer.py b/surfsense_backend/app/tasks/connector_indexers/mastodon_indexer.py new file mode 100644 index 000000000..276a513c6 --- /dev/null +++ b/surfsense_backend/app/tasks/connector_indexers/mastodon_indexer.py @@ -0,0 +1,320 @@ +""" +Mastodon/ActivityPub connector indexer. +""" + +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import config +from app.connectors.mastodon_connector import MastodonConnector +from app.db import Document, DocumentType, SearchSourceConnectorType +from app.services.task_logging_service import TaskLoggingService +from app.utils.document_converters import ( + create_document_chunks, + generate_content_hash, + generate_unique_identifier_hash, +) + +from .base import ( + check_document_by_unique_identifier, + get_connector_by_id, + logger, + update_connector_last_indexed, +) + + +async def index_mastodon_data( + session: AsyncSession, + connector_id: int, + search_space_id: int, + user_id: str, + start_date: str | None = None, + end_date: str | None = None, + update_last_indexed: bool = True, +) -> tuple[int, str | None]: + """ + Index Mastodon data including user's statuses, favourites, and bookmarks. + + Args: + session: Database session + connector_id: ID of the Mastodon connector + search_space_id: ID of the search space to store documents in + user_id: User ID + start_date: Start date for indexing (not used for Mastodon, uses since_id instead) + end_date: End date for indexing (not used for Mastodon) + update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) + + Returns: + Tuple containing (number of documents indexed, error message or None) + """ + task_logger = TaskLoggingService(session, search_space_id) + + # Log task start + log_entry = await task_logger.log_task_start( + task_name="mastodon_indexing", + source="connector_indexing_task", + message=f"Starting Mastodon indexing for connector {connector_id}", + metadata={ + "connector_id": connector_id, + "user_id": str(user_id), + }, + ) + + try: + # Get the connector from the database + connector = await get_connector_by_id( + session, connector_id, SearchSourceConnectorType.MASTODON_CONNECTOR + ) + + if not connector: + await task_logger.log_task_failure( + log_entry, + f"Connector with ID {connector_id} not found", + "Connector not found", + {"error_type": "ConnectorNotFound"}, + ) + return 0, f"Connector with ID {connector_id} not found" + + # Get Mastodon credentials from the connector config + instance_url = connector.config.get("INSTANCE_URL") + access_token = connector.config.get("ACCESS_TOKEN") + + if not instance_url or not access_token: + await task_logger.log_task_failure( + log_entry, + f"Mastodon credentials not found in connector config for connector {connector_id}", + "Missing Mastodon credentials", + {"error_type": "MissingCredentials"}, + ) + return 0, "Mastodon credentials not found in connector config" + + # Initialize Mastodon client + await task_logger.log_task_progress( + log_entry, + f"Initializing Mastodon client for connector {connector_id}", + {"stage": "client_initialization"}, + ) + + mastodon_client = MastodonConnector( + instance_url=instance_url, access_token=access_token + ) + + # Test connection + connected, connection_error = await mastodon_client.test_connection() + if not connected: + await task_logger.log_task_failure( + log_entry, + f"Failed to connect to Mastodon: {connection_error}", + "Connection failed", + {"error_type": "ConnectionError"}, + ) + return 0, f"Failed to connect to Mastodon: {connection_error}" + + # Get last indexed status ID for incremental indexing + since_id = connector.config.get("LAST_STATUS_ID") + + await task_logger.log_task_progress( + log_entry, + f"Fetching Mastodon data (since_id: {since_id or 'None'})", + {"stage": "fetching_data", "since_id": since_id}, + ) + + # Get all indexable data + items, fetch_error = await mastodon_client.get_all_indexable_data( + max_statuses=200, + max_favourites=100, + max_bookmarks=100, + since_id=since_id, + ) + + if fetch_error: + logger.warning(f"Some data fetch errors occurred: {fetch_error}") + + if not items: + logger.info("No items found to index from Mastodon") + if update_last_indexed: + await update_connector_last_indexed( + session, connector, update_last_indexed + ) + await session.commit() + + await task_logger.log_task_success( + log_entry, + "No new items found to index", + {"items_indexed": 0}, + ) + return 0, None + + await task_logger.log_task_progress( + log_entry, + f"Processing {len(items)} items from Mastodon", + {"stage": "processing_items", "total_items": len(items)}, + ) + + # Track the newest status ID for next incremental indexing + newest_status_id = None + + # Process each item + documents_indexed = 0 + + for item in items: + try: + status = item.get("data", {}) + source = item.get("source", "unknown") + status_id = status.get("id", "") + + # Track newest ID for incremental indexing + if status_id and (not newest_status_id or status_id > newest_status_id): + newest_status_id = status_id + + # Generate unique identifier based on status ID and source + unique_id = f"{status_id}_{source}" + unique_identifier_hash = generate_unique_identifier_hash(unique_id) + + # Check for existing document + existing_doc = await check_document_by_unique_identifier( + session, unique_identifier_hash + ) + + # Format content to markdown + content = mastodon_client.format_item_to_markdown(item) + content_hash = generate_content_hash(content) + + # Extract account info + account = status.get("account", {}) + username = account.get("acct", "unknown") + + # Build metadata + metadata = { + "status_id": status_id, + "source": source, + "instance_url": instance_url, + "username": username, + "created_at": status.get("created_at", ""), + "visibility": status.get("visibility", "public"), + "url": status.get("url", ""), + "reblogs_count": status.get("reblogs_count", 0), + "favourites_count": status.get("favourites_count", 0), + "replies_count": status.get("replies_count", 0), + } + + # Add tags + tags = status.get("tags", []) + if tags: + metadata["tags"] = [tag.get("name", "") for tag in tags] + + # Generate title + content_preview = content[:50].replace("\n", " ").strip() + if len(content) > 50: + content_preview += "..." + + source_label = { + "own": "Post", + "favourite": "Favourite", + "bookmark": "Bookmark", + }.get(source, "Post") + title = f"Mastodon {source_label}: {content_preview}" + + if existing_doc: + # Check if content has changed + if existing_doc.content_hash == content_hash: + logger.debug(f"Skipping unchanged item: {status_id}") + continue + + # Update existing document + existing_doc.title = title + existing_doc.content = content + existing_doc.content_hash = content_hash + existing_doc.document_metadata = metadata + + # Delete old chunks and create new ones + existing_doc.chunks.clear() + + # Create new chunks + chunks = create_document_chunks( + content, + existing_doc, + config.chunker_instance, + config.embedding_model_instance, + ) + for chunk in chunks: + existing_doc.chunks.append(chunk) + + documents_indexed += 1 + logger.info(f"Updated existing Mastodon item: {status_id}") + + else: + # Create new document + document = Document( + title=title, + document_type=DocumentType.MASTODON_CONNECTOR, + document_metadata=metadata, + content=content, + content_hash=content_hash, + unique_identifier_hash=unique_identifier_hash, + search_space_id=search_space_id, + ) + + # Create chunks + chunks = create_document_chunks( + content, + document, + config.chunker_instance, + config.embedding_model_instance, + ) + for chunk in chunks: + document.chunks.append(chunk) + + session.add(document) + documents_indexed += 1 + logger.info(f"Indexed new Mastodon item: {status_id}") + + except Exception as e: + logger.error(f"Error processing item {item.get('data', {}).get('id', 'unknown')}: {e!s}") + continue + + # Update last indexed timestamp and newest status ID + if update_last_indexed: + await update_connector_last_indexed(session, connector, update_last_indexed) + + # Store newest status ID for incremental indexing + if newest_status_id: + connector.config["LAST_STATUS_ID"] = newest_status_id + + # Commit all changes + await session.commit() + + await task_logger.log_task_success( + log_entry, + f"Successfully indexed {documents_indexed} Mastodon items", + {"items_indexed": documents_indexed, "newest_status_id": newest_status_id}, + ) + + logger.info(f"Successfully indexed {documents_indexed} Mastodon items") + return documents_indexed, None + + except SQLAlchemyError as e: + await session.rollback() + error_message = f"Database error during Mastodon indexing: {e!s}" + logger.error(error_message) + + await task_logger.log_task_failure( + log_entry, + error_message, + "Database error", + {"error_type": "DatabaseError"}, + ) + return 0, error_message + + except Exception as e: + await session.rollback() + error_message = f"Unexpected error during Mastodon indexing: {e!s}" + logger.error(error_message) + + await task_logger.log_task_failure( + log_entry, + error_message, + "Unexpected error", + {"error_type": "UnexpectedError"}, + ) + return 0, error_message