mirror of
https://github.com/MODSetter/SurfSense.git
synced 2026-07-08 22:22:17 +02:00
refactor(connectors): remove legacy web-crawler KB indexer paths (keep enums)
This commit is contained in:
CREDO23
parent
e3ed3b2be3
commit
ab6be6cbda
14 changed files with 4 additions and 1456 deletions
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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},
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
@ -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,
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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 []
|
||||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
@ -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"}
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue