diff --git a/surfsense_backend/alembic/versions/36_remove_fk_constraints_for_global_llm_configs.py b/surfsense_backend/alembic/versions/36_remove_fk_constraints_for_global_llm_configs.py index fa4c929ce..1faebf2ed 100644 --- a/surfsense_backend/alembic/versions/36_remove_fk_constraints_for_global_llm_configs.py +++ b/surfsense_backend/alembic/versions/36_remove_fk_constraints_for_global_llm_configs.py @@ -8,6 +8,8 @@ Create Date: 2025-11-13 23:20:12.912741 from collections.abc import Sequence +from sqlalchemy import text + from alembic import op # revision identifiers, used by Alembic. @@ -17,6 +19,20 @@ branch_labels: str | Sequence[str] | None = None depends_on: str | Sequence[str] | None = None +def constraint_exists(connection, table_name: str, constraint_name: str) -> bool: + """Check if a constraint exists on the given table.""" + result = connection.execute( + text( + """ + SELECT 1 FROM information_schema.table_constraints + WHERE table_name = :table_name AND constraint_name = :constraint_name + """ + ), + {"table_name": table_name, "constraint_name": constraint_name}, + ) + return result.fetchone() is not None + + def upgrade() -> None: """ Remove foreign key constraints on LLM preference columns to allow global configs (negative IDs). @@ -24,50 +40,55 @@ def upgrade() -> None: Global LLM configs use negative IDs and don't exist in the llm_configs table, so we need to remove the foreign key constraints that were preventing their use. """ - # Drop the foreign key constraints - op.drop_constraint( + connection = op.get_bind() + + # Drop the foreign key constraints if they exist + constraints_to_drop = [ "user_search_space_preferences_long_context_llm_id_fkey", - "user_search_space_preferences", - type_="foreignkey", - ) - op.drop_constraint( "user_search_space_preferences_fast_llm_id_fkey", - "user_search_space_preferences", - type_="foreignkey", - ) - op.drop_constraint( "user_search_space_preferences_strategic_llm_id_fkey", - "user_search_space_preferences", - type_="foreignkey", - ) + ] + + for constraint_name in constraints_to_drop: + if constraint_exists( + connection, "user_search_space_preferences", constraint_name + ): + op.drop_constraint( + constraint_name, + "user_search_space_preferences", + type_="foreignkey", + ) + else: + print(f"Constraint '{constraint_name}' does not exist. Skipping.") def downgrade() -> None: """ Re-add foreign key constraints (will fail if any negative IDs exist in the table). """ - # Re-add the foreign key constraints - op.create_foreign_key( - "user_search_space_preferences_long_context_llm_id_fkey", - "user_search_space_preferences", - "llm_configs", - ["long_context_llm_id"], - ["id"], - ondelete="SET NULL", - ) - op.create_foreign_key( - "user_search_space_preferences_fast_llm_id_fkey", - "user_search_space_preferences", - "llm_configs", - ["fast_llm_id"], - ["id"], - ondelete="SET NULL", - ) - op.create_foreign_key( - "user_search_space_preferences_strategic_llm_id_fkey", - "user_search_space_preferences", - "llm_configs", - ["strategic_llm_id"], - ["id"], - ondelete="SET NULL", - ) + connection = op.get_bind() + + # Re-add the foreign key constraints if they don't exist + constraints_to_create = [ + ( + "user_search_space_preferences_long_context_llm_id_fkey", + "long_context_llm_id", + ), + ("user_search_space_preferences_fast_llm_id_fkey", "fast_llm_id"), + ("user_search_space_preferences_strategic_llm_id_fkey", "strategic_llm_id"), + ] + + for constraint_name, column_name in constraints_to_create: + if not constraint_exists( + connection, "user_search_space_preferences", constraint_name + ): + op.create_foreign_key( + constraint_name, + "user_search_space_preferences", + "llm_configs", + [column_name], + ["id"], + ondelete="SET NULL", + ) + else: + print(f"Constraint '{constraint_name}' already exists. Skipping.") diff --git a/surfsense_backend/alembic/versions/37_add_system_prompts_to_searchspaces.py b/surfsense_backend/alembic/versions/37_add_system_prompts_to_searchspaces.py index afdee4942..6c82b4bad 100644 --- a/surfsense_backend/alembic/versions/37_add_system_prompts_to_searchspaces.py +++ b/surfsense_backend/alembic/versions/37_add_system_prompts_to_searchspaces.py @@ -9,6 +9,7 @@ Create Date: 2025-11-19 00:00:00.000000 from collections.abc import Sequence import sqlalchemy as sa +from sqlalchemy import text from alembic import op @@ -19,24 +20,55 @@ branch_labels: str | Sequence[str] | None = None depends_on: str | Sequence[str] | None = None +def column_exists(connection, table_name: str, column_name: str) -> bool: + """Check if a column exists on the given table.""" + result = connection.execute( + text( + """ + SELECT 1 FROM information_schema.columns + WHERE table_name = :table_name AND column_name = :column_name + """ + ), + {"table_name": table_name, "column_name": column_name}, + ) + return result.fetchone() is not None + + def upgrade() -> None: """Add QnA configuration columns to searchspaces table.""" + connection = op.get_bind() + # Add citations_enabled boolean (default True) - op.add_column( - "searchspaces", - sa.Column( - "citations_enabled", sa.Boolean(), nullable=False, server_default="true" - ), - ) + if not column_exists(connection, "searchspaces", "citations_enabled"): + op.add_column( + "searchspaces", + sa.Column( + "citations_enabled", sa.Boolean(), nullable=False, server_default="true" + ), + ) + else: + print("Column 'citations_enabled' already exists. Skipping.") # Add custom instructions text field (nullable, defaults to empty) - op.add_column( - "searchspaces", - sa.Column("qna_custom_instructions", sa.Text(), nullable=True), - ) + if not column_exists(connection, "searchspaces", "qna_custom_instructions"): + op.add_column( + "searchspaces", + sa.Column("qna_custom_instructions", sa.Text(), nullable=True), + ) + else: + print("Column 'qna_custom_instructions' already exists. Skipping.") def downgrade() -> None: """Remove QnA configuration columns from searchspaces table.""" - op.drop_column("searchspaces", "qna_custom_instructions") - op.drop_column("searchspaces", "citations_enabled") + connection = op.get_bind() + + if column_exists(connection, "searchspaces", "qna_custom_instructions"): + op.drop_column("searchspaces", "qna_custom_instructions") + else: + print("Column 'qna_custom_instructions' does not exist. Skipping.") + + if column_exists(connection, "searchspaces", "citations_enabled"): + op.drop_column("searchspaces", "citations_enabled") + else: + print("Column 'citations_enabled' does not exist. Skipping.") diff --git a/surfsense_backend/alembic/versions/38_add_webcrawler_connector_enum.py b/surfsense_backend/alembic/versions/38_add_webcrawler_connector_enum.py new file mode 100644 index 000000000..1b33c31b3 --- /dev/null +++ b/surfsense_backend/alembic/versions/38_add_webcrawler_connector_enum.py @@ -0,0 +1,59 @@ +"""Add Webcrawler connector enums + +Revision ID: 38 +Revises: 37 +Create Date: 2025-11-17 17:00:00.000000 + +""" + +from collections.abc import Sequence + +from alembic import op + +revision: str = "38" +down_revision: str | None = "37" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Safely add 'WEBCRAWLER_CONNECTOR' to enum types if missing.""" + + # Add to searchsourceconnectortype enum + op.execute( + """ + DO $$ + BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_type t + JOIN pg_enum e ON t.oid = e.enumtypid + WHERE t.typname = 'searchsourceconnectortype' AND e.enumlabel = 'WEBCRAWLER_CONNECTOR' + ) THEN + ALTER TYPE searchsourceconnectortype ADD VALUE 'WEBCRAWLER_CONNECTOR'; + END IF; + END + $$; + """ + ) + + # Add to documenttype enum + op.execute( + """ + DO $$ + BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_type t + JOIN pg_enum e ON t.oid = e.enumtypid + WHERE t.typname = 'documenttype' AND e.enumlabel = 'CRAWLED_URL' + ) THEN + ALTER TYPE documenttype ADD VALUE 'CRAWLED_URL'; + END IF; + END + $$; + """ + ) + + +def downgrade() -> None: + """Remove 'WEBCRAWLER_CONNECTOR' from enum types.""" + pass diff --git a/surfsense_backend/app/agents/researcher/nodes.py b/surfsense_backend/app/agents/researcher/nodes.py index 7b0e18a11..8269f08de 100644 --- a/surfsense_backend/app/agents/researcher/nodes.py +++ b/surfsense_backend/app/agents/researcher/nodes.py @@ -667,7 +667,7 @@ async def fetch_relevant_documents( } ) - elif connector == "CRAWLED_URL": + elif connector == "WEBCRAWLER_CONNECTOR": ( source_object, crawled_urls_chunks, @@ -689,7 +689,7 @@ async def fetch_relevant_documents( writer( { "yield_value": streaming_service.format_terminal_info_delta( - f"🌐 Found {len(crawled_urls_chunks)} Web Pages chunks related to your query" + f"🌐 Found {len(crawled_urls_chunks)} Web Page chunks related to your query" ) } ) diff --git a/surfsense_backend/app/agents/researcher/qna_agent/default_prompts.py b/surfsense_backend/app/agents/researcher/qna_agent/default_prompts.py index 18ad16682..7b5d251fe 100644 --- a/surfsense_backend/app/agents/researcher/qna_agent/default_prompts.py +++ b/surfsense_backend/app/agents/researcher/qna_agent/default_prompts.py @@ -17,7 +17,6 @@ You are SurfSense, an advanced AI research assistant that provides detailed, wel {chat_history_section} - EXTENSION: "Web content saved via SurfSense browser extension" (personal browsing history) -- CRAWLED_URL: "Webpages indexed by SurfSense web crawler" (personally selected websites) - FILE: "User-uploaded documents (PDFs, Word, etc.)" (personal files) - SLACK_CONNECTOR: "Slack conversations and shared content" (personal workspace communications) - NOTION_CONNECTOR: "Notion workspace pages and databases" (personal knowledge management) @@ -35,6 +34,7 @@ You are SurfSense, an advanced AI research assistant that provides detailed, wel - TAVILY_API: "Tavily search API results" (personalized search results) - LINKUP_API: "Linkup search API results" (personalized search results) - LUMA_CONNECTOR: "Luma events" +- WEBCRAWLER_CONNECTOR: "Webpages indexed by SurfSense" (personally selected websites) diff --git a/surfsense_backend/app/agents/researcher/utils.py b/surfsense_backend/app/agents/researcher/utils.py index a2c211f28..41d5a1f55 100644 --- a/surfsense_backend/app/agents/researcher/utils.py +++ b/surfsense_backend/app/agents/researcher/utils.py @@ -19,7 +19,6 @@ def get_connector_emoji(connector_name: str) -> str: connector_emojis = { "YOUTUBE_VIDEO": "📹", "EXTENSION": "🧩", - "CRAWLED_URL": "🌐", "FILE": "📄", "SLACK_CONNECTOR": "💬", "NOTION_CONNECTOR": "📘", @@ -34,6 +33,7 @@ def get_connector_emoji(connector_name: str) -> str: "AIRTABLE_CONNECTOR": "🗃️", "LUMA_CONNECTOR": "✨", "ELASTICSEARCH_CONNECTOR": "⚡", + "WEBCRAWLER_CONNECTOR": "🌐", } return connector_emojis.get(connector_name, "🔎") @@ -43,7 +43,6 @@ def get_connector_friendly_name(connector_name: str) -> str: connector_friendly_names = { "YOUTUBE_VIDEO": "YouTube", "EXTENSION": "Browser Extension", - "CRAWLED_URL": "Web Pages", "FILE": "Files", "SLACK_CONNECTOR": "Slack", "NOTION_CONNECTOR": "Notion", @@ -59,6 +58,7 @@ def get_connector_friendly_name(connector_name: str) -> str: "AIRTABLE_CONNECTOR": "Airtable", "LUMA_CONNECTOR": "Luma", "ELASTICSEARCH_CONNECTOR": "Elasticsearch", + "WEBCRAWLER_CONNECTOR": "Web Pages", } return connector_friendly_names.get(connector_name, connector_name) diff --git a/surfsense_backend/app/config/__init__.py b/surfsense_backend/app/config/__init__.py index 7d06643e1..efa51d2db 100644 --- a/surfsense_backend/app/config/__init__.py +++ b/surfsense_backend/app/config/__init__.py @@ -208,9 +208,6 @@ class Config: # LlamaCloud API Key LLAMA_CLOUD_API_KEY = os.getenv("LLAMA_CLOUD_API_KEY") - # Firecrawl API Key - FIRECRAWL_API_KEY = os.getenv("FIRECRAWL_API_KEY", None) - # Litellm TTS Configuration TTS_SERVICE = os.getenv("TTS_SERVICE") TTS_SERVICE_API_BASE = os.getenv("TTS_SERVICE_API_BASE") diff --git a/surfsense_backend/app/connectors/webcrawler_connector.py b/surfsense_backend/app/connectors/webcrawler_connector.py new file mode 100644 index 000000000..9bb72f1ce --- /dev/null +++ b/surfsense_backend/app/connectors/webcrawler_connector.py @@ -0,0 +1,191 @@ +""" +WebCrawler Connector Module + +A module for crawling web pages and extracting content using Firecrawl or AsyncChromiumLoader. +Provides a unified interface for web scraping. +""" + +from typing import Any + +import validators +from firecrawl import AsyncFirecrawlApp +from langchain_community.document_loaders import AsyncChromiumLoader + + +class WebCrawlerConnector: + """Class for crawling web pages and extracting content.""" + + def __init__(self, firecrawl_api_key: str | None = None): + """ + Initialize the WebCrawlerConnector class. + + Args: + firecrawl_api_key: Firecrawl API key (optional, will use AsyncChromiumLoader if not provided) + """ + self.firecrawl_api_key = firecrawl_api_key + self.use_firecrawl = bool(firecrawl_api_key) + + def set_api_key(self, api_key: str) -> None: + """ + Set the Firecrawl API key and enable Firecrawl usage. + + Args: + api_key: Firecrawl API key + """ + self.firecrawl_api_key = api_key + self.use_firecrawl = True + + async def crawl_url( + self, url: str, formats: list[str] | None = None + ) -> tuple[dict[str, Any] | None, str | None]: + """ + Crawl a single URL and extract its content. + + Args: + url: URL to crawl + formats: List of formats to extract (e.g., ["markdown", "html"]) - only for Firecrawl + + Returns: + Tuple containing (crawl result dict, error message or None) + Result dict contains: + - content: Extracted content (markdown or HTML) + - metadata: Page metadata (title, description, etc.) + - source: Original URL + - crawler_type: Type of crawler used + """ + try: + # Validate URL + if not validators.url(url): + return None, f"Invalid URL: {url}" + + if self.use_firecrawl: + result = await self._crawl_with_firecrawl(url, formats) + else: + result = await self._crawl_with_chromium(url) + + return result, None + + except Exception as e: + return None, f"Error crawling URL {url}: {e!s}" + + async def _crawl_with_firecrawl( + self, url: str, formats: list[str] | None = None + ) -> dict[str, Any]: + """ + Crawl URL using Firecrawl. + + Args: + url: URL to crawl + formats: List of formats to extract + + Returns: + Dict containing crawled content and metadata + + Raises: + ValueError: If Firecrawl scraping fails + """ + if not self.firecrawl_api_key: + raise ValueError("Firecrawl API key not set. Call set_api_key() first.") + + firecrawl_app = AsyncFirecrawlApp(api_key=self.firecrawl_api_key) + + # Default to markdown format + if formats is None: + formats = ["markdown"] + + scrape_result = await firecrawl_app.scrape_url(url=url, formats=formats) + + if not scrape_result or not scrape_result.success: + error_msg = ( + scrape_result.error + if scrape_result and hasattr(scrape_result, "error") + else "Unknown error" + ) + raise ValueError(f"Firecrawl failed to scrape URL: {error_msg}") + + # Extract content based on format + content = scrape_result.markdown or scrape_result.html or "" + + # Extract metadata + metadata = scrape_result.metadata if scrape_result.metadata else {} + + return { + "content": content, + "metadata": { + "source": url, + "title": metadata.get("title", url), + "description": metadata.get("description", ""), + "language": metadata.get("language", ""), + "sourceURL": metadata.get("sourceURL", url), + **metadata, + }, + "crawler_type": "firecrawl", + } + + async def _crawl_with_chromium(self, url: str) -> dict[str, Any]: + """ + Crawl URL using AsyncChromiumLoader. + + Args: + url: URL to crawl + + Returns: + Dict containing crawled content and metadata + + Raises: + Exception: If crawling fails + """ + crawl_loader = AsyncChromiumLoader(urls=[url], headless=True) + documents = await crawl_loader.aload() + + if not documents: + raise ValueError(f"Failed to load content from {url}") + + doc = documents[0] + + # Extract basic metadata from the document + metadata = doc.metadata if doc.metadata else {} + + return { + "content": doc.page_content, + "metadata": { + "source": url, + "title": metadata.get("title", url), + **metadata, + }, + "crawler_type": "chromium", + } + + def format_to_structured_document(self, crawl_result: dict[str, Any]) -> str: + """ + Format crawl result as a structured document. + + Args: + crawl_result: Result from crawl_url method + + Returns: + Structured document string + """ + metadata = crawl_result["metadata"] + content = crawl_result["content"] + + document_parts = ["", ""] + + # Add all metadata fields + for key, value in metadata.items(): + document_parts.append(f"{key.upper()}: {value}") + + document_parts.extend( + [ + "", + "", + "FORMAT: markdown", + "TEXT_START", + content, + "TEXT_END", + "", + "", + ] + ) + + return "\n".join(document_parts) diff --git a/surfsense_backend/app/db.py b/surfsense_backend/app/db.py index 4ad31b508..06abb7a39 100644 --- a/surfsense_backend/app/db.py +++ b/surfsense_backend/app/db.py @@ -73,6 +73,7 @@ class SearchSourceConnectorType(str, Enum): AIRTABLE_CONNECTOR = "AIRTABLE_CONNECTOR" LUMA_CONNECTOR = "LUMA_CONNECTOR" ELASTICSEARCH_CONNECTOR = "ELASTICSEARCH_CONNECTOR" + WEBCRAWLER_CONNECTOR = "WEBCRAWLER_CONNECTOR" class ChatType(str, Enum): diff --git a/surfsense_backend/app/routes/documents_routes.py b/surfsense_backend/app/routes/documents_routes.py index 344a2503d..ae9df0cf4 100644 --- a/surfsense_backend/app/routes/documents_routes.py +++ b/surfsense_backend/app/routes/documents_routes.py @@ -65,13 +65,6 @@ async def create_documents( process_extension_document_task.delay( document_dict, request.search_space_id, str(user.id) ) - elif request.document_type == DocumentType.CRAWLED_URL: - from app.tasks.celery_tasks.document_tasks import process_crawled_url_task - - for url in request.content: - process_crawled_url_task.delay( - url, request.search_space_id, str(user.id) - ) elif request.document_type == DocumentType.YOUTUBE_VIDEO: from app.tasks.celery_tasks.document_tasks import process_youtube_video_task diff --git a/surfsense_backend/app/routes/search_source_connectors_routes.py b/surfsense_backend/app/routes/search_source_connectors_routes.py index 4e62035ff..bf397a352 100644 --- a/surfsense_backend/app/routes/search_source_connectors_routes.py +++ b/surfsense_backend/app/routes/search_source_connectors_routes.py @@ -39,6 +39,7 @@ from app.tasks.connector_indexers import ( index_airtable_records, index_clickup_tasks, index_confluence_pages, + index_crawled_urls, index_discord_messages, index_elasticsearch_documents, index_github_repos, @@ -482,6 +483,7 @@ async def index_connector_content( - DISCORD_CONNECTOR: Indexes messages from all accessible Discord channels - LUMA_CONNECTOR: Indexes events from Luma - ELASTICSEARCH_CONNECTOR: Indexes documents from Elasticsearch + - WEBCRAWLER_CONNECTOR: Indexes web pages from crawled websites Args: connector_id: ID of the connector to use @@ -688,6 +690,17 @@ async def index_connector_content( ) response_message = "Elasticsearch indexing started in the background." + elif connector.connector_type == SearchSourceConnectorType.WEBCRAWLER_CONNECTOR: + from app.tasks.celery_tasks.connector_tasks import index_crawled_urls_task + + logger.info( + f"Triggering web pages indexing for connector {connector_id} into search space {search_space_id} from {indexing_from} to {indexing_to}" + ) + index_crawled_urls_task.delay( + connector_id, search_space_id, str(user.id), indexing_from, indexing_to + ) + response_message = "Web page indexing started in the background." + else: raise HTTPException( status_code=400, @@ -1523,3 +1536,64 @@ async def run_elasticsearch_indexing( f"Critical error in run_elasticsearch_indexing for connector {connector_id}: {e}", exc_info=True, ) + + +# Add new helper functions for crawled web page indexing +async def run_web_page_indexing_with_new_session( + connector_id: int, + search_space_id: int, + user_id: str, + start_date: str, + end_date: str, +): + """ + Create a new session and run the Web page indexing task. + This prevents session leaks by creating a dedicated session for the background task. + """ + async with async_session_maker() as session: + await run_web_page_indexing( + session, connector_id, search_space_id, user_id, start_date, end_date + ) + + +async def run_web_page_indexing( + session: AsyncSession, + connector_id: int, + search_space_id: int, + user_id: str, + start_date: str, + end_date: str, +): + """ + Background task to run Web page indexing. + Args: + session: Database session + connector_id: ID of the webcrawler connector + search_space_id: ID of the search space + user_id: ID of the user + start_date: Start date for indexing + end_date: End date for indexing + """ + try: + documents_processed, error_or_warning = await index_crawled_urls( + session=session, + connector_id=connector_id, + search_space_id=search_space_id, + user_id=user_id, + start_date=start_date, + end_date=end_date, + update_last_indexed=False, # Don't update timestamp in the indexing function + ) + + # Only update last_indexed_at if indexing was successful (either new docs or updated docs) + if documents_processed > 0: + await update_connector_last_indexed(session, connector_id) + logger.info( + f"Web page indexing completed successfully: {documents_processed} documents processed" + ) + else: + logger.error( + f"Web page indexing failed or no documents processed: {error_or_warning}" + ) + except Exception as e: + logger.error(f"Error in background Web page indexing task: {e!s}") diff --git a/surfsense_backend/app/services/connector_service.py b/surfsense_backend/app/services/connector_service.py index 28f70d285..3445d69f7 100644 --- a/surfsense_backend/app/services/connector_service.py +++ b/surfsense_backend/app/services/connector_service.py @@ -70,6 +70,13 @@ class ConnectorService: """ Search for crawled URLs 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) """ @@ -109,15 +116,43 @@ class ConnectorService: document = chunk.get("document", {}) metadata = document.get("metadata", {}) - # Create a source entry + # Extract webcrawler-specific metadata + url = metadata.get("source", metadata.get("url", "")) + title = document.get( + "title", metadata.get("title", "Untitled Document") + ) + description = metadata.get("description", "") + language = metadata.get("language", "") + last_crawled_at = metadata.get("last_crawled_at", "") + + # Build description with crawler info + content_preview = chunk.get("content", "") + if not description and content_preview: + # Use content preview if no description + description = content_preview[:200] + if len(content_preview) > 200: + description += "..." + + # Add crawler metadata to description if available + info_parts = [] + if language: + info_parts.append(f"Language: {language}") + if last_crawled_at: + info_parts.append(f"Last crawled: {last_crawled_at}") + + if info_parts: + if description: + description += f" | {' | '.join(info_parts)}" + else: + description = " | ".join(info_parts) + source = { "id": chunk.get("chunk_id", self.source_id_counter), - "title": document.get("title", "Untitled Document"), - "description": metadata.get( - "og:description", - metadata.get("ogDescription", chunk.get("content", "")), - ), - "url": metadata.get("url", ""), + "title": title, + "description": description, + "url": url, + "language": language, + "last_crawled_at": last_crawled_at, } self.source_id_counter += 1 diff --git a/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py b/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py index 5e6907499..b735741fe 100644 --- a/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py +++ b/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py @@ -600,3 +600,46 @@ async def _index_elasticsearch_documents( await run_elasticsearch_indexing( session, connector_id, search_space_id, user_id, start_date, end_date ) + + +@celery_app.task(name="index_crawled_urls", bind=True) +def index_crawled_urls_task( + self, + connector_id: int, + search_space_id: int, + user_id: str, + start_date: str, + end_date: str, +): + """Celery task to index Web page Urls.""" + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + try: + loop.run_until_complete( + _index_crawled_urls( + connector_id, search_space_id, user_id, start_date, end_date + ) + ) + finally: + loop.close() + + +async def _index_crawled_urls( + connector_id: int, + search_space_id: int, + user_id: str, + start_date: str, + end_date: str, +): + """Index Web page Urls with new session.""" + from app.routes.search_source_connectors_routes import ( + run_web_page_indexing, + ) + + async with get_celery_session_maker()() as session: + await run_web_page_indexing( + session, connector_id, search_space_id, user_id, start_date, end_date + ) diff --git a/surfsense_backend/app/tasks/celery_tasks/document_tasks.py b/surfsense_backend/app/tasks/celery_tasks/document_tasks.py index 73af21550..5cf5a662a 100644 --- a/surfsense_backend/app/tasks/celery_tasks/document_tasks.py +++ b/surfsense_backend/app/tasks/celery_tasks/document_tasks.py @@ -9,7 +9,6 @@ from app.celery_app import celery_app from app.config import config from app.services.task_logging_service import TaskLoggingService from app.tasks.document_processors import ( - add_crawled_url_document, add_extension_received_document, add_youtube_video_document, ) @@ -120,71 +119,6 @@ async def _process_extension_document( raise -@celery_app.task(name="process_crawled_url", bind=True) -def process_crawled_url_task(self, url: str, search_space_id: int, user_id: str): - """ - Celery task to process crawled URL. - - Args: - url: URL to crawl and process - search_space_id: ID of the search space - user_id: ID of the user - """ - import asyncio - - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - - try: - loop.run_until_complete(_process_crawled_url(url, search_space_id, user_id)) - finally: - loop.close() - - -async def _process_crawled_url(url: str, search_space_id: int, user_id: str): - """Process crawled URL with new session.""" - async with get_celery_session_maker()() as session: - task_logger = TaskLoggingService(session, search_space_id) - - log_entry = await task_logger.log_task_start( - task_name="process_crawled_url", - source="document_processor", - message=f"Starting URL crawling and processing for: {url}", - metadata={"document_type": "CRAWLED_URL", "url": url, "user_id": user_id}, - ) - - try: - result = await add_crawled_url_document( - session, url, search_space_id, user_id - ) - - if result: - await task_logger.log_task_success( - log_entry, - f"Successfully crawled and processed URL: {url}", - { - "document_id": result.id, - "title": result.title, - "content_hash": result.content_hash, - }, - ) - else: - await task_logger.log_task_success( - log_entry, - f"URL document already exists (duplicate): {url}", - {"duplicate_detected": True}, - ) - except Exception as e: - await task_logger.log_task_failure( - log_entry, - f"Failed to crawl URL: {url}", - str(e), - {"error_type": type(e).__name__}, - ) - logger.error(f"Error processing crawled URL: {e!s}") - raise - - @celery_app.task(name="process_youtube_video", bind=True) def process_youtube_video_task(self, url: str, search_space_id: int, user_id: str): """ diff --git a/surfsense_backend/app/tasks/celery_tasks/schedule_checker_task.py b/surfsense_backend/app/tasks/celery_tasks/schedule_checker_task.py index 39d6bf840..dbc326406 100644 --- a/surfsense_backend/app/tasks/celery_tasks/schedule_checker_task.py +++ b/surfsense_backend/app/tasks/celery_tasks/schedule_checker_task.py @@ -67,6 +67,7 @@ async def _check_and_trigger_schedules(): index_airtable_records_task, index_clickup_tasks_task, index_confluence_pages_task, + index_crawled_urls_task, index_discord_messages_task, index_elasticsearch_documents_task, index_github_repos_task, @@ -94,6 +95,7 @@ async def _check_and_trigger_schedules(): SearchSourceConnectorType.DISCORD_CONNECTOR: index_discord_messages_task, SearchSourceConnectorType.LUMA_CONNECTOR: index_luma_events_task, SearchSourceConnectorType.ELASTICSEARCH_CONNECTOR: index_elasticsearch_documents_task, + SearchSourceConnectorType.WEBCRAWLER_CONNECTOR: index_crawled_urls_task, } # Trigger indexing for each due connector diff --git a/surfsense_backend/app/tasks/connector_indexers/__init__.py b/surfsense_backend/app/tasks/connector_indexers/__init__.py index 766506f70..f62739679 100644 --- a/surfsense_backend/app/tasks/connector_indexers/__init__.py +++ b/surfsense_backend/app/tasks/connector_indexers/__init__.py @@ -17,6 +17,7 @@ Available indexers: - Google Gmail: Index messages from Google Gmail - Google Calendar: Index events from Google Calendar - Luma: Index events from Luma +- Webcrawler: Index crawled URLs - Elasticsearch: Index documents from Elasticsearch instances """ @@ -41,6 +42,7 @@ from .luma_indexer import index_luma_events # Documentation and knowledge management from .notion_indexer import index_notion_pages from .slack_indexer import index_slack_messages +from .webcrawler_indexer import index_crawled_urls __all__ = [ # noqa: RUF022 "index_airtable_records", @@ -58,6 +60,7 @@ __all__ = [ # noqa: RUF022 "index_linear_issues", # Documentation and knowledge management "index_notion_pages", + "index_crawled_urls", # Communication platforms "index_slack_messages", "index_google_gmail_messages", diff --git a/surfsense_backend/app/tasks/connector_indexers/webcrawler_indexer.py b/surfsense_backend/app/tasks/connector_indexers/webcrawler_indexer.py new file mode 100644 index 000000000..6a6cb0ef8 --- /dev/null +++ b/surfsense_backend/app/tasks/connector_indexers/webcrawler_indexer.py @@ -0,0 +1,450 @@ +""" +Webcrawler connector indexer. +""" + +from datetime import datetime + +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import config +from app.connectors.webcrawler_connector import WebCrawlerConnector +from app.db import Document, DocumentType, SearchSourceConnectorType +from app.services.llm_service import get_user_long_context_llm +from app.services.task_logging_service import TaskLoggingService +from app.utils.document_converters import ( + create_document_chunks, + generate_content_hash, + generate_document_summary, + generate_unique_identifier_hash, +) + +from .base import ( + check_document_by_unique_identifier, + get_connector_by_id, + logger, + update_connector_last_indexed, +) + + +async def index_crawled_urls( + 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 web page URLs. + + Args: + session: Database session + connector_id: ID of the webcrawler connector + search_space_id: ID of the search space to store documents in + user_id: User ID + start_date: Start date for filtering (YYYY-MM-DD format) - optional + end_date: End date for filtering (YYYY-MM-DD format) - optional + 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="crawled_url_indexing", + source="connector_indexing_task", + message=f"Starting web page URL indexing for connector {connector_id}", + metadata={ + "connector_id": connector_id, + "user_id": str(user_id), + "start_date": start_date, + "end_date": end_date, + }, + ) + + try: + # Get the connector + await task_logger.log_task_progress( + log_entry, + f"Retrieving webcrawler connector {connector_id} from database", + {"stage": "connector_retrieval"}, + ) + + # Get the connector from the database + connector = await get_connector_by_id( + session, connector_id, SearchSourceConnectorType.WEBCRAWLER_CONNECTOR + ) + + if not connector: + await task_logger.log_task_failure( + log_entry, + f"Connector with ID {connector_id} not found or is not a webcrawler connector", + "Connector not found", + {"error_type": "ConnectorNotFound"}, + ) + return ( + 0, + f"Connector with ID {connector_id} not found or is not a webcrawler connector", + ) + + # Get the Firecrawl API key from the connector config (optional) + api_key = connector.config.get("FIRECRAWL_API_KEY") + + # Get URLs from connector config + initial_urls = connector.config.get("INITIAL_URLS", "") + if isinstance(initial_urls, str): + urls = [url.strip() for url in initial_urls.split("\n") if url.strip()] + elif isinstance(initial_urls, list): + urls = [url.strip() for url in initial_urls if url.strip()] + else: + urls = [] + + logger.info( + f"Starting crawled web page indexing for connector {connector_id} with {len(urls)} URLs" + ) + + # Initialize webcrawler client + await task_logger.log_task_progress( + log_entry, + f"Initializing webcrawler client for connector {connector_id}", + { + "stage": "client_initialization", + "use_firecrawl": bool(api_key), + }, + ) + + crawler = WebCrawlerConnector(firecrawl_api_key=api_key) + + # Validate URLs + if not urls: + await task_logger.log_task_failure( + log_entry, + "No URLs provided for indexing", + "Empty URL list", + {"error_type": "ValidationError"}, + ) + return 0, "No URLs provided for indexing" + + await task_logger.log_task_progress( + log_entry, + f"Starting to crawl {len(urls)} URLs", + { + "stage": "crawling", + "total_urls": len(urls), + }, + ) + + documents_indexed = 0 + documents_updated = 0 + documents_skipped = 0 + failed_urls = [] + + for idx, url in enumerate(urls, 1): + try: + logger.info(f"Processing URL {idx}/{len(urls)}: {url}") + + await task_logger.log_task_progress( + log_entry, + f"Crawling URL {idx}/{len(urls)}: {url}", + { + "stage": "crawling_url", + "url_index": idx, + "url": url, + }, + ) + + # Crawl the URL + crawl_result, error = await crawler.crawl_url(url) + + if error or not crawl_result: + logger.warning(f"Failed to crawl URL {url}: {error}") + failed_urls.append((url, error or "Unknown error")) + continue + + # Extract content and metadata + content = crawl_result.get("content", "") + metadata = crawl_result.get("metadata", {}) + crawler_type = crawl_result.get("crawler_type", "unknown") + + if not content.strip(): + logger.warning(f"Skipping URL with no content: {url}") + failed_urls.append((url, "No content extracted")) + documents_skipped += 1 + continue + + # Format content as structured document + structured_document = crawler.format_to_structured_document( + crawl_result + ) + + # Generate unique identifier hash for this URL + unique_identifier_hash = generate_unique_identifier_hash( + DocumentType.CRAWLED_URL, url, search_space_id + ) + + # Generate content hash + # TODO: To fix this by not including dynamic content like date, time, etc. + content_hash = generate_content_hash( + structured_document, search_space_id + ) + + # Check if document with this unique identifier already exists + existing_document = await check_document_by_unique_identifier( + session, unique_identifier_hash + ) + + # Extract useful metadata + title = metadata.get("title", url) + description = metadata.get("description", "") + language = metadata.get("language", "") + + if existing_document: + # Document exists - check if content has changed + if existing_document.content_hash == content_hash: + logger.info(f"Document for URL {url} unchanged. Skipping.") + documents_skipped += 1 + continue + else: + # Content has changed - update the existing document + logger.info( + f"Content changed for URL {url}. Updating document." + ) + + # Generate summary with metadata + user_llm = await get_user_long_context_llm( + session, user_id, search_space_id + ) + + if user_llm: + document_metadata = { + "url": url, + "title": title, + "description": description, + "language": language, + "document_type": "Crawled URL", + "crawler_type": crawler_type, + } + ( + summary_content, + summary_embedding, + ) = await generate_document_summary( + structured_document, user_llm, document_metadata + ) + else: + # Fallback to simple summary if no LLM configured + summary_content = f"Crawled URL: {title}\n\n" + summary_content += f"URL: {url}\n" + if description: + summary_content += f"Description: {description}\n" + if language: + summary_content += f"Language: {language}\n" + summary_content += f"Crawler: {crawler_type}\n\n" + + # Add content preview + content_preview = content[:1000] + if len(content) > 1000: + content_preview += "..." + summary_content += f"Content Preview:\n{content_preview}\n" + + summary_embedding = config.embedding_model_instance.embed( + summary_content + ) + + # Process chunks + chunks = await create_document_chunks(content) + + # Update existing document + existing_document.title = title + existing_document.content = summary_content + existing_document.content_hash = content_hash + existing_document.embedding = summary_embedding + existing_document.document_metadata = { + **metadata, + "crawler_type": crawler_type, + "last_crawled_at": datetime.now().strftime( + "%Y-%m-%d %H:%M:%S" + ), + } + existing_document.chunks = chunks + + documents_updated += 1 + logger.info(f"Successfully updated URL {url}") + continue + + # Document doesn't exist - create new one + # Generate summary with metadata + user_llm = await get_user_long_context_llm( + session, user_id, search_space_id + ) + + if user_llm: + document_metadata = { + "url": url, + "title": title, + "description": description, + "language": language, + "document_type": "Crawled URL", + "crawler_type": crawler_type, + } + ( + summary_content, + summary_embedding, + ) = await generate_document_summary( + structured_document, user_llm, document_metadata + ) + else: + # Fallback to simple summary if no LLM configured + summary_content = f"Crawled URL: {title}\n\n" + summary_content += f"URL: {url}\n" + if description: + summary_content += f"Description: {description}\n" + if language: + summary_content += f"Language: {language}\n" + summary_content += f"Crawler: {crawler_type}\n\n" + + # Add content preview + content_preview = content[:1000] + if len(content) > 1000: + content_preview += "..." + summary_content += f"Content Preview:\n{content_preview}\n" + + summary_embedding = config.embedding_model_instance.embed( + summary_content + ) + + chunks = await create_document_chunks(content) + + document = Document( + search_space_id=search_space_id, + title=title, + document_type=DocumentType.CRAWLED_URL, + document_metadata={ + **metadata, + "crawler_type": crawler_type, + "indexed_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + }, + content=summary_content, + content_hash=content_hash, + unique_identifier_hash=unique_identifier_hash, + embedding=summary_embedding, + chunks=chunks, + ) + + session.add(document) + documents_indexed += 1 + logger.info(f"Successfully indexed new URL {url}") + + # Batch commit every 10 documents + if (documents_indexed + documents_updated) % 10 == 0: + logger.info( + f"Committing batch: {documents_indexed + documents_updated} URLs processed so far" + ) + await session.commit() + + except Exception as e: + logger.error( + f"Error processing URL {url}: {e!s}", + exc_info=True, + ) + failed_urls.append((url, str(e))) + continue + + total_processed = documents_indexed + documents_updated + + if total_processed > 0: + await update_connector_last_indexed(session, connector, update_last_indexed) + + # Final commit for any remaining documents not yet committed in batches + logger.info( + f"Final commit: Total {documents_indexed} new, {documents_updated} updated URLs processed" + ) + await session.commit() + + # Build result message + result_message = None + if failed_urls: + failed_summary = "; ".join( + [f"{url}: {error}" for url, error in failed_urls[:5]] + ) + if len(failed_urls) > 5: + failed_summary += f" (and {len(failed_urls) - 5} more)" + result_message = ( + f"Completed with {len(failed_urls)} failures: {failed_summary}" + ) + + await task_logger.log_task_success( + log_entry, + f"Successfully completed crawled web page indexing for connector {connector_id}", + { + "urls_processed": total_processed, + "documents_indexed": documents_indexed, + "documents_updated": documents_updated, + "documents_skipped": documents_skipped, + "failed_urls_count": len(failed_urls), + }, + ) + + logger.info( + f"Web page indexing completed: {documents_indexed} new, " + f"{documents_updated} updated, {documents_skipped} skipped, " + f"{len(failed_urls)} failed" + ) + return total_processed, result_message + + except SQLAlchemyError as db_error: + await session.rollback() + await task_logger.log_task_failure( + log_entry, + f"Database error during web page indexing for connector {connector_id}", + str(db_error), + {"error_type": "SQLAlchemyError"}, + ) + logger.error(f"Database error: {db_error!s}", exc_info=True) + return 0, f"Database error: {db_error!s}" + except Exception as e: + await session.rollback() + await task_logger.log_task_failure( + log_entry, + f"Failed to index web page URLs for connector {connector_id}", + str(e), + {"error_type": type(e).__name__}, + ) + logger.error(f"Failed to index web page URLs: {e!s}", exc_info=True) + return 0, f"Failed to index web page URLs: {e!s}" + + +async def get_crawled_url_documents( + session: AsyncSession, + search_space_id: int, + connector_id: int | None = None, +) -> list[Document]: + """ + Get all crawled URL documents for a search space. + + Args: + session: Database session + search_space_id: ID of the search space + connector_id: Optional connector ID to filter by + + Returns: + List of Document objects + """ + from sqlalchemy import select + + query = select(Document).filter( + Document.search_space_id == search_space_id, + Document.document_type == DocumentType.CRAWLED_URL, + ) + + if connector_id: + # Filter by connector if needed - you might need to add a connector_id field to Document + # or filter by some other means depending on your schema + pass + + result = await session.execute(query) + documents = result.scalars().all() + return list(documents) diff --git a/surfsense_backend/app/tasks/document_processors/__init__.py b/surfsense_backend/app/tasks/document_processors/__init__.py index a238ac877..e70c41cb4 100644 --- a/surfsense_backend/app/tasks/document_processors/__init__.py +++ b/surfsense_backend/app/tasks/document_processors/__init__.py @@ -6,7 +6,6 @@ and sources. Each processor is responsible for handling a specific type of docum processing task in the background. Available processors: -- URL crawler: Process web pages from URLs - Extension processor: Handle documents from browser extension - Markdown processor: Process markdown files - File processors: Handle files using different ETL services (Unstructured, LlamaCloud, Docling) @@ -26,14 +25,11 @@ from .file_processors import ( # Markdown processor from .markdown_processor import add_received_markdown_file_document -from .url_crawler import add_crawled_url_document # YouTube processor from .youtube_processor import add_youtube_video_document __all__ = [ - # URL processing - "add_crawled_url_document", # Extension processing "add_extension_received_document", "add_received_file_document_using_docling", diff --git a/surfsense_backend/app/tasks/document_processors/url_crawler.py b/surfsense_backend/app/tasks/document_processors/url_crawler.py deleted file mode 100644 index ce2764897..000000000 --- a/surfsense_backend/app/tasks/document_processors/url_crawler.py +++ /dev/null @@ -1,330 +0,0 @@ -""" -URL crawler document processor. -""" - -import logging - -import validators -from firecrawl import AsyncFirecrawlApp -from langchain_community.document_loaders import AsyncChromiumLoader -from langchain_core.documents import Document as LangchainDocument -from sqlalchemy.exc import SQLAlchemyError -from sqlalchemy.ext.asyncio import AsyncSession - -from app.config import config -from app.db import Document, DocumentType -from app.services.llm_service import get_user_long_context_llm -from app.services.task_logging_service import TaskLoggingService -from app.utils.document_converters import ( - create_document_chunks, - generate_content_hash, - generate_document_summary, - generate_unique_identifier_hash, -) - -from .base import ( - check_document_by_unique_identifier, - md, -) - - -async def add_crawled_url_document( - session: AsyncSession, url: str, search_space_id: int, user_id: str -) -> Document | None: - """ - Process and store a document from a crawled URL. - - Args: - session: Database session - url: URL to crawl - search_space_id: ID of the search space - user_id: ID of the user - - Returns: - Document object if successful, None if failed - """ - task_logger = TaskLoggingService(session, search_space_id) - - # Log task start - log_entry = await task_logger.log_task_start( - task_name="crawl_url_document", - source="background_task", - message=f"Starting URL crawling process for: {url}", - metadata={"url": url, "user_id": str(user_id)}, - ) - - try: - # URL validation step - await task_logger.log_task_progress( - log_entry, f"Validating URL: {url}", {"stage": "validation"} - ) - - if not validators.url(url): - raise ValueError(f"Url {url} is not a valid URL address") - - # Set up crawler - await task_logger.log_task_progress( - log_entry, - f"Setting up crawler for URL: {url}", - { - "stage": "crawler_setup", - "firecrawl_available": bool(config.FIRECRAWL_API_KEY), - }, - ) - - use_firecrawl = bool(config.FIRECRAWL_API_KEY) - - if use_firecrawl: - # Use Firecrawl SDK directly - firecrawl_app = AsyncFirecrawlApp(api_key=config.FIRECRAWL_API_KEY) - else: - crawl_loader = AsyncChromiumLoader(urls=[url], headless=True) - - # Perform crawling - await task_logger.log_task_progress( - log_entry, - f"Crawling URL content: {url}", - { - "stage": "crawling", - "crawler_type": "AsyncFirecrawlApp" - if use_firecrawl - else "AsyncChromiumLoader", - }, - ) - - if use_firecrawl: - # Use async Firecrawl SDK with v1 API - properly awaited - scrape_result = await firecrawl_app.scrape_url( - url=url, formats=["markdown"] - ) - - # scrape_result is a Pydantic ScrapeResponse object - # Access attributes directly - if scrape_result and scrape_result.success: - # Extract markdown content - markdown_content = scrape_result.markdown or "" - - # Extract metadata - this is a DICT - metadata = scrape_result.metadata if scrape_result.metadata else {} - - # Convert to LangChain Document format - url_crawled = [ - LangchainDocument( - page_content=markdown_content, - metadata={ - "source": url, - "title": metadata.get("title", url), - "description": metadata.get("description", ""), - "language": metadata.get("language", ""), - "sourceURL": metadata.get("sourceURL", url), - **metadata, # Include all other metadata fields - }, - ) - ] - content_in_markdown = url_crawled[0].page_content - else: - error_msg = ( - scrape_result.error - if scrape_result and hasattr(scrape_result, "error") - else "Unknown error" - ) - raise ValueError(f"Firecrawl failed to scrape URL: {error_msg}") - else: - # Use AsyncChromiumLoader as fallback - url_crawled = await crawl_loader.aload() - content_in_markdown = md.transform_documents(url_crawled)[0].page_content - - # Format document - await task_logger.log_task_progress( - log_entry, - f"Processing crawled content from: {url}", - {"stage": "content_processing", "content_length": len(content_in_markdown)}, - ) - - # Format document metadata in a more maintainable way - metadata_sections = [ - ( - "METADATA", - [ - f"{key.upper()}: {value}" - for key, value in url_crawled[0].metadata.items() - ], - ), - ( - "CONTENT", - ["FORMAT: markdown", "TEXT_START", content_in_markdown, "TEXT_END"], - ), - ] - - # Build the document string more efficiently - document_parts = [] - document_parts.append("") - - for section_title, section_content in metadata_sections: - document_parts.append(f"<{section_title}>") - document_parts.extend(section_content) - document_parts.append(f"") - - document_parts.append("") - combined_document_string = "\n".join(document_parts) - - # Generate unique identifier hash for this URL - unique_identifier_hash = generate_unique_identifier_hash( - DocumentType.CRAWLED_URL, url, search_space_id - ) - - # Generate content hash - content_hash = generate_content_hash(combined_document_string, search_space_id) - - # Check if document with this unique identifier already exists - await task_logger.log_task_progress( - log_entry, - f"Checking for existing URL: {url}", - {"stage": "duplicate_check", "url": url}, - ) - - existing_document = await check_document_by_unique_identifier( - session, unique_identifier_hash - ) - - if existing_document: - # Document exists - check if content has changed - if existing_document.content_hash == content_hash: - await task_logger.log_task_success( - log_entry, - f"URL document unchanged: {url}", - { - "duplicate_detected": True, - "existing_document_id": existing_document.id, - }, - ) - logging.info(f"Document for URL {url} unchanged. Skipping.") - return existing_document - else: - # Content has changed - update the existing document - logging.info(f"Content changed for URL {url}. Updating document.") - await task_logger.log_task_progress( - log_entry, - f"Updating URL document: {url}", - {"stage": "document_update", "url": url}, - ) - - # Get LLM for summary generation (needed for both create and update) - await task_logger.log_task_progress( - log_entry, - f"Preparing for summary generation: {url}", - {"stage": "llm_setup"}, - ) - - # Get user's long context LLM - user_llm = await get_user_long_context_llm(session, user_id, search_space_id) - if not user_llm: - raise RuntimeError( - f"No long context LLM configured for user {user_id} in search space {search_space_id}" - ) - - # Generate summary - await task_logger.log_task_progress( - log_entry, - f"Generating summary for URL content: {url}", - {"stage": "summary_generation"}, - ) - - # Generate summary with metadata - document_metadata = { - "url": url, - "title": url_crawled[0].metadata.get("title", url), - "document_type": "Crawled URL Document", - "crawler_type": "FirecrawlApp" if use_firecrawl else "AsyncChromiumLoader", - } - summary_content, summary_embedding = await generate_document_summary( - combined_document_string, user_llm, document_metadata - ) - - # Process chunks - await task_logger.log_task_progress( - log_entry, - f"Processing content chunks for URL: {url}", - {"stage": "chunk_processing"}, - ) - - chunks = await create_document_chunks(content_in_markdown) - - # Update or create document - if existing_document: - # Update existing document - await task_logger.log_task_progress( - log_entry, - f"Updating document in database for URL: {url}", - {"stage": "document_update", "chunks_count": len(chunks)}, - ) - - existing_document.title = url_crawled[0].metadata.get( - "title", url_crawled[0].metadata.get("source", url) - ) - existing_document.content = summary_content - existing_document.content_hash = content_hash - existing_document.embedding = summary_embedding - existing_document.document_metadata = url_crawled[0].metadata - existing_document.chunks = chunks - - document = existing_document - else: - # Create new document - await task_logger.log_task_progress( - log_entry, - f"Creating document in database for URL: {url}", - {"stage": "document_creation", "chunks_count": len(chunks)}, - ) - - document = Document( - search_space_id=search_space_id, - title=url_crawled[0].metadata.get( - "title", url_crawled[0].metadata.get("source", url) - ), - document_type=DocumentType.CRAWLED_URL, - document_metadata=url_crawled[0].metadata, - content=summary_content, - embedding=summary_embedding, - chunks=chunks, - content_hash=content_hash, - unique_identifier_hash=unique_identifier_hash, - ) - - session.add(document) - await session.commit() - await session.refresh(document) - - # Log success - await task_logger.log_task_success( - log_entry, - f"Successfully crawled and processed URL: {url}", - { - "document_id": document.id, - "title": document.title, - "content_hash": content_hash, - "chunks_count": len(chunks), - "summary_length": len(summary_content), - }, - ) - - return document - - except SQLAlchemyError as db_error: - await session.rollback() - await task_logger.log_task_failure( - log_entry, - f"Database error while processing URL: {url}", - str(db_error), - {"error_type": "SQLAlchemyError"}, - ) - raise db_error - except Exception as e: - await session.rollback() - await task_logger.log_task_failure( - log_entry, - f"Failed to crawl URL: {url}", - str(e), - {"error_type": type(e).__name__}, - ) - raise RuntimeError(f"Failed to crawl URL: {e!s}") from e diff --git a/surfsense_backend/app/utils/periodic_scheduler.py b/surfsense_backend/app/utils/periodic_scheduler.py index 225425714..e33661d65 100644 --- a/surfsense_backend/app/utils/periodic_scheduler.py +++ b/surfsense_backend/app/utils/periodic_scheduler.py @@ -31,6 +31,7 @@ CONNECTOR_TASK_MAP = { SearchSourceConnectorType.DISCORD_CONNECTOR: "index_discord_messages", SearchSourceConnectorType.LUMA_CONNECTOR: "index_luma_events", SearchSourceConnectorType.ELASTICSEARCH_CONNECTOR: "index_elasticsearch_documents", + SearchSourceConnectorType.WEBCRAWLER_CONNECTOR: "index_crawled_urls", } @@ -69,6 +70,7 @@ def create_periodic_schedule( index_airtable_records_task, index_clickup_tasks_task, index_confluence_pages_task, + index_crawled_urls_task, index_discord_messages_task, index_elasticsearch_documents_task, index_github_repos_task, @@ -96,6 +98,7 @@ def create_periodic_schedule( SearchSourceConnectorType.DISCORD_CONNECTOR: index_discord_messages_task, SearchSourceConnectorType.LUMA_CONNECTOR: index_luma_events_task, SearchSourceConnectorType.ELASTICSEARCH_CONNECTOR: index_elasticsearch_documents_task, + SearchSourceConnectorType.WEBCRAWLER_CONNECTOR: index_crawled_urls_task, } # Trigger the first run immediately diff --git a/surfsense_backend/app/utils/validators.py b/surfsense_backend/app/utils/validators.py index a8460cd14..6b69fb3e1 100644 --- a/surfsense_backend/app/utils/validators.py +++ b/surfsense_backend/app/utils/validators.py @@ -469,6 +469,22 @@ def validate_connector_config( if not isinstance(value, list) or not value: raise ValueError(f"{field_name} must be a non-empty list of strings") + def validate_firecrawl_api_key_format() -> None: + """Validate Firecrawl API key format if provided.""" + api_key = config.get("FIRECRAWL_API_KEY", "") + if api_key and api_key.strip() and not api_key.strip().startswith("fc-"): + raise ValueError( + "Firecrawl API key should start with 'fc-'. Please verify your API key." + ) + + def validate_initial_urls() -> None: + initial_urls = config.get("INITIAL_URLS", "") + if initial_urls and initial_urls.strip(): + urls = [url.strip() for url in initial_urls.split("\n") if url.strip()] + for url in urls: + if not validators.url(url): + raise ValueError(f"Invalid URL format in INITIAL_URLS: {url}") + # Lookup table for connector validation rules connector_rules = { "SERPER_API": {"required": ["SERPER_API_KEY"], "validators": {}}, @@ -550,6 +566,14 @@ def validate_connector_config( # "validators": {} # }, "LUMA_CONNECTOR": {"required": ["LUMA_API_KEY"], "validators": {}}, + "WEBCRAWLER_CONNECTOR": { + "required": [], # No required fields - API key is optional + "optional": ["FIRECRAWL_API_KEY", "INITIAL_URLS"], + "validators": { + "FIRECRAWL_API_KEY": lambda: validate_firecrawl_api_key_format(), + "INITIAL_URLS": lambda: validate_initial_urls(), + }, + }, } rules = connector_rules.get(connector_type_str) diff --git a/surfsense_web/app/dashboard/[search_space_id]/connectors/[connector_id]/edit/page.tsx b/surfsense_web/app/dashboard/[search_space_id]/connectors/[connector_id]/edit/page.tsx index f09069521..ac580bbfc 100644 --- a/surfsense_web/app/dashboard/[search_space_id]/connectors/[connector_id]/edit/page.tsx +++ b/surfsense_web/app/dashboard/[search_space_id]/connectors/[connector_id]/edit/page.tsx @@ -18,7 +18,16 @@ import { CardHeader, CardTitle, } from "@/components/ui/card"; -import { Form } from "@/components/ui/form"; +import { + Form, + FormControl, + FormDescription, + FormField, + FormItem, + FormLabel, + FormMessage, +} from "@/components/ui/form"; +import { Textarea } from "@/components/ui/textarea"; import { getConnectorIcon } from "@/contracts/enums/connectorIcons"; import { useConnectorEditPage } from "@/hooks/use-connector-edit-page"; // Import Utils, Types, Hook, and Components @@ -282,6 +291,40 @@ export default function EditConnectorPage() { placeholder="Your Elasticsearch API Key" /> )} + + {/* == Webcrawler == */} + {connector.connector_type === "WEBCRAWLER_CONNECTOR" && ( +
+ + ( + + URLs to Crawl + +