From ab6be6cbda961cac9c1e9d75fb1ff0e66406291f Mon Sep 17 00:00:00 2001 From: CREDO23 Date: Fri, 3 Jul 2026 11:53:33 +0200 Subject: [PATCH] refactor(connectors): remove legacy web-crawler KB indexer paths (keep enums) --- .../runtime/connector_searchable_types.py | 1 - surfsense_backend/app/celery_app.py | 1 - .../routes/search_source_connectors_routes.py | 80 --- .../app/services/connector_service.py | 90 --- .../app/tasks/celery_tasks/connector_tasks.py | 39 -- .../celery_tasks/schedule_checker_task.py | 36 - .../app/tasks/connector_indexers/__init__.py | 2 - .../connector_indexers/webcrawler_indexer.py | 661 ------------------ .../app/utils/periodic_scheduler.py | 17 - surfsense_backend/app/utils/validators.py | 15 - .../app/utils/webcrawler_utils.py | 28 - .../scripts/e2e_phase3_crawl_billing.py | 176 +---- .../test_web_crawl_billing.py | 306 -------- .../tests/unit/utils/test_validators.py | 8 +- 14 files changed, 4 insertions(+), 1456 deletions(-) delete mode 100644 surfsense_backend/app/tasks/connector_indexers/webcrawler_indexer.py delete mode 100644 surfsense_backend/app/utils/webcrawler_utils.py delete mode 100644 surfsense_backend/tests/unit/connector_indexers/test_web_crawl_billing.py diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/runtime/connector_searchable_types.py b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/runtime/connector_searchable_types.py index 09fa0756a..339154060 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/runtime/connector_searchable_types.py +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/runtime/connector_searchable_types.py @@ -40,7 +40,6 @@ _CONNECTOR_TYPE_TO_SEARCHABLE: dict[str, str] = { "AIRTABLE_CONNECTOR": "AIRTABLE_CONNECTOR", "LUMA_CONNECTOR": "LUMA_CONNECTOR", "ELASTICSEARCH_CONNECTOR": "ELASTICSEARCH_CONNECTOR", - "WEBCRAWLER_CONNECTOR": "CRAWLED_URL", # Maps to document type "BOOKSTACK_CONNECTOR": "BOOKSTACK_CONNECTOR", "CIRCLEBACK_CONNECTOR": "CIRCLEBACK", # Connector type differs from document type "OBSIDIAN_CONNECTOR": "OBSIDIAN_CONNECTOR", diff --git a/surfsense_backend/app/celery_app.py b/surfsense_backend/app/celery_app.py index 331ed0f40..471d80197 100644 --- a/surfsense_backend/app/celery_app.py +++ b/surfsense_backend/app/celery_app.py @@ -244,7 +244,6 @@ celery_app.conf.update( "index_google_gmail_messages": {"queue": CONNECTORS_QUEUE}, "index_google_drive_files": {"queue": CONNECTORS_QUEUE}, "index_elasticsearch_documents": {"queue": CONNECTORS_QUEUE}, - "index_crawled_urls": {"queue": CONNECTORS_QUEUE}, "index_bookstack_pages": {"queue": CONNECTORS_QUEUE}, "index_composio_connector": {"queue": CONNECTORS_QUEUE}, "index_obsidian_attachment": {"queue": CONNECTORS_QUEUE}, diff --git a/surfsense_backend/app/routes/search_source_connectors_routes.py b/surfsense_backend/app/routes/search_source_connectors_routes.py index 33da72417..2bdf29756 100644 --- a/surfsense_backend/app/routes/search_source_connectors_routes.py +++ b/surfsense_backend/app/routes/search_source_connectors_routes.py @@ -1051,34 +1051,6 @@ 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 - from app.utils.webcrawler_utils import parse_webcrawler_urls - - # Check if URLs are configured before triggering indexing - connector_config = connector.config or {} - urls = parse_webcrawler_urls(connector_config.get("INITIAL_URLS")) - - if not urls: - # URLs are optional - skip indexing gracefully - logger.info( - f"Webcrawler connector {connector_id} has no URLs configured, skipping indexing" - ) - response_message = "No URLs configured for this connector. Add URLs in the connector settings to enable indexing." - indexing_started = False - else: - logger.info( - f"Triggering web pages indexing for connector {connector_id} into workspace {workspace_id} from {indexing_from} to {indexing_to}" - ) - index_crawled_urls_task.delay( - connector_id, - workspace_id, - str(user.id), - indexing_from, - indexing_to, - ) - response_message = "Web page indexing started in the background." - elif ( connector.connector_type == SearchSourceConnectorType.COMPOSIO_GOOGLE_DRIVE_CONNECTOR @@ -2478,58 +2450,6 @@ async def run_elasticsearch_indexing( ) -# Add new helper functions for crawled web page indexing -async def run_web_page_indexing_with_new_session( - connector_id: int, - workspace_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, workspace_id, user_id, start_date, end_date - ) - - -async def run_web_page_indexing( - session: AsyncSession, - connector_id: int, - workspace_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 - workspace_id: ID of the workspace - user_id: ID of the user - start_date: Start date for indexing - end_date: End date for indexing - """ - from app.tasks.connector_indexers import index_crawled_urls - - await _run_indexing_with_notifications( - session=session, - connector_id=connector_id, - workspace_id=workspace_id, - user_id=user_id, - start_date=start_date, - end_date=end_date, - indexing_function=index_crawled_urls, - update_timestamp_func=_update_connector_timestamp_by_id, - supports_heartbeat_callback=True, - ) - - # Add new helper functions for BookStack indexing async def run_bookstack_indexing_with_new_session( connector_id: int, diff --git a/surfsense_backend/app/services/connector_service.py b/surfsense_backend/app/services/connector_service.py index a13a9b3ea..ff7ded854 100644 --- a/surfsense_backend/app/services/connector_service.py +++ b/surfsense_backend/app/services/connector_service.py @@ -62,96 +62,6 @@ class ConnectorService: # Fallback to default value self.source_id_counter = 1 - async def search_crawled_urls( - self, - user_query: str, - workspace_id: int, - top_k: int = 20, - start_date: datetime | None = None, - end_date: datetime | None = None, - ) -> tuple: - """ - Search for crawled URLs and return both the source information and langchain documents. - - Uses combined chunk-level and document-level hybrid search with RRF fusion. - - Args: - user_query: The user's query - workspace_id: The workspace ID to search in - top_k: Maximum number of results to return - start_date: Optional start date for filtering documents by updated_at - end_date: Optional end date for filtering documents by updated_at - - Returns: - tuple: (sources_info, langchain_documents) - """ - crawled_urls_docs = await self._combined_rrf_search( - query_text=user_query, - workspace_id=workspace_id, - document_type="CRAWLED_URL", - top_k=top_k, - start_date=start_date, - end_date=end_date, - ) - - # Early return if no results - if not crawled_urls_docs: - return { - "id": 1, - "name": "Crawled URLs", - "type": "CRAWLED_URL", - "sources": [], - }, [] - - def _title_fn(doc_info: dict[str, Any], metadata: dict[str, Any]) -> str: - return doc_info.get("title") or metadata.get("title") or "Untitled Document" - - def _url_fn(_doc_info: dict[str, Any], metadata: dict[str, Any]) -> str: - return metadata.get("source") or metadata.get("url") or "" - - def _description_fn( - chunk: dict[str, Any], _doc_info: dict[str, Any], metadata: dict[str, Any] - ) -> str: - description = metadata.get("description") or self._chunk_preview( - chunk.get("content", "") - ) - info_parts = [] - language = metadata.get("language", "") - last_crawled_at = metadata.get("last_crawled_at", "") - if language: - info_parts.append(f"Language: {language}") - if last_crawled_at: - info_parts.append(f"Last crawled: {last_crawled_at}") - if info_parts: - description = (description + " | " + " | ".join(info_parts)).strip(" |") - return description - - def _extra_fields_fn( - _chunk: dict[str, Any], _doc_info: dict[str, Any], metadata: dict[str, Any] - ) -> dict[str, Any]: - return { - "language": metadata.get("language", ""), - "last_crawled_at": metadata.get("last_crawled_at", ""), - } - - sources_list = self._build_chunk_sources_from_documents( - crawled_urls_docs, - title_fn=_title_fn, - description_fn=_description_fn, - url_fn=_url_fn, - extra_fields_fn=_extra_fields_fn, - ) - - # Create result object - result_object = { - "id": 1, - "name": "Crawled URLs", - "type": "CRAWLED_URL", - "sources": sources_list, - } - - return result_object, crawled_urls_docs - async def search_files( self, user_query: str, diff --git a/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py b/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py index a3490e8b4..ec078e031 100644 --- a/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py +++ b/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py @@ -415,45 +415,6 @@ async def _index_elasticsearch_documents( ) -@celery_app.task(name="index_crawled_urls", bind=True) -def index_crawled_urls_task( - self, - connector_id: int, - workspace_id: int, - user_id: str, - start_date: str, - end_date: str, -): - """Celery task to index Web page Urls.""" - try: - return run_async_celery_task( - lambda: _index_crawled_urls( - connector_id, workspace_id, user_id, start_date, end_date - ) - ) - except Exception as e: - _handle_greenlet_error(e, "index_crawled_urls", connector_id) - raise - - -async def _index_crawled_urls( - connector_id: int, - workspace_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, workspace_id, user_id, start_date, end_date - ) - - @celery_app.task(name="index_bookstack_pages", bind=True) def index_bookstack_pages_task( self, 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 7243522d7..55aa04d22 100644 --- a/surfsense_backend/app/tasks/celery_tasks/schedule_checker_task.py +++ b/surfsense_backend/app/tasks/celery_tasks/schedule_checker_task.py @@ -49,7 +49,6 @@ async def _check_and_trigger_schedules(): # Teams, Gmail, Calendar, Luma) use real-time tools instead. from app.tasks.celery_tasks.connector_tasks import ( index_confluence_pages_task, - index_crawled_urls_task, index_elasticsearch_documents_task, index_github_repos_task, index_google_drive_files_task, @@ -61,7 +60,6 @@ async def _check_and_trigger_schedules(): SearchSourceConnectorType.GITHUB_CONNECTOR: index_github_repos_task, SearchSourceConnectorType.CONFLUENCE_CONNECTOR: index_confluence_pages_task, SearchSourceConnectorType.ELASTICSEARCH_CONNECTOR: index_elasticsearch_documents_task, - SearchSourceConnectorType.WEBCRAWLER_CONNECTOR: index_crawled_urls_task, SearchSourceConnectorType.GOOGLE_DRIVE_CONNECTOR: index_google_drive_files_task, SearchSourceConnectorType.COMPOSIO_GOOGLE_DRIVE_CONNECTOR: index_google_drive_files_task, } @@ -165,40 +163,6 @@ async def _check_and_trigger_schedules(): await session.commit() continue - # Special handling for Webcrawler - skip if no URLs configured - elif ( - connector.connector_type - == SearchSourceConnectorType.WEBCRAWLER_CONNECTOR - ): - from app.utils.webcrawler_utils import parse_webcrawler_urls - - connector_config = connector.config or {} - urls = parse_webcrawler_urls( - connector_config.get("INITIAL_URLS") - ) - - if urls: - task.delay( - connector.id, - connector.workspace_id, - str(connector.user_id), - None, # start_date - None, # end_date - ) - else: - # No URLs configured - skip indexing but still update next_scheduled_at - logger.info( - f"Webcrawler connector {connector.id} has no URLs configured, " - "skipping periodic indexing (will check again at next scheduled time)" - ) - from datetime import timedelta - - connector.next_scheduled_at = now + timedelta( - minutes=connector.indexing_frequency_minutes - ) - await session.commit() - continue - else: task.delay( connector.id, diff --git a/surfsense_backend/app/tasks/connector_indexers/__init__.py b/surfsense_backend/app/tasks/connector_indexers/__init__.py index 218f21066..0f867bdcf 100644 --- a/surfsense_backend/app/tasks/connector_indexers/__init__.py +++ b/surfsense_backend/app/tasks/connector_indexers/__init__.py @@ -14,12 +14,10 @@ from .google_calendar_indexer import index_google_calendar_events from .google_drive_indexer import index_google_drive_files from .google_gmail_indexer import index_google_gmail_messages from .notion_indexer import index_notion_pages -from .webcrawler_indexer import index_crawled_urls __all__ = [ "index_bookstack_pages", "index_confluence_pages", - "index_crawled_urls", "index_elasticsearch_documents", "index_github_repos", "index_google_calendar_events", diff --git a/surfsense_backend/app/tasks/connector_indexers/webcrawler_indexer.py b/surfsense_backend/app/tasks/connector_indexers/webcrawler_indexer.py deleted file mode 100644 index 8535e8d83..000000000 --- a/surfsense_backend/app/tasks/connector_indexers/webcrawler_indexer.py +++ /dev/null @@ -1,661 +0,0 @@ -""" -Webcrawler connector indexer. - -Implements 2-phase document status updates for real-time UI feedback: -- Phase 1: Create all documents with 'pending' status (visible in UI immediately) -- Phase 2: Process each document: pending → processing → ready/failed -""" - -import time -from collections.abc import Awaitable, Callable -from datetime import datetime - -from sqlalchemy.exc import SQLAlchemyError -from sqlalchemy.ext.asyncio import AsyncSession - -from app.proprietary.web_crawler import ( - CrawlOutcomeStatus, - WebCrawlerConnector, -) -from app.config import config -from app.db import Document, DocumentStatus, DocumentType, SearchSourceConnectorType -from app.services.etl_credit_service import InsufficientCreditsError -from app.services.task_logging_service import TaskLoggingService -from app.services.token_tracking_service import record_token_usage -from app.services.web_crawl_credit_service import WebCrawlCreditService -from app.utils.captcha import captcha_enabled -from app.utils.document_converters import ( - create_document_chunks, - embed_text, - generate_content_hash, - generate_unique_identifier_hash, -) -from app.utils.webcrawler_utils import parse_webcrawler_urls - -from .base import ( - check_document_by_unique_identifier, - check_duplicate_document_by_hash, - get_connector_by_id, - get_current_timestamp, - logger, - safe_set_chunks, - update_connector_last_indexed, -) - -# Type hint for heartbeat callback -HeartbeatCallbackType = Callable[[int], Awaitable[None]] - -# Heartbeat interval in seconds -HEARTBEAT_INTERVAL_SECONDS = 30 - - -async def _resolve_workspace_owner(session: AsyncSession, workspace_id: int): - """Return the ``user_id`` that owns ``workspace_id`` (the crawl payer). - - Phase 3c bills the **workspace owner**, not the triggering user (which for - periodic/scheduled runs may differ). Returns ``None`` if the workspace is - gone, letting the caller fall back to the triggering user. - """ - from sqlalchemy import select - - from app.db import Workspace - - result = await session.execute( - select(Workspace.user_id).where(Workspace.id == workspace_id) - ) - return result.scalar_one_or_none() - - -async def index_crawled_urls( - session: AsyncSession, - connector_id: int, - workspace_id: int, - user_id: str, - start_date: str | None = None, - end_date: str | None = None, - update_last_indexed: bool = True, - on_heartbeat_callback: HeartbeatCallbackType | None = None, -) -> tuple[int, str | None]: - """ - Index web page URLs with real-time document status updates. - - Implements 2-phase approach for real-time UI feedback: - - Phase 1: Create all documents with 'pending' status (visible in UI immediately) - - Phase 2: Process each document: pending → processing → ready/failed - - Args: - session: Database session - connector_id: ID of the webcrawler connector - workspace_id: ID of the workspace 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) - on_heartbeat_callback: Optional callback to update notification during long-running indexing. - - Returns: - Tuple containing (number of documents indexed, error message or None) - """ - task_logger = TaskLoggingService(session, workspace_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 URLs from connector config - raw_initial_urls = connector.config.get("INITIAL_URLS") - urls = parse_webcrawler_urls(raw_initial_urls) - - # DEBUG: Log connector config details for troubleshooting empty URL issues - logger.info( - f"Starting crawled web page indexing for connector {connector_id} with {len(urls)} URLs. " - f"Connector name: {connector.name}, " - f"INITIAL_URLS type: {type(raw_initial_urls).__name__}, " - f"INITIAL_URLS value: {repr(raw_initial_urls)[:200] if raw_initial_urls else 'None'}" - ) - - # Initialize webcrawler client - await task_logger.log_task_progress( - log_entry, - f"Initializing webcrawler client for connector {connector_id}", - { - "stage": "client_initialization", - }, - ) - - crawler = WebCrawlerConnector() - - # Validate URLs - if not urls: - # DEBUG: Log detailed connector config for troubleshooting - logger.error( - f"No URLs provided for indexing. Connector ID: {connector_id}, " - f"Connector name: {connector.name}, " - f"Config keys: {list(connector.config.keys()) if connector.config else 'None'}, " - f"INITIAL_URLS raw value: {raw_initial_urls!r}" - ) - await task_logger.log_task_failure( - log_entry, - "No URLs provided for indexing", - f"Empty URL list. INITIAL_URLS value: {repr(raw_initial_urls)[:100]}", - {"error_type": "ValidationError", "connector_name": connector.name}, - ) - return 0, "No URLs provided for indexing" - - # ======================================================================= - # Phase 3c/3d: crawl-credit pre-flight. Bill the workspace OWNER. The - # estimate is a safe UPPER BOUND so the wallet can't strand mid-batch: - # crawl = len(urls) successes (3c; actual successes are <=) - # captcha = len(urls) * MAX_ATTEMPTS (3d; per-attempt, worst case) - # No-op (and no owner lookup) when both billers are disabled. - # ======================================================================= - credit_service = WebCrawlCreditService(session) - owner_user_id = user_id - if credit_service.billing_enabled() or credit_service.captcha_billing_enabled(): - owner_user_id = ( - await _resolve_workspace_owner(session, workspace_id) or user_id - ) - required_micros = 0 - if credit_service.billing_enabled(): - required_micros += credit_service.successes_to_micros(len(urls)) - # Only reserve captcha budget when solving is actually enabled; - # with solving off, attempts can never happen, so reserving would - # wrongly block runs for captcha that will never be attempted. - if credit_service.captcha_billing_enabled() and captcha_enabled(): - worst_case_attempts = len(urls) * config.CAPTCHA_MAX_ATTEMPTS_PER_URL - required_micros += credit_service.captcha_solves_to_micros( - worst_case_attempts - ) - try: - await credit_service.check_balance(owner_user_id, required_micros) - except InsufficientCreditsError as credit_error: - await task_logger.log_task_failure( - log_entry, - f"Insufficient crawl credit for connector {connector_id}", - str(credit_error), - {"error_type": "InsufficientCreditsError"}, - ) - logger.warning( - f"Skipping crawl for connector {connector_id}: {credit_error!s}" - ) - return 0, f"Insufficient crawl credit: {credit_error!s}" - - await task_logger.log_task_progress( - log_entry, - f"Starting to process {len(urls)} URLs", - { - "stage": "processing", - "total_urls": len(urls), - }, - ) - - documents_indexed = 0 - documents_updated = 0 - documents_skipped = 0 - documents_failed = 0 - duplicate_content_count = 0 - # Explicit "crawl succeeded" count: one per URL that yielded usable - # extracted content, independent of downstream KB dedupe/unchanged - # bucketing. This is the billable unit Phase 3c meters on. - crawls_succeeded = 0 - # Phase 3d: per-attempt captcha telemetry summed across the batch. Only - # accumulated when captcha billing is on (avoids touching the outcome - # field otherwise — and keeps MagicMock outcomes in tests arithmetic-safe). - total_captcha_attempts = 0 - total_captcha_solved = 0 - - # Heartbeat tracking - update notification periodically to prevent appearing stuck - last_heartbeat_time = time.time() - - # ======================================================================= - # PHASE 1: Analyze all URLs, create pending documents for new ones - # This makes ALL new documents visible in the UI immediately with pending status - # ======================================================================= - urls_to_process = [] # List of dicts with document and URL data - new_documents_created = False - - for url in urls: - try: - # Generate unique identifier hash for this URL - unique_identifier_hash = generate_unique_identifier_hash( - DocumentType.CRAWLED_URL, url, workspace_id - ) - - # Check if document with this unique identifier already exists - existing_document = await check_document_by_unique_identifier( - session, unique_identifier_hash - ) - - if existing_document: - # Document exists - check if it's already being processed - if DocumentStatus.is_state( - existing_document.status, DocumentStatus.PENDING - ): - logger.info(f"URL {url} already pending. Skipping.") - documents_skipped += 1 - continue - if DocumentStatus.is_state( - existing_document.status, DocumentStatus.PROCESSING - ): - logger.info(f"URL {url} already processing. Skipping.") - documents_skipped += 1 - continue - - # Queue existing document for potential update check - urls_to_process.append( - { - "document": existing_document, - "is_new": False, - "url": url, - "unique_identifier_hash": unique_identifier_hash, - } - ) - continue - - # Create new document with PENDING status (visible in UI immediately) - document = Document( - workspace_id=workspace_id, - title=url[:100], # Placeholder - URL as title (truncated) - document_type=DocumentType.CRAWLED_URL, - document_metadata={ - "url": url, - "connector_id": connector_id, - }, - content="Pending crawl...", # Placeholder content - content_hash=unique_identifier_hash, # Temporary unique value - unique_identifier_hash=unique_identifier_hash, - embedding=None, - chunks=[], # Empty at creation - safe for async - status=DocumentStatus.pending(), # PENDING status - visible in UI - updated_at=get_current_timestamp(), - created_by_id=user_id, - connector_id=connector_id, - ) - session.add(document) - new_documents_created = True - - urls_to_process.append( - { - "document": document, - "is_new": True, - "url": url, - "unique_identifier_hash": unique_identifier_hash, - } - ) - - except Exception as e: - logger.error(f"Error in Phase 1 for URL {url}: {e!s}", exc_info=True) - documents_failed += 1 - continue - - # Commit all pending documents - they all appear in UI now - if new_documents_created: - logger.info( - f"Phase 1: Committing {len([u for u in urls_to_process if u['is_new']])} pending documents" - ) - await session.commit() - - # ======================================================================= - # PHASE 2: Process each URL one by one - # Each document transitions: pending → processing → ready/failed - # ======================================================================= - logger.info(f"Phase 2: Processing {len(urls_to_process)} URLs") - - for item in urls_to_process: - # Send heartbeat periodically - if on_heartbeat_callback: - current_time = time.time() - if current_time - last_heartbeat_time >= HEARTBEAT_INTERVAL_SECONDS: - await on_heartbeat_callback(documents_indexed + documents_updated) - last_heartbeat_time = current_time - - document = item["document"] - url = item["url"] - is_new = item["is_new"] - - try: - # Set to PROCESSING and commit - shows "processing" in UI for THIS document only - document.status = DocumentStatus.processing() - await session.commit() - - await task_logger.log_task_progress( - log_entry, - f"Crawling URL: {url}", - { - "stage": "crawling_url", - "url": url, - }, - ) - - # Crawl the URL - outcome = await crawler.crawl_url(url) - - # Phase 3d: count captcha attempts BEFORE the success branch — a - # failed solve still cost real solver money and must bill. - if credit_service.captcha_billing_enabled(): - total_captcha_attempts += outcome.captcha_attempts - if outcome.captcha_solved: - total_captcha_solved += 1 - - if ( - outcome.status is not CrawlOutcomeStatus.SUCCESS - or not outcome.result - ): - logger.warning(f"Failed to crawl URL {url}: {outcome.error}") - document.status = DocumentStatus.failed( - outcome.error or "Crawl failed" - ) - document.updated_at = get_current_timestamp() - await session.commit() - documents_failed += 1 - continue - - # A tier extracted usable content: count the successful crawl now, - # before the dedupe/unchanged branches (those still represent a - # successful crawl and must bill — see 03c). - crawls_succeeded += 1 - crawl_result = outcome.result - - # 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}") - document.status = DocumentStatus.failed("No content extracted") - document.updated_at = get_current_timestamp() - await session.commit() - documents_failed += 1 - continue - - # Format content as structured document for summary generation - crawler.format_to_structured_document(crawl_result) - - # Generate content hash using a version WITHOUT metadata - structured_document_for_hash = crawler.format_to_structured_document( - crawl_result, exclude_metadata=True - ) - content_hash = generate_content_hash( - structured_document_for_hash, workspace_id - ) - - # Extract useful metadata - title = metadata.get("title", url) - metadata.get("description", "") - metadata.get("language", "") - - # Update title immediately for better UX - document.title = title - await session.commit() - - # For existing documents, check if content has changed - if not is_new and document.content_hash == content_hash: - logger.info(f"Document for URL {url} unchanged. Marking as ready.") - # Ensure status is ready (might have been stuck) - document.status = DocumentStatus.ready() - await session.commit() - documents_skipped += 1 - continue - - # For new documents, check if duplicate content exists elsewhere - if is_new: - with session.no_autoflush: - duplicate_by_content = await check_duplicate_document_by_hash( - session, content_hash - ) - - if duplicate_by_content: - logger.info( - f"URL {url} already indexed by another connector " - f"(existing document ID: {duplicate_by_content.id}). " - f"Marking as failed." - ) - document.status = DocumentStatus.failed( - "Duplicate content exists" - ) - document.updated_at = get_current_timestamp() - await session.commit() - duplicate_content_count += 1 - documents_skipped += 1 - continue - - # Select deterministic document content - - summary_content = f"Crawled URL: {title}\n\nURL: {url}\n\n{content}" - summary_embedding = embed_text(summary_content) - - # Process chunks - chunks = await create_document_chunks(content) - - # Update document to READY with actual content - document.title = title - document.content = summary_content - document.content_hash = content_hash - document.embedding = summary_embedding - document.document_metadata = { - **metadata, - "crawler_type": crawler_type, - "indexed_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - "connector_id": connector_id, - } - await safe_set_chunks(session, document, chunks) - document.status = DocumentStatus.ready() # READY status - document.updated_at = get_current_timestamp() - - if is_new: - documents_indexed += 1 - else: - documents_updated += 1 - - logger.info(f"Successfully processed URL {url}") - - # Batch commit every 10 documents (for ready status updates) - 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) - # Mark document as failed with reason (visible in UI) - try: - document.status = DocumentStatus.failed(str(e)[:200]) - document.updated_at = get_current_timestamp() - await session.commit() - except Exception as status_error: - logger.error( - f"Failed to update document status to failed: {status_error}" - ) - documents_failed += 1 - continue - - total_processed = documents_indexed + documents_updated - - # CRITICAL: Always update timestamp (even if 0 documents indexed) so Zero syncs - 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" - ) - try: - await session.commit() - logger.info( - "Successfully committed all webcrawler document changes to database" - ) - except Exception as e: - # Handle any remaining integrity errors gracefully - if "duplicate key value violates unique constraint" in str(e).lower(): - logger.warning( - f"Duplicate content_hash detected during final commit. " - f"Rolling back and continuing. Error: {e!s}" - ) - await session.rollback() - else: - raise - - # ======================================================================= - # Phase 3c: charge the workspace owner for successful crawls. Audit row - # is staged first (add only); charge_credits' commit flushes both the - # ``web_crawl`` TokenUsage row and the balance debit in one transaction. - # Skipped when billing is disabled or nothing crawled successfully. - # ======================================================================= - if credit_service.billing_enabled() and crawls_succeeded > 0: - await record_token_usage( - session, - usage_type="web_crawl", - workspace_id=workspace_id, - user_id=owner_user_id, - cost_micros=credit_service.successes_to_micros(crawls_succeeded), - call_details={ - "urls": len(urls), - "successes": crawls_succeeded, - "connector_id": connector_id, - }, - message_id=None, - ) - await credit_service.charge_credits(owner_user_id, crawls_succeeded) - - # ======================================================================= - # Phase 3d: charge the workspace owner per captcha *attempt* (separate - # unit; charged even when the crawl ultimately failed). Same stage + - # owner as the crawl charge above. - # ======================================================================= - if credit_service.captcha_billing_enabled() and total_captcha_attempts > 0: - await record_token_usage( - session, - usage_type="web_crawl_captcha", - workspace_id=workspace_id, - user_id=owner_user_id, - cost_micros=credit_service.captcha_solves_to_micros( - total_captcha_attempts - ), - call_details={ - "attempts": total_captcha_attempts, - "solved": total_captcha_solved, - "connector_id": connector_id, - }, - message_id=None, - ) - await credit_service.charge_captcha(owner_user_id, total_captcha_attempts) - - # Build warning message if there were issues - warning_parts = [] - if duplicate_content_count > 0: - warning_parts.append(f"{duplicate_content_count} duplicate") - if documents_failed > 0: - warning_parts.append(f"{documents_failed} failed") - warning_message = ", ".join(warning_parts) if warning_parts else None - - await task_logger.log_task_success( - log_entry, - f"Successfully completed crawled web page indexing for connector {connector_id}", - { - "urls_processed": total_processed, - "crawls_succeeded": crawls_succeeded, - "documents_indexed": documents_indexed, - "documents_updated": documents_updated, - "documents_skipped": documents_skipped, - "documents_failed": documents_failed, - "duplicate_content_count": duplicate_content_count, - }, - ) - - logger.info( - f"Web page indexing completed: {documents_indexed} new, " - f"{documents_updated} updated, {documents_skipped} skipped, " - f"{documents_failed} failed" - ) - - if warning_message: - return total_processed, f"Completed with issues: {warning_message}" - - return total_processed, None - - 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, - workspace_id: int, - connector_id: int | None = None, -) -> list[Document]: - """ - Get all crawled URL documents for a workspace. - - Args: - session: Database session - workspace_id: ID of the workspace - connector_id: Optional connector ID to filter by - - Returns: - List of Document objects - """ - from sqlalchemy import select - - query = select(Document).filter( - Document.workspace_id == workspace_id, - Document.document_type == DocumentType.CRAWLED_URL, - ) - - if connector_id: - query = query.filter(Document.connector_id == connector_id) - - result = await session.execute(query) - documents = result.scalars().all() - return list(documents) diff --git a/surfsense_backend/app/utils/periodic_scheduler.py b/surfsense_backend/app/utils/periodic_scheduler.py index 5cce245f8..992f5038f 100644 --- a/surfsense_backend/app/utils/periodic_scheduler.py +++ b/surfsense_backend/app/utils/periodic_scheduler.py @@ -22,7 +22,6 @@ CONNECTOR_TASK_MAP = { SearchSourceConnectorType.GITHUB_CONNECTOR: "index_github_repos", SearchSourceConnectorType.CONFLUENCE_CONNECTOR: "index_confluence_pages", SearchSourceConnectorType.ELASTICSEARCH_CONNECTOR: "index_elasticsearch_documents", - SearchSourceConnectorType.WEBCRAWLER_CONNECTOR: "index_crawled_urls", SearchSourceConnectorType.BOOKSTACK_CONNECTOR: "index_bookstack_pages", } @@ -54,20 +53,6 @@ def create_periodic_schedule( True if successful, False otherwise """ try: - # Special handling for connectors that require config validation - if connector_type == SearchSourceConnectorType.WEBCRAWLER_CONNECTOR: - from app.utils.webcrawler_utils import parse_webcrawler_urls - - config = connector_config or {} - urls = parse_webcrawler_urls(config.get("INITIAL_URLS")) - - if not urls: - logger.info( - f"Webcrawler connector {connector_id} has no URLs configured, " - "skipping first indexing run (will run when URLs are added)" - ) - return True # Return success - schedule is created, just no first run - logger.info( f"Periodic indexing enabled for connector {connector_id} " f"(frequency: {frequency_minutes} minutes). Triggering first run..." @@ -76,7 +61,6 @@ def create_periodic_schedule( from app.tasks.celery_tasks.connector_tasks import ( index_bookstack_pages_task, index_confluence_pages_task, - index_crawled_urls_task, index_elasticsearch_documents_task, index_github_repos_task, index_notion_pages_task, @@ -87,7 +71,6 @@ def create_periodic_schedule( SearchSourceConnectorType.GITHUB_CONNECTOR: index_github_repos_task, SearchSourceConnectorType.CONFLUENCE_CONNECTOR: index_confluence_pages_task, SearchSourceConnectorType.ELASTICSEARCH_CONNECTOR: index_elasticsearch_documents_task, - SearchSourceConnectorType.WEBCRAWLER_CONNECTOR: index_crawled_urls_task, SearchSourceConnectorType.BOOKSTACK_CONNECTOR: index_bookstack_pages_task, } diff --git a/surfsense_backend/app/utils/validators.py b/surfsense_backend/app/utils/validators.py index 4511fb6ac..187cba21c 100644 --- a/surfsense_backend/app/utils/validators.py +++ b/surfsense_backend/app/utils/validators.py @@ -469,14 +469,6 @@ 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_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": {}}, @@ -562,13 +554,6 @@ def validate_connector_config( # "validators": {} # }, "LUMA_CONNECTOR": {"required": ["LUMA_API_KEY"], "validators": {}}, - "WEBCRAWLER_CONNECTOR": { - "required": [], # No required fields - URLs are optional - "optional": ["INITIAL_URLS"], - "validators": { - "INITIAL_URLS": lambda: validate_initial_urls(), - }, - }, } rules = connector_rules.get(connector_type_str) diff --git a/surfsense_backend/app/utils/webcrawler_utils.py b/surfsense_backend/app/utils/webcrawler_utils.py deleted file mode 100644 index 31d2ebe50..000000000 --- a/surfsense_backend/app/utils/webcrawler_utils.py +++ /dev/null @@ -1,28 +0,0 @@ -""" -Utility functions for webcrawler connector. -""" - - -def parse_webcrawler_urls(initial_urls: str | list | None) -> list[str]: - """ - Parse URLs from webcrawler INITIAL_URLS value. - - Handles both string (newline-separated) and list formats. - - Args: - initial_urls: The INITIAL_URLS value (string, list, or None) - - Returns: - List of parsed, stripped, non-empty URLs - """ - if initial_urls is None: - return [] - - if isinstance(initial_urls, str): - return [url.strip() for url in initial_urls.split("\n") if url.strip()] - elif isinstance(initial_urls, list): - return [ - url.strip() for url in initial_urls if isinstance(url, str) and url.strip() - ] - else: - return [] diff --git a/surfsense_backend/scripts/e2e_phase3_crawl_billing.py b/surfsense_backend/scripts/e2e_phase3_crawl_billing.py index 0d51b238b..31b8c832b 100644 --- a/surfsense_backend/scripts/e2e_phase3_crawl_billing.py +++ b/surfsense_backend/scripts/e2e_phase3_crawl_billing.py @@ -10,13 +10,6 @@ What it exercises (everything REAL — live network, live proxy, live DB reads): Stage 1 (3a + 3b) — direct fetch + proxy egress-IP proof + crawl_url ladder. Stage 2 (3c chat surface) — the scrape_webpage tool folds one successful crawl into the current chat turn's accumulator (billed at finalize). - Stage 3 (3c connector surface) — index_crawled_urls debits the WORKSPACE - OWNER per successful crawl and writes one `web_crawl` TokenUsage row. - -SAFETY: Stage 3 creates a scratch user/workspace/connector inside an outer -transaction that is ALWAYS rolled back (``join_transaction_mode= -"create_savepoint"``), so nothing persists to your database. The only real -side effect is a handful of HTTP requests (small proxy spend). This is NOT a pytest test (it needs a live stack + proxy creds + network). It is the manual functional counterpart to the unit suites; the undetectability / @@ -25,7 +18,6 @@ anti-bot scorecard is a separate deliverable (03f), after 03d/03e. import asyncio import sys -import uuid from pathlib import Path from urllib.parse import urlsplit @@ -39,8 +31,6 @@ for _candidate in (_BACKEND_ROOT / ".env", _BACKEND_ROOT.parent / ".env"): load_dotenv(_candidate) break -from sqlalchemy import select, text # noqa: E402 -from sqlalchemy.ext.asyncio import AsyncSession # noqa: E402 from app.config import config # noqa: E402 @@ -93,7 +83,7 @@ async def stage1_crawl_and_proxy() -> bool: try: direct = await AsyncFetcher.get(_IP_ECHO, impersonate="chrome", timeout=30) direct_ip = direct.json().get("ip") - except Exception as exc: # noqa: BLE001 + except Exception as exc: print(f" [INFO] direct IP fetch failed: {exc}") if proxy_url: try: @@ -101,7 +91,7 @@ async def stage1_crawl_and_proxy() -> bool: _IP_ECHO, impersonate="chrome", proxy=proxy_url, timeout=45 ) proxied_ip = via.json().get("ip") - except Exception as exc: # noqa: BLE001 + except Exception as exc: print(f" [INFO] proxied IP fetch failed: {exc}") print(f" egress IP (direct) : {direct_ip}") print(f" egress IP (via proxy) : {proxied_ip}") @@ -158,176 +148,16 @@ async def stage2_chat_fold() -> bool: ) -# =========================================================================== -# Stage 3 — connector indexer bills the workspace owner (3c surface 1) -# =========================================================================== -async def stage3_indexer_billing() -> bool: - _hr("STAGE 3 — index_crawled_urls bills workspace owner (3c) [rolled back]") - config.WEB_CRAWL_CREDIT_BILLING_ENABLED = True - price = config.WEB_CRAWL_MICROS_PER_SUCCESS - start_balance = 10_000_000 # $10 — plenty for the pre-flight check - - from app.db import ( - Base, - SearchSourceConnector, - SearchSourceConnectorType, - TokenUsage, - User, - Workspace, - engine, - ) - from app.tasks.connector_indexers.webcrawler_indexer import index_crawled_urls - - # Self-bootstrap: if the configured DB has no schema (e.g. an empty - # surfsense_test), create it like the integration harness. No-op against an - # already-migrated DB. DDL is outside the rolled-back txn, so tables persist - # but the scratch rows below do not. - async with engine.connect() as probe: - has_schema = ( - await probe.execute(text("select to_regclass('public.\"user\"')")) - ).scalar() is not None - if not has_schema: - print(" [INFO] empty database — creating schema (vector ext + create_all)") - async with engine.begin() as ddl: - await ddl.execute(text("CREATE EXTENSION IF NOT EXISTS vector")) - await ddl.run_sync(Base.metadata.create_all) - - async with engine.connect() as conn: - outer = await conn.begin() - try: - async with AsyncSession( - bind=conn, - expire_on_commit=False, - join_transaction_mode="create_savepoint", - ) as session: - owner = User( - id=uuid.uuid4(), - email=f"e2e-phase3+{uuid.uuid4().hex[:8]}@surfsense.test", - hashed_password="not-a-real-hash", - is_active=True, - is_superuser=False, - is_verified=True, - credit_micros_balance=start_balance, - ) - session.add(owner) - await session.flush() - - # A DISTINCT triggering user (real row — documents.created_by_id - # FKs to it) to prove the OWNER, not the trigger, gets billed. - trigger = User( - id=uuid.uuid4(), - email=f"e2e-trigger+{uuid.uuid4().hex[:8]}@surfsense.test", - hashed_password="not-a-real-hash", - is_active=True, - is_superuser=False, - is_verified=True, - credit_micros_balance=start_balance, - ) - session.add(trigger) - await session.flush() - - ws = Workspace(name="E2E Phase3 Scratch", user_id=owner.id) - session.add(ws) - await session.flush() - - connector = SearchSourceConnector( - name="E2E WebCrawler Scratch", - connector_type=SearchSourceConnectorType.WEBCRAWLER_CONNECTOR, - config={"INITIAL_URLS": _ARTICLE_URLS}, - is_indexable=True, - workspace_id=ws.id, - user_id=owner.id, - ) - session.add(connector) - await session.flush() - - # Snapshot plain values before the indexer's commits so post-run - # reads never lazy-load an expired ORM attribute. - owner_id = owner.id - trigger_id = trigger.id - ws_id = ws.id - connector_id = connector.id - print(f" owner user id : {owner_id}") - print(f" triggering user id : {trigger_id}") - print(f" urls : {len(_ARTICLE_URLS)}") - - total, warning = await index_crawled_urls( - session, connector_id, ws_id, str(trigger_id) - ) - await session.refresh(owner) - await session.refresh(trigger) - debit = start_balance - owner.credit_micros_balance - trigger_debit = start_balance - trigger.credit_micros_balance - - rows = ( - ( - await session.execute( - select(TokenUsage).where( - TokenUsage.usage_type == "web_crawl", - TokenUsage.workspace_id == ws_id, - ) - ) - ) - .scalars() - .all() - ) - - print(f" index result : processed={total} warning={warning}") - print(f" owner debit (micros) : {debit}") - print(f" trigger debit (micros): {trigger_debit}") - print(f" web_crawl usage rows : {len(rows)}") - - ok = True - if not rows: - print( - " [INFO] 0 successful crawls (site/proxy blocked) — " - "nothing billed; cannot assert debit" - ) - return False - - row = rows[0] - successes = (row.call_details or {}).get("successes") - print( - f" audit row : cost_micros={row.cost_micros} " - f"successes={successes} user_id={row.user_id}" - ) - ok &= _check("exactly one web_crawl audit row", len(rows) == 1) - ok &= _check( - "audit billed to OWNER (not trigger user)", - str(row.user_id) == str(owner_id), - ) - ok &= _check( - "triggering user NOT debited", - trigger_debit == 0, - f"trigger_debit={trigger_debit}", - ) - ok &= _check( - "cost == successes * configured price", - row.cost_micros == successes * price, - f"{row.cost_micros} == {successes} * {price}", - ) - ok &= _check( - "wallet debit matches audit cost", - debit == row.cost_micros, - f"debit={debit} cost={row.cost_micros}", - ) - return ok - finally: - await outer.rollback() - print(" [INFO] transaction rolled back — no scratch rows persisted") - - async def main() -> int: print("Phase 3 functional e2e (3a/3b/3c) — live network + proxy, DB rolled back") results: dict[str, bool] = {} for name, coro in ( ("Stage 1 crawl+proxy", stage1_crawl_and_proxy), ("Stage 2 chat fold", stage2_chat_fold), - ("Stage 3 indexer billing", stage3_indexer_billing), ): try: results[name] = await coro() - except Exception as exc: # noqa: BLE001 + except Exception as exc: import traceback traceback.print_exc() diff --git a/surfsense_backend/tests/unit/connector_indexers/test_web_crawl_billing.py b/surfsense_backend/tests/unit/connector_indexers/test_web_crawl_billing.py deleted file mode 100644 index 5f8dbdabc..000000000 --- a/surfsense_backend/tests/unit/connector_indexers/test_web_crawl_billing.py +++ /dev/null @@ -1,306 +0,0 @@ -"""Phase 3c crawl-billing wiring in the webcrawler indexer. - -System boundaries mocked: DB session, TaskLoggingService, the crawler, and the -document-conversion helpers. NOT mocked: WebCrawlCreditService (our own code) -and the indexer's billing decisions (owner resolution, pre-flight gate, -audit-then-charge, success counting). -""" - -from unittest.mock import AsyncMock, MagicMock -from uuid import UUID - -import pytest - -import app.tasks.connector_indexers.webcrawler_indexer as _mod -from app.config import config -from app.proprietary.web_crawler import CrawlOutcomeStatus - -pytestmark = pytest.mark.unit - -_CONNECTOR_ID = 7 -_WORKSPACE_ID = 1 -_TRIGGER_USER = "00000000-0000-0000-0000-0000000000aa" -_OWNER_USER = UUID("00000000-0000-0000-0000-0000000000bb") - - -class _FakeUser: - def __init__(self, balance_micros: int, reserved_micros: int = 0): - self.credit_micros_balance = balance_micros - self.credit_micros_reserved = reserved_micros - - -def _make_session(owner_id, balance_micros, reserved_micros=0): - """Mock session serving owner-resolution, get_available_micros and charge.""" - fake_user = _FakeUser(balance_micros, reserved_micros) - session = AsyncMock() - session.add = MagicMock() - session.no_autoflush = MagicMock() - - def _make_result(*_args, **_kwargs): - result = MagicMock() - # _resolve_workspace_owner → select(Workspace.user_id).scalar_one_or_none() - result.scalar_one_or_none.return_value = owner_id - # get_available_micros → .first() → (balance, reserved) - result.first.return_value = ( - fake_user.credit_micros_balance, - fake_user.credit_micros_reserved, - ) - # charge_credits → .unique().scalar_one_or_none() → User - result.unique.return_value.scalar_one_or_none.return_value = fake_user - return result - - session.execute = AsyncMock(side_effect=_make_result) - return session, fake_user - - -def _outcome( - success: bool, - content: str = "Hello content", - captcha_attempts: int = 0, - captcha_solved: bool = False, -): - o = MagicMock() - if success: - o.status = CrawlOutcomeStatus.SUCCESS - o.result = { - "content": content, - "metadata": {"title": "Title"}, - "crawler_type": "scrapling-static", - } - o.error = None - else: - o.status = CrawlOutcomeStatus.FAILED - o.result = None - o.error = "blocked" - # Real ints/bools (not MagicMock) so 03d accumulation arithmetic is safe. - o.captcha_attempts = captcha_attempts - o.captcha_solved = captcha_solved - return o - - -@pytest.fixture -def indexer_env(monkeypatch): - """Patch every system boundary the indexer touches; return the handles.""" - monkeypatch.setattr(config, "WEB_CRAWL_CREDIT_BILLING_ENABLED", True) - monkeypatch.setattr(config, "WEB_CRAWL_MICROS_PER_SUCCESS", 1000) - - # Task logger: async no-op methods, log_task_start returns a sentinel. - task_logger = MagicMock() - task_logger.log_task_start = AsyncMock(return_value=MagicMock()) - task_logger.log_task_progress = AsyncMock() - task_logger.log_task_failure = AsyncMock() - task_logger.log_task_success = AsyncMock() - monkeypatch.setattr(_mod, "TaskLoggingService", MagicMock(return_value=task_logger)) - - # Connector + URL parsing. - connector = MagicMock() - connector.name = "wc" - monkeypatch.setattr(_mod, "get_connector_by_id", AsyncMock(return_value=connector)) - monkeypatch.setattr(_mod, "parse_webcrawler_urls", lambda raw: list(raw)) - - # Crawler. - crawler = MagicMock() - crawler.crawl_url = AsyncMock() - crawler.format_to_structured_document = MagicMock(return_value="doc") - monkeypatch.setattr(_mod, "WebCrawlerConnector", MagicMock(return_value=crawler)) - - # Document-conversion + persistence helpers (system boundary). - monkeypatch.setattr( - _mod, "check_document_by_unique_identifier", AsyncMock(return_value=None) - ) - monkeypatch.setattr( - _mod, "check_duplicate_document_by_hash", AsyncMock(return_value=None) - ) - monkeypatch.setattr(_mod, "embed_text", lambda *_a, **_k: [0.1, 0.2]) - monkeypatch.setattr(_mod, "create_document_chunks", AsyncMock(return_value=[])) - monkeypatch.setattr(_mod, "safe_set_chunks", AsyncMock()) - monkeypatch.setattr(_mod, "update_connector_last_indexed", AsyncMock()) - - # Audit helper — assert it's the owner who is billed. - record_usage = AsyncMock(return_value=MagicMock()) - monkeypatch.setattr(_mod, "record_token_usage", record_usage) - - return { - "connector": connector, - "crawler": crawler, - "record_usage": record_usage, - "task_logger": task_logger, - } - - -async def _run(env, session, urls): - env["connector"].config = {"INITIAL_URLS": urls} - return await _mod.index_crawled_urls( - session, - _CONNECTOR_ID, - _WORKSPACE_ID, - _TRIGGER_USER, - ) - - -async def test_charges_owner_for_successful_crawls(indexer_env): - session, user = _make_session(_OWNER_USER, balance_micros=100_000) - indexer_env["crawler"].crawl_url.side_effect = [_outcome(True), _outcome(True)] - - total, warning = await _run(indexer_env, session, ["https://a.com", "https://b.com"]) - - assert total == 2 - assert warning is None - # Owner debited 2 * 1000. - assert user.credit_micros_balance == 100_000 - 2000 - # One audit row, billed to the OWNER (not the triggering user), correct cost. - indexer_env["record_usage"].assert_awaited_once() - kwargs = indexer_env["record_usage"].await_args.kwargs - assert kwargs["usage_type"] == "web_crawl" - assert kwargs["user_id"] == _OWNER_USER - assert kwargs["workspace_id"] == _WORKSPACE_ID - assert kwargs["cost_micros"] == 2000 - - -async def test_preflight_blocks_run_when_insufficient(indexer_env): - # balance 1000 < 3 URLs * 1000 required. - session, user = _make_session(_OWNER_USER, balance_micros=1000) - - total, warning = await _run( - indexer_env, session, ["https://a.com", "https://b.com", "https://c.com"] - ) - - assert total == 0 - assert "insufficient crawl credit" in warning.lower() - # Never crawled, never billed, balance untouched. - indexer_env["crawler"].crawl_url.assert_not_awaited() - indexer_env["record_usage"].assert_not_awaited() - assert user.credit_micros_balance == 1000 - - -async def test_failed_crawls_are_free(indexer_env): - session, user = _make_session(_OWNER_USER, balance_micros=100_000) - indexer_env["crawler"].crawl_url.side_effect = [_outcome(False)] - - total, _warning = await _run(indexer_env, session, ["https://a.com"]) - - assert total == 0 - # crawls_succeeded == 0 → no audit, no charge. - indexer_env["record_usage"].assert_not_awaited() - assert user.credit_micros_balance == 100_000 - - -async def test_disabled_skips_billing_but_still_crawls(indexer_env, monkeypatch): - monkeypatch.setattr(config, "WEB_CRAWL_CREDIT_BILLING_ENABLED", False) - session, user = _make_session(_OWNER_USER, balance_micros=100_000) - indexer_env["crawler"].crawl_url.side_effect = [_outcome(True)] - - total, _warning = await _run(indexer_env, session, ["https://a.com"]) - - assert total == 1 - indexer_env["crawler"].crawl_url.assert_awaited_once() - indexer_env["record_usage"].assert_not_awaited() - assert user.credit_micros_balance == 100_000 - - -# =================================================================== -# Phase 3d — captcha per-attempt billing in the indexer -# =================================================================== - - -async def test_captcha_attempts_charged_to_owner(indexer_env, monkeypatch): - """Captcha billed per attempt to the OWNER as a separate web_crawl_captcha - unit. Crawl billing OFF here to isolate the captcha charge.""" - monkeypatch.setattr(config, "WEB_CRAWL_CREDIT_BILLING_ENABLED", False) - monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", True) - monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE", 3000) - session, user = _make_session(_OWNER_USER, balance_micros=100_000) - indexer_env["crawler"].crawl_url.side_effect = [ - _outcome(True, captcha_attempts=1, captcha_solved=True), - _outcome(True, captcha_attempts=1, captcha_solved=True), - ] - - total, _warning = await _run( - indexer_env, session, ["https://a.com", "https://b.com"] - ) - - assert total == 2 - # Owner debited 2 attempts * 3000 (no crawl charge — crawl billing off). - assert user.credit_micros_balance == 100_000 - 6000 - indexer_env["record_usage"].assert_awaited_once() - kwargs = indexer_env["record_usage"].await_args.kwargs - assert kwargs["usage_type"] == "web_crawl_captcha" - assert kwargs["user_id"] == _OWNER_USER - assert kwargs["cost_micros"] == 6000 - assert kwargs["call_details"]["attempts"] == 2 - assert kwargs["call_details"]["solved"] == 2 - - -async def test_captcha_attempt_billed_even_when_crawl_fails(indexer_env, monkeypatch): - """A failed solve still cost solver money → bill the attempt anyway.""" - monkeypatch.setattr(config, "WEB_CRAWL_CREDIT_BILLING_ENABLED", False) - monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", True) - monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE", 3000) - session, user = _make_session(_OWNER_USER, balance_micros=100_000) - indexer_env["crawler"].crawl_url.side_effect = [ - _outcome(False, captcha_attempts=1, captcha_solved=False) - ] - - total, _warning = await _run(indexer_env, session, ["https://a.com"]) - - assert total == 0 # crawl failed → not indexed - assert user.credit_micros_balance == 100_000 - 3000 # but solve attempt billed - kwargs = indexer_env["record_usage"].await_args.kwargs - assert kwargs["usage_type"] == "web_crawl_captcha" - assert kwargs["call_details"]["attempts"] == 1 - assert kwargs["call_details"]["solved"] == 0 - - -async def test_captcha_billing_on_but_solving_off_does_not_overreserve( - indexer_env, monkeypatch -): - """Regression: captcha billing ON but solving OFF must NOT reserve the - worst-case captcha budget in pre-flight. - - With solving off, attempts can never happen, so a balance below the - (would-be) worst-case reservation must still let the run proceed rather than - being wrongly blocked for captcha that never runs. - """ - monkeypatch.setattr(config, "WEB_CRAWL_CREDIT_BILLING_ENABLED", False) - monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", True) - monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_MICROS_PER_SOLVE", 3000) - monkeypatch.setattr(config, "CAPTCHA_MAX_ATTEMPTS_PER_URL", 3) - # Solving OFF → captcha_enabled() is False (no flag + no key). - monkeypatch.setattr(config, "CAPTCHA_SOLVING_ENABLED", False) - monkeypatch.setattr(config, "CAPTCHA_SOLVER_API_KEY", "") - - # Balance (5000) is below the buggy worst-case reservation - # (2 URLs * 3 attempts * 3000 = 18000) but the run must still proceed. - session, user = _make_session(_OWNER_USER, balance_micros=5000) - indexer_env["crawler"].crawl_url.side_effect = [ - _outcome(True, captcha_attempts=0), - _outcome(True, captcha_attempts=0), - ] - - total, warning = await _run( - indexer_env, session, ["https://a.com", "https://b.com"] - ) - - assert total == 2 - assert warning is None - assert indexer_env["crawler"].crawl_url.await_count == 2 - # Nothing billed: crawl billing off, no captcha attempts. - indexer_env["record_usage"].assert_not_awaited() - assert user.credit_micros_balance == 5000 - - -async def test_captcha_disabled_skips_captcha_billing(indexer_env, monkeypatch): - """Captcha billing OFF (default) → attempts on the outcome are ignored.""" - monkeypatch.setattr(config, "WEB_CRAWL_CAPTCHA_BILLING_ENABLED", False) - session, user = _make_session(_OWNER_USER, balance_micros=100_000) - indexer_env["crawler"].crawl_url.side_effect = [ - _outcome(True, captcha_attempts=5, captcha_solved=True) - ] - - total, _warning = await _run(indexer_env, session, ["https://a.com"]) - - assert total == 1 - # Only the crawl unit billed (1 * 1000); captcha ignored entirely. - assert user.credit_micros_balance == 100_000 - 1000 - kwargs = indexer_env["record_usage"].await_args.kwargs - assert kwargs["usage_type"] == "web_crawl" diff --git a/surfsense_backend/tests/unit/utils/test_validators.py b/surfsense_backend/tests/unit/utils/test_validators.py index 82846bdb9..a359ee550 100644 --- a/surfsense_backend/tests/unit/utils/test_validators.py +++ b/surfsense_backend/tests/unit/utils/test_validators.py @@ -11,10 +11,10 @@ from app.utils.validators import ( validate_messages, validate_research_mode, validate_search_mode, - validate_workspace_id, validate_top_k, validate_url, validate_uuid, + validate_workspace_id, ) pytestmark = pytest.mark.unit @@ -331,9 +331,3 @@ def test_validate_connector_config_invalid(): # Invalid URL format in SEARXNG_API with pytest.raises(ValueError): validate_connector_config("SEARXNG_API", {"SEARXNG_HOST": "not-a-url"}) - - # WEBCRAWLER_CONNECTOR custom validation: malformed INITIAL_URLS rejected. - with pytest.raises(ValueError): - validate_connector_config( - "WEBCRAWLER_CONNECTOR", {"INITIAL_URLS": "not-a-url"} - )