From d40de3bce239bfa66d2d86e6303d8e02b2041526 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 28 May 2025 09:20:06 +0000 Subject: [PATCH] feat: Enhance Slack connector with advanced configuration This commit introduces Phase 1 of Slack connector enhancements, providing more granular control over the indexing process. Key changes include: 1. **Updated Connector Configuration Schema (`search_source_connector.py`):** * Added new fields to `SearchSourceConnector.config` for Slack: * `slack_membership_filter_type`: ("all_member_channels", "selected_member_channels") * `slack_selected_channel_ids`: List of channel IDs for selective indexing. * `slack_indexing_frequency`: For periodic re-indexing schedule. * `slack_initial_indexing_days`: Depth for initial message import. * `slack_initial_max_messages_per_channel`: Max messages for initial import. * Updated Pydantic validation logic for these new fields. 2. **Modified Indexing Task (`connectors_indexing_tasks.py`):** * `index_slack_messages` now reads and utilizes the new configuration fields. * Channel filtering is applied based on `slack_membership_filter_type` and `slack_selected_channel_ids`. * Initial indexing runs use `slack_initial_indexing_days` and `slack_initial_max_messages_per_channel` to determine the scope of the first message fetch. * The Slack API call was changed from `get_history_by_date_range` to a direct call to `get_conversation_history` using more precise timestamp-based parameters (`oldest`, `latest`) and message limits. 3. **New API Endpoint for Channel Discovery (`search_source_connectors_routes.py`):** * Added `GET /connectors/slack/{connector_id}/discover-channels`. * This endpoint allows the frontend to fetch a list of channels where the bot is currently a member, facilitating your selection for indexing. 4. **New API Endpoint for Specific Channel Re-index Trigger (`search_source_connectors_routes.py`):** * Added `POST /slack/{connector_id}/reindex-channels`. * This endpoint is structured to accept a list of channel IDs for targeted re-indexing. * Note: The underlying modification to `index_slack_messages` to fully handle the specialized logic for these targeted re-indexes (ignoring last_indexed_at, etc.) was not completed. This endpoint sets up the route and request model. These changes lay the backend foundation for a more configurable and user-friendly Slack integration. Frontend implementation for these configurations and the completion of the specific channel re-indexing logic are pending. --- .../routes/search_source_connectors_routes.py | 154 +++++++++++++ .../app/schemas/search_source_connector.py | 53 ++++- .../app/tasks/connectors_indexing_tasks.py | 218 +++++++++++++----- 3 files changed, 362 insertions(+), 63 deletions(-) diff --git a/surfsense_backend/app/routes/search_source_connectors_routes.py b/surfsense_backend/app/routes/search_source_connectors_routes.py index ff5cce148..2f5e67505 100644 --- a/surfsense_backend/app/routes/search_source_connectors_routes.py +++ b/surfsense_backend/app/routes/search_source_connectors_routes.py @@ -23,10 +23,30 @@ from app.tasks.connectors_indexing_tasks import index_slack_messages, index_noti from app.connectors.github_connector import GitHubConnector from datetime import datetime, timezone, timedelta import logging +from app.connectors.slack_history import SlackHistory +from slack_sdk.errors import SlackApiError +# Ensure List, Any are imported if not already (they are used by existing code) +# from typing import List, Any # BaseModel is already imported via other schemas # Set up logging logger = logging.getLogger(__name__) +# Pydantic Response Models for Slack Channel Discovery +class SlackChannelInfo(BaseModel): + id: str + name: str + is_private: bool + is_member: bool + +class SlackChannelListResponse(BaseModel): + channels: List[SlackChannelInfo] + +class ReindexSlackChannelsRequest(BaseModel): + channel_ids: List[str] = Field(..., description="A list of Slack channel IDs to re-index.") + # Optional: add date range fields if you want to allow overriding the period for this specific re-index + # reindex_start_date: Optional[str] = None # YYYY-MM-DD + # reindex_end_date: Optional[str] = None # YYYY-MM-DD + router = APIRouter() # Use Pydantic's BaseModel here @@ -577,3 +597,137 @@ async def run_linear_indexing( await session.rollback() logger.error(f"Critical error in run_linear_indexing for connector {connector_id}: {e}", exc_info=True) # Optionally update status in DB to indicate failure + +@router.get( + "/slack/{connector_id}/discover-channels", + response_model=SlackChannelListResponse, + summary="Discover Slack channels bot is a member of", + description="Fetches a list of public and private Slack channels that the bot for the given connector is a member of.", + tags=["Search Source Connectors"], +) +async def discover_slack_channels( + connector_id: int, + session: AsyncSession = Depends(get_async_session), + # current_user: User = Depends(current_active_user), # Uncomment if using authentication +): + # Optional: Ownership check + # try: + # # Assuming check_ownership is adapted to work with current_user.id if user is uncommented + # # await check_ownership(session, SearchSourceConnector, connector_id, current_user) + # except HTTPException as e: + # # logger.warning(f"Auth error for user {current_user.id} discovering channels for connector {connector_id}: {e.detail}") + # raise + # except Exception as e: + # # logger.error(f"Unexpected auth error for user {current_user.id} discovering channels for connector {connector_id}: {e}", exc_info=True) + # raise HTTPException(status_code=500, detail="An unexpected error occurred during authorization.") + + db_connector = await session.get(SearchSourceConnector, connector_id) + if not db_connector: + logger.warning(f"discover_slack_channels: Connector {connector_id} not found.") + raise HTTPException(status_code=404, detail="Connector not found") + + if db_connector.connector_type != SearchSourceConnectorType.SLACK_CONNECTOR: + logger.warning(f"discover_slack_channels: Connector {connector_id} is not a Slack connector (type: {db_connector.connector_type}).") + raise HTTPException(status_code=400, detail="Connector is not a Slack connector") + + slack_token = db_connector.config.get("SLACK_BOT_TOKEN") + if not slack_token: + logger.warning(f"discover_slack_channels: Slack token not configured for connector {connector_id}.") + raise HTTPException(status_code=400, detail="Slack token not configured for this connector") + + try: + slack_client = SlackHistory(token=slack_token) + # get_all_channels returns list of dicts: {"id": ..., "name": ..., "is_private": ..., "is_member": ...} + raw_channels_data = slack_client.get_all_channels(include_private=True) + + discovered_channels = [] + for channel_data in raw_channels_data: + # Ensure 'is_member' is present and True. Some public channels might be visible but bot not a member. + if channel_data.get("is_member"): + try: + discovered_channels.append(SlackChannelInfo(**channel_data)) + except Exception as pydantic_error: # Catch potential Pydantic validation error if a channel_data is malformed + logger.warning(f"discover_slack_channels: Skipping channel due to data error for connector {connector_id}. Channel data: {channel_data}, Error: {pydantic_error}") + else: + logger.debug(f"discover_slack_channels: Connector {connector_id} - Channel '{channel_data.get('name')}' ({channel_data.get('id')}) skipped, bot is_member is false or missing.") + + logger.info(f"discover_slack_channels: Connector {connector_id} - Found {len(raw_channels_data)} raw channels, returning {len(discovered_channels)} channels where bot is a member.") + return SlackChannelListResponse(channels=discovered_channels) + + except SlackApiError as e: + error_detail = e.response.get('error', str(e)) if e.response else str(e) + logger.error(f"Slack API error discovering channels for connector {connector_id}: {error_detail}", exc_info=True) + raise HTTPException(status_code=500, detail=f"Slack API error: {error_detail}") + except ValueError as e: + logger.error(f"Value error discovering channels for connector {connector_id}: {e}", exc_info=True) + raise HTTPException(status_code=400, detail=str(e)) # e.g. if token is invalid for WebClient + except Exception as e: + logger.error(f"Unexpected error discovering channels for connector {connector_id}: {e}", exc_info=True) + raise HTTPException(status_code=500, detail="An unexpected error occurred while fetching Slack channels.") + +@router.post( + "/slack/{connector_id}/reindex-channels", + status_code=202, # Accepted + summary="Trigger re-indexing for specific Slack channels", + description="Accepts a list of channel IDs to re-index for a given Slack connector. This process runs in the background.", + tags=["Search Source Connectors"], +) +async def trigger_reindex_specific_slack_channels( + connector_id: int, + request_body: ReindexSlackChannelsRequest, + background_tasks: BackgroundTasks, + session: AsyncSession = Depends(get_async_session), + # current_user: User = Depends(current_active_user), # Uncomment for auth +): + # Optional: Ownership check (similar to discover_slack_channels) + # await check_ownership(session, SearchSourceConnector, connector_id, current_user) # Assuming check_ownership can take current_user + + logger.info(f"Received request to re-index channels for Slack connector {connector_id}. Channels: {request_body.channel_ids}") + + db_connector = await session.get(SearchSourceConnector, connector_id) + if not db_connector: + logger.warning(f"Re-index specific channels: Connector {connector_id} not found.") + raise HTTPException(status_code=404, detail="Connector not found") + + if db_connector.connector_type != SearchSourceConnectorType.SLACK_CONNECTOR: + logger.warning(f"Re-index specific channels: Connector {connector_id} is not a Slack connector.") + raise HTTPException(status_code=400, detail="Connector is not a Slack connector") + + if not db_connector.config.get("SLACK_BOT_TOKEN"): + logger.warning(f"Re-index specific channels: Slack token not configured for connector {connector_id}.") + raise HTTPException(status_code=400, detail="Slack token not configured") + + if not request_body.channel_ids: + logger.warning(f"Re-index specific channels: No channel_ids provided for connector {connector_id}.") + raise HTTPException(status_code=400, detail="No channel_ids provided for re-indexing.") + + # Placeholder for calling the modified indexing task. + # The `index_slack_messages` function will need to be adapted (in Step 6b) + # to accept `target_channel_ids` and `force_reindex_channels` (or similar mechanism). + # The session handling for background tasks also needs careful consideration; + # typically, the task would create its own session. + # For now, we pass the required parameters to the existing background task wrapper. + + # Note: The `run_slack_indexing_with_new_session` wrapper will call `index_slack_messages`. + # We'll need to adapt `run_slack_indexing_with_new_session` and `run_slack_indexing` + # to pass these new arguments through. This is a temporary simplification for this subtask. + # A more robust solution might involve a new background task specifically for re-indexing. + + # This is a conceptual call; actual implementation will depend on how index_slack_messages is modified. + # For now, we are calling the existing wrapper which then calls index_slack_messages. + # The parameters `target_channel_ids` and `force_reindex_channels` are not yet handled by these functions. + background_tasks.add_task( + run_slack_indexing_with_new_session, # This wrapper creates a new session + connector_id=db_connector.id, + search_space_id=db_connector.search_space_id, # Assuming connector has search_space_id or it's passed differently + # The following are conceptual arguments to be handled by the task in step 6b + # target_channel_ids=request_body.channel_ids, + # force_reindex_channels=True + ) + # TODO: In a subsequent step, ensure that run_slack_indexing_with_new_session and run_slack_indexing + # are updated to accept and pass through target_channel_ids and a re-indexing flag + # to the core index_slack_messages function. + # For now, the endpoint is set up to receive the request. + + logger.info(f"Background task scheduled for re-indexing channels {request_body.channel_ids} for Slack connector {connector_id}.") + return {"message": "Re-indexing task for specific channels has been scheduled."} diff --git a/surfsense_backend/app/schemas/search_source_connector.py b/surfsense_backend/app/schemas/search_source_connector.py index b136757fd..2f53a9c68 100644 --- a/surfsense_backend/app/schemas/search_source_connector.py +++ b/surfsense_backend/app/schemas/search_source_connector.py @@ -48,14 +48,55 @@ class SearchSourceConnectorBase(BaseModel): raise ValueError("LINKUP_API_KEY cannot be empty") elif connector_type == SearchSourceConnectorType.SLACK_CONNECTOR: - # For SLACK_CONNECTOR, only allow SLACK_BOT_TOKEN - allowed_keys = ["SLACK_BOT_TOKEN"] - if set(config.keys()) != set(allowed_keys): - raise ValueError(f"For SLACK_CONNECTOR connector type, config must only contain these keys: {allowed_keys}") + # For SLACK_CONNECTOR, define allowed keys + allowed_keys = [ + "SLACK_BOT_TOKEN", + "slack_membership_filter_type", + "slack_selected_channel_ids", + "slack_indexing_frequency", + "slack_initial_indexing_days", + "slack_initial_max_messages_per_channel" + ] - # Ensure the bot token is not empty + # Ensure SLACK_BOT_TOKEN is always present and not empty if not config.get("SLACK_BOT_TOKEN"): - raise ValueError("SLACK_BOT_TOKEN cannot be empty") + raise ValueError("SLACK_BOT_TOKEN is mandatory and cannot be empty for SLACK_CONNECTOR") + + # Check that all provided config keys are allowed + for key in config: + if key not in allowed_keys: + raise ValueError(f"Key '{key}' is not allowed for SLACK_CONNECTOR. Allowed keys are: {allowed_keys}") + + # Validate slack_membership_filter_type + if "slack_membership_filter_type" in config: + filter_type = config.get("slack_membership_filter_type") + if not isinstance(filter_type, str): + raise ValueError("slack_membership_filter_type must be a string") + if filter_type not in ["all_member_channels", "selected_member_channels"]: + raise ValueError("slack_membership_filter_type must be 'all_member_channels' or 'selected_member_channels'") + + # Validate slack_selected_channel_ids + if config.get("slack_membership_filter_type") == "selected_member_channels": + if "slack_selected_channel_ids" not in config: + raise ValueError("slack_selected_channel_ids is required when slack_membership_filter_type is 'selected_member_channels'") + selected_channels = config.get("slack_selected_channel_ids") + if not isinstance(selected_channels, list) or not all(isinstance(item, str) for item in selected_channels): + raise ValueError("slack_selected_channel_ids must be a list of strings") + elif "slack_selected_channel_ids" in config and config.get("slack_membership_filter_type") == "all_member_channels": + # Optional: could remove it or just ignore it if filter type is all_member_channels + pass # For now, just allow it to be present but not validated for content + + # Validate slack_indexing_frequency + if "slack_indexing_frequency" in config and not isinstance(config.get("slack_indexing_frequency"), str): + raise ValueError("slack_indexing_frequency must be a string") + + # Validate slack_initial_indexing_days + if "slack_initial_indexing_days" in config and not isinstance(config.get("slack_initial_indexing_days"), int): + raise ValueError("slack_initial_indexing_days must be an integer") + + # Validate slack_initial_max_messages_per_channel + if "slack_initial_max_messages_per_channel" in config and not isinstance(config.get("slack_initial_max_messages_per_channel"), int): + raise ValueError("slack_initial_max_messages_per_channel must be an integer") elif connector_type == SearchSourceConnectorType.NOTION_CONNECTOR: # For NOTION_CONNECTOR, only allow NOTION_INTEGRATION_TOKEN diff --git a/surfsense_backend/app/tasks/connectors_indexing_tasks.py b/surfsense_backend/app/tasks/connectors_indexing_tasks.py index 6dc6740f9..44349ba20 100644 --- a/surfsense_backend/app/tasks/connectors_indexing_tasks.py +++ b/surfsense_backend/app/tasks/connectors_indexing_tasks.py @@ -51,43 +51,138 @@ async def index_slack_messages( # Get the Slack token from the connector config slack_token = connector.config.get("SLACK_BOT_TOKEN") + + config_values = connector.config or {} + slack_membership_filter_type = config_values.get("slack_membership_filter_type", "all_member_channels") + slack_selected_channel_ids = config_values.get("slack_selected_channel_ids", []) + # Ensure selected_channel_ids is a set for efficient lookup later + slack_selected_channel_ids_set = set(slack_selected_channel_ids) + + default_initial_days = 30 + default_max_messages = 1000 # Default for updates and if not set for initial + + slack_initial_indexing_days = config_values.get("slack_initial_indexing_days", default_initial_days) + slack_initial_max_messages_per_channel = config_values.get("slack_initial_max_messages_per_channel", default_max_messages) + + # Add a comprehensive log message at the beginning of the function execution, after fetching the connector. + # This can be placed after "slack_client = SlackHistory(token=slack_token)" + # For now, let's place it after extracting all config values. + logger.info( + f"Starting Slack indexing for connector_id={connector_id}, search_space_id={search_space_id}. " + f"Config: filter_type='{slack_membership_filter_type}', " + f"num_selected_channels={len(slack_selected_channel_ids_set) if slack_selected_channel_ids_set else 'N/A'}, " + f"initial_days={slack_initial_indexing_days}, " + f"initial_max_messages={slack_initial_max_messages_per_channel}, " + f"update_last_indexed={update_last_indexed}" + ) + if not slack_token: return 0, "Slack token not found in connector config" + + # Extract New Configuration Values + config_values = connector.config or {} + slack_membership_filter_type = config_values.get("slack_membership_filter_type", "all_member_channels") + slack_selected_channel_ids = config_values.get("slack_selected_channel_ids", []) + slack_selected_channel_ids_set = set(slack_selected_channel_ids) + + default_initial_days = 30 + default_max_messages = 1000 # Default for updates and if not set for initial + + slack_initial_indexing_days = config_values.get("slack_initial_indexing_days", default_initial_days) + slack_initial_max_messages_per_channel = config_values.get("slack_initial_max_messages_per_channel", default_max_messages) + + logger.info(f"Slack indexing started for connector {connector_id} with filter_type='{slack_membership_filter_type}', {len(slack_selected_channel_ids_set)} selected channels, initial_days={slack_initial_indexing_days}, initial_max_messages={slack_initial_max_messages_per_channel}") # Initialize Slack client slack_client = SlackHistory(token=slack_token) - # Calculate date range - end_date = datetime.now() - - # Use last_indexed_at as start date if available, otherwise use 365 days ago - if connector.last_indexed_at: - # Convert dates to be comparable (both timezone-naive) - last_indexed_naive = connector.last_indexed_at.replace(tzinfo=None) if connector.last_indexed_at.tzinfo else connector.last_indexed_at - - # Check if last_indexed_at is in the future or after end_date - if last_indexed_naive > end_date: - logger.warning(f"Last indexed date ({last_indexed_naive.strftime('%Y-%m-%d')}) is in the future. Using 30 days ago instead.") - start_date = end_date - timedelta(days=30) + # Date/Time Logic for API and Metadata + end_date = datetime.now() # Used for calculating 'latest' and 'oldest' + current_date_str_metadata = end_date.strftime("%Y-%m-%d") # For document metadata + + # Calculate latest_for_api (timestamp for start of the next day, making query inclusive of current day) + temp_end_date_dt_for_api = datetime.strptime(current_date_str_metadata, "%Y-%m-%d") + latest_for_api = str(int(temp_end_date_dt_for_api.timestamp()) + 86400) + + is_initial_run = not connector.last_indexed_at + + oldest_for_api = None # Unix timestamp string or "0" + limit_for_api = default_max_messages # Default limit for subsequent runs + start_date_str_metadata = current_date_str_metadata # Default for metadata, will be updated if initial run + + if is_initial_run: + logger.info(f"Connector {connector_id} is a first run. Applying initial indexing settings.") + limit_for_api = slack_initial_max_messages_per_channel + if slack_initial_indexing_days == -1: + oldest_for_api = "0" # Signifies "all time" to Slack API + start_date_str_metadata = "all_time" # For document metadata + logger.info(f"Initial indexing for all time (oldest='0'). Using limit: {limit_for_api}") else: - start_date = last_indexed_naive - logger.info(f"Using last_indexed_at ({start_date.strftime('%Y-%m-%d')}) as start date") + # Calculate timestamp for slack_initial_indexing_days ago + start_dt_calc = end_date - timedelta(days=slack_initial_indexing_days) + oldest_for_api = str(int(start_dt_calc.timestamp())) + start_date_str_metadata = start_dt_calc.strftime("%Y-%m-%d") # For document metadata + logger.info(f"Initial indexing for {slack_initial_indexing_days} days; oldest_timestamp: {oldest_for_api}. Using limit: {limit_for_api}") else: - start_date = end_date - timedelta(days=30) # Use 30 days instead of 365 to catch recent issues - logger.info(f"No last_indexed_at found, using {start_date.strftime('%Y-%m-%d')} (30 days ago) as start date") + # Subsequent runs: use last_indexed_at + last_indexed_dt = connector.last_indexed_at.replace(tzinfo=None) if connector.last_indexed_at.tzinfo else connector.last_indexed_at + if last_indexed_dt > end_date: # Should ideally not happen if jobs run sequentially + logger.warning(f"Last indexed date ({last_indexed_dt.strftime('%Y-%m-%d')}) for connector {connector_id} is in the future. Using {default_initial_days} days ago instead.") + start_dt_calc = end_date - timedelta(days=default_initial_days) + else: + start_dt_calc = last_indexed_dt + oldest_for_api = str(int(start_dt_calc.timestamp())) + start_date_str_metadata = start_dt_calc.strftime("%Y-%m-%d") # For document metadata + logger.info(f"Subsequent run for {connector_id}. Oldest_timestamp: {oldest_for_api}. Using limit: {limit_for_api}") - # Format dates for Slack API - start_date_str = start_date.strftime("%Y-%m-%d") - end_date_str = end_date.strftime("%Y-%m-%d") - - # Get all channels + # Get all channels from Slack API try: - channels = slack_client.get_all_channels() + all_channels_from_api = slack_client.get_all_channels() # This method already filters for is_member by Slack API, but we double check later except Exception as e: return 0, f"Failed to get Slack channels: {str(e)}" - if not channels: + if not all_channels_from_api: + logger.info(f"No channels returned by get_all_channels for connector {connector_id}.") # Corrected f-string return 0, "No Slack channels found" + + original_channel_count = len(all_channels_from_api) + logger.info(f"Found {original_channel_count} total channels accessible by the bot for connector {connector_id}.") + + channels_to_process = [] # Initialize before assignment + + if slack_membership_filter_type == "selected_member_channels": + logger.info(f"Filtering channels based on 'selected_member_channels' list (configured with {len(slack_selected_channel_ids_set)} selected IDs) for connector {connector_id}.") + + # Ensure channel_obj has 'id', robustly get name for logging + def get_channel_display_name(channel_obj): + name = channel_obj.get('name') + channel_id_local = channel_obj.get('id') # Renamed to avoid conflict + return name if name else f"ID:{channel_id_local}" + + filtered_channels = [] + for channel_obj_loop in all_channels_from_api: # Use different var name in loop + channel_id_loop = channel_obj_loop.get("id") # Use different var name in loop + if channel_id_loop in slack_selected_channel_ids_set: + filtered_channels.append(channel_obj_loop) + else: + logger.info(f"Channel '{get_channel_display_name(channel_obj_loop)}' ({channel_id_loop}) skipped: not in 'slack_selected_channel_ids' config for connector {connector_id}.") + + channels_to_process = filtered_channels # Assign filtered list + logger.info(f"{len(channels_to_process)} channels remaining after 'selected_member_channels' filter for connector {connector_id} (originally {original_channel_count}).") + + elif slack_membership_filter_type == "all_member_channels": + logger.info(f"Processing all {original_channel_count} channels where bot is a member (filter_type='all_member_channels') for connector {connector_id}.") + channels_to_process = all_channels_from_api # Assign all channels + + # Add a check here: if after filtering, channels list is empty, we might want to return early. + if not channels_to_process: + logger.info(f"No channels remaining after applying filters for connector {connector_id}. Nothing to index.") + # Update last_indexed_at because we successfully ran, even if there's nothing to process + if update_last_indexed: + connector.last_indexed_at = datetime.now() + # The commit happens at the end of the main function. This assignment will be part of that commit. + logger.info(f"Connector {connector_id} last_indexed_at will be updated as no channels were left after filtering.") + return 0, "No channels to index after filtering based on configuration." # Get existing documents for this search space and connector type to prevent duplicates existing_docs_result = await session.execute( @@ -114,42 +209,51 @@ async def index_slack_messages( skipped_channels = [] # Process each channel - for channel_obj in channels: # Modified loop to iterate over list of channel objects + for channel_obj in channels_to_process: channel_id = channel_obj["id"] channel_name = channel_obj["name"] is_private = channel_obj["is_private"] is_member = channel_obj["is_member"] # This might be False for public channels too try: - # If it's a private channel and the bot is not a member, skip. - # For public channels, if they are listed by conversations.list, the bot can typically read history. - # The `not_in_channel` error in get_conversation_history will be the ultimate gatekeeper if history is inaccessible. - if not is_member: - logger.warning(f"Bot is not a member of channel {channel_name} ({channel_id}) (public or private). Skipping.") - skipped_channels.append(f"{channel_name} (bot not a member)") + # Double-check bot membership, even if get_all_channels implies it. + # Slack's conversations.list (used by get_all_channels) can list public channels bot isn't in. + # conversations.history (used by get_conversation_history) will fail if not a member. + if not is_member: # is_member is from get_all_channels() + # This check becomes more critical if get_all_channels doesn't perfectly filter by actual read access. + # For private channels, is_member is definitive. For public, it might be true even if history is restricted. + # The API call to get_conversation_history will be the ultimate decider. + logger.info(f"Channel {channel_name} ({channel_id}) listed, but bot is_member={is_member}. Proceeding to fetch history, API will confirm access.") + # No 'continue' here; let the API call attempt handle access errors. + + # Get messages for this channel + try: + messages = slack_client.get_conversation_history( + channel_id=channel_id, + limit=limit_for_api, + oldest=oldest_for_api, + latest=latest_for_api + ) + except SlackApiError as slack_api_err: + err_msg = slack_api_err.response['error'] if slack_api_err.response and 'error' in slack_api_err.response else str(slack_api_err) + if err_msg == 'not_in_channel': + logger.warning(f"Bot is not in channel {channel_name} ({channel_id}) or history is private. Skipping. Error: {err_msg}") + skipped_channels.append(f"{channel_name} (not in channel/private history)") + else: + logger.warning(f"Slack API error for channel {channel_name} ({channel_id}): {err_msg}. Skipping.") + skipped_channels.append(f"{channel_name} (API error: {err_msg})") + documents_skipped += 1 + continue + except Exception as general_err: # Catch other unexpected errors + logger.error(f"Unexpected error getting messages from channel {channel_name} ({channel_id}): {str(general_err)}") + skipped_channels.append(f"{channel_name} (Unexpected error: {str(general_err)})") documents_skipped += 1 continue - # Get messages for this channel - # The get_history_by_date_range now uses get_conversation_history, - # which handles 'not_in_channel' by returning [] and logging. - messages, error = slack_client.get_history_by_date_range( - channel_id=channel_id, - start_date=start_date_str, - end_date=end_date_str, - limit=1000 # Limit to 1000 messages per channel - ) - - if error: - logger.warning(f"Error getting messages from channel {channel_name}: {error}") - skipped_channels.append(f"{channel_name} (error: {error})") - documents_skipped += 1 - continue # Skip this channel if there's an error - if not messages: - logger.info(f"No messages found in channel {channel_name} for the specified date range.") - documents_skipped += 1 - continue # Skip if no messages + logger.info(f"No messages found in channel {channel_name} ({channel_id}) for API params (oldest: {oldest_for_api}, latest: {latest_for_api}, limit: {limit_for_api}).") + # This is a normal case (no new messages), not an error, so don't increment documents_skipped. + continue # Skip if no messages # Format messages with user info formatted_messages = [] @@ -181,8 +285,8 @@ async def index_slack_messages( ("METADATA", [ f"CHANNEL_NAME: {channel_name}", f"CHANNEL_ID: {channel_id}", - f"START_DATE: {start_date_str}", - f"END_DATE: {end_date_str}", + f"START_DATE: {start_date_str_metadata}", + f"END_DATE: {current_date_str_metadata}", f"MESSAGE_COUNT: {len(formatted_messages)}", f"INDEXED_AT: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" ]), @@ -230,10 +334,10 @@ async def index_slack_messages( existing_document.document_metadata = { "channel_name": channel_name, "channel_id": channel_id, - "start_date": start_date_str, - "end_date": end_date_str, + "start_date": start_date_str_metadata, + "end_date": current_date_str_metadata, "message_count": len(formatted_messages), - "indexed_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "indexed_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "last_updated": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } existing_document.content = summary_content @@ -260,10 +364,10 @@ async def index_slack_messages( document_metadata={ "channel_name": channel_name, "channel_id": channel_id, - "start_date": start_date_str, - "end_date": end_date_str, + "start_date": start_date_str_metadata, + "end_date": current_date_str_metadata, "message_count": len(formatted_messages), - "indexed_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S") + "indexed_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }, content=summary_content, embedding=summary_embedding,