diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/scrape_webpage.py b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/scrape_webpage.py index 82dcaa9c1..2c1173bfc 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/scrape_webpage.py +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/scrape_webpage.py @@ -18,11 +18,11 @@ from requests import Session from scrapling.fetchers import AsyncFetcher from youtube_transcript_api import YouTubeTranscriptApi +from app.proprietary.platforms.youtube.url_resolver import get_youtube_video_id from app.proprietary.web_crawler import ( CrawlOutcomeStatus, WebCrawlerConnector, ) -from app.tasks.document_processors.youtube_processor import get_youtube_video_id from app.utils.proxy import get_proxy_url, get_requests_proxies logger = logging.getLogger(__name__) diff --git a/surfsense_backend/app/proprietary/platforms/youtube/subtitles.py b/surfsense_backend/app/proprietary/platforms/youtube/subtitles.py index cbec49ec8..dc5647b30 100644 --- a/surfsense_backend/app/proprietary/platforms/youtube/subtitles.py +++ b/surfsense_backend/app/proprietary/platforms/youtube/subtitles.py @@ -2,7 +2,7 @@ Uses the library's built-in formatters (no hand-rolled srt/vtt conversion). The transcript API has no XML formatter, so ``xml`` falls back to the raw snippet -data. Fetch/proxy wiring mirrors ``youtube_processor.py``. +data. """ from __future__ import annotations diff --git a/surfsense_backend/app/proprietary/platforms/youtube/url_resolver.py b/surfsense_backend/app/proprietary/platforms/youtube/url_resolver.py index 43bad2e95..0e68425a9 100644 --- a/surfsense_backend/app/proprietary/platforms/youtube/url_resolver.py +++ b/surfsense_backend/app/proprietary/platforms/youtube/url_resolver.py @@ -1,8 +1,7 @@ """Classify a YouTube URL and extract its identifier. Covers the ``startUrls`` shapes the Apify spec accepts: video, channel, -playlist, hashtag, and search-results pages. Video-ID extraction reuses the -logic already in ``app/tasks/document_processors/youtube_processor.py``. +playlist, hashtag, and search-results pages. """ from __future__ import annotations diff --git a/surfsense_backend/app/routes/documents_routes.py b/surfsense_backend/app/routes/documents_routes.py index ade17a59f..6dbf65b20 100644 --- a/surfsense_backend/app/routes/documents_routes.py +++ b/surfsense_backend/app/routes/documents_routes.py @@ -98,13 +98,6 @@ async def create_documents( process_extension_document_task.delay( document_dict, request.workspace_id, str(user.id) ) - elif request.document_type == DocumentType.YOUTUBE_VIDEO: - from app.tasks.celery_tasks.document_tasks import process_youtube_video_task - - for url in request.content: - process_youtube_video_task.delay( - url, request.workspace_id, str(user.id) - ) else: raise HTTPException(status_code=400, detail="Invalid document type") diff --git a/surfsense_backend/app/tasks/celery_tasks/document_tasks.py b/surfsense_backend/app/tasks/celery_tasks/document_tasks.py index ba8a49f67..694bf6409 100644 --- a/surfsense_backend/app/tasks/celery_tasks/document_tasks.py +++ b/surfsense_backend/app/tasks/celery_tasks/document_tasks.py @@ -19,7 +19,6 @@ from app.tasks.connector_indexers.local_folder_indexer import ( ) from app.tasks.document_processors import ( add_extension_received_document, - add_youtube_video_document, ) logger = logging.getLogger(__name__) @@ -405,131 +404,6 @@ async def _process_extension_document( raise -@celery_app.task(name="process_youtube_video", bind=True) -def process_youtube_video_task(self, url: str, workspace_id: int, user_id: str): - """ - Celery task to process YouTube video. - - Args: - url: YouTube video URL - workspace_id: ID of the workspace - user_id: ID of the user - """ - return run_async_celery_task( - lambda: _process_youtube_video(url, workspace_id, user_id) - ) - - -async def _process_youtube_video(url: str, workspace_id: int, user_id: str): - """Process YouTube video with new session.""" - async with get_celery_session_maker()() as session: - task_logger = TaskLoggingService(session, workspace_id) - - # Extract video title from URL for notification (will be updated later) - video_name = url.split("v=")[-1][:11] if "v=" in url else url - - # Create notification for document processing - notification = ( - await NotificationService.document_processing.notify_processing_started( - session=session, - user_id=UUID(user_id), - document_type="YOUTUBE_VIDEO", - document_name=f"YouTube: {video_name}", - workspace_id=workspace_id, - ) - ) - - # Start Redis heartbeat for stale task detection - _start_heartbeat(notification.id) - heartbeat_task = asyncio.create_task(_run_heartbeat_loop(notification.id)) - - log_entry = await task_logger.log_task_start( - task_name="process_youtube_video", - source="document_processor", - message=f"Starting YouTube video processing for: {url}", - metadata={"document_type": "YOUTUBE_VIDEO", "url": url, "user_id": user_id}, - ) - - try: - # Update notification: parsing (fetching transcript) - await NotificationService.document_processing.notify_processing_progress( - session, - notification, - stage="parsing", - stage_message="Fetching video transcript", - ) - - result = await add_youtube_video_document( - session, url, workspace_id, user_id, notification=notification - ) - - if result: - await task_logger.log_task_success( - log_entry, - f"Successfully processed YouTube video: {result.title}", - { - "document_id": result.id, - "video_id": result.document_metadata.get("video_id"), - "content_hash": result.content_hash, - }, - ) - - # Update notification on success - await ( - NotificationService.document_processing.notify_processing_completed( - session=session, - notification=notification, - document_id=result.id, - chunks_count=None, - ) - ) - else: - await task_logger.log_task_success( - log_entry, - f"YouTube video document already exists (duplicate): {url}", - {"duplicate_detected": True}, - ) - - # Update notification for duplicate - await ( - NotificationService.document_processing.notify_processing_completed( - session=session, - notification=notification, - error_message="Video already exists (duplicate)", - ) - ) - except Exception as e: - await task_logger.log_task_failure( - log_entry, - f"Failed to process YouTube video: {url}", - str(e), - {"error_type": type(e).__name__}, - ) - - # Update notification on failure - wrapped in try-except to ensure it doesn't fail silently - try: - # Refresh notification to ensure it's not stale after any rollback - await session.refresh(notification) - await ( - NotificationService.document_processing.notify_processing_completed( - session=session, - notification=notification, - error_message=str(e)[:100], - ) - ) - except Exception as notif_error: - logger.error( - f"Failed to update notification on failure: {notif_error!s}" - ) - - logger.error(f"Error processing YouTube video: {e!s}") - raise - finally: - # Stop heartbeat — key deleted on success, expires on crash - heartbeat_task.cancel() - _stop_heartbeat(notification.id) - - @celery_app.task(name="process_file_upload", bind=True) def process_file_upload_task( self, file_path: str, filename: str, workspace_id: int, user_id: str diff --git a/surfsense_backend/app/tasks/document_processors/__init__.py b/surfsense_backend/app/tasks/document_processors/__init__.py index f82c10883..90fd69269 100644 --- a/surfsense_backend/app/tasks/document_processors/__init__.py +++ b/surfsense_backend/app/tasks/document_processors/__init__.py @@ -3,15 +3,13 @@ Document processors module for background tasks. Content extraction is handled by ``app.etl_pipeline.EtlPipelineService``. This package keeps orchestration (save, notify, page-limit) and -non-ETL processors (extension, markdown, youtube). +non-ETL processors (extension, markdown). """ from .extension_processor import add_extension_received_document from .markdown_processor import add_received_markdown_file_document -from .youtube_processor import add_youtube_video_document __all__ = [ "add_extension_received_document", "add_received_markdown_file_document", - "add_youtube_video_document", ] diff --git a/surfsense_backend/app/tasks/document_processors/youtube_processor.py b/surfsense_backend/app/tasks/document_processors/youtube_processor.py deleted file mode 100644 index 255ba4f7e..000000000 --- a/surfsense_backend/app/tasks/document_processors/youtube_processor.py +++ /dev/null @@ -1,473 +0,0 @@ -""" -YouTube video document processor. - -Implements 2-phase document status updates for real-time UI feedback: -- Phase 1: Create document with 'pending' status (visible in UI immediately) -- Phase 2: Process document: pending → processing → ready/failed -""" - -import logging -import time -from urllib.parse import parse_qs, urlparse - -from fake_useragent import UserAgent -from requests import Session -from scrapling.fetchers import AsyncFetcher -from sqlalchemy.exc import SQLAlchemyError -from sqlalchemy.ext.asyncio import AsyncSession -from youtube_transcript_api import YouTubeTranscriptApi - -from app.db import Document, DocumentStatus, DocumentType -from app.services.task_logging_service import TaskLoggingService -from app.utils.document_converters import ( - create_document_chunks, - embed_text, - generate_content_hash, - generate_unique_identifier_hash, -) -from app.utils.proxy import get_proxy_url, get_requests_proxies - -from .base import ( - check_document_by_unique_identifier, - get_current_timestamp, - safe_set_chunks, -) - - -def get_youtube_video_id(url: str) -> str | None: - """ - Extract video ID from various YouTube URL formats. - - Args: - url: YouTube URL - - Returns: - Video ID if found, None otherwise - """ - parsed_url = urlparse(url) - hostname = parsed_url.hostname - - if hostname == "youtu.be": - return parsed_url.path[1:] - if hostname in ("www.youtube.com", "youtube.com"): - if parsed_url.path == "/watch": - query_params = parse_qs(parsed_url.query) - return query_params.get("v", [None])[0] - if parsed_url.path.startswith("/embed/"): - return parsed_url.path.split("/")[2] - if parsed_url.path.startswith("/v/"): - return parsed_url.path.split("/")[2] - return None - - -async def add_youtube_video_document( - session: AsyncSession, - url: str, - workspace_id: int, - user_id: str, - notification=None, -) -> Document: - """ - Process a YouTube video URL, extract transcripts, and store as a document. - - Implements 2-phase document status updates for real-time UI feedback: - - Phase 1: Create document with 'pending' status (visible in UI immediately) - - Phase 2: Process document: pending → processing → ready/failed - - Args: - session: Database session for storing the document - url: YouTube video URL (supports standard, shortened, and embed formats) - workspace_id: ID of the workspace to add the document to - user_id: ID of the user - notification: Optional notification object — if provided, the document_id - is stored in its metadata right after document creation so the stale - cleanup task can identify stuck documents. - - Returns: - Document: The created document object - - Raises: - ValueError: If the YouTube video ID cannot be extracted from the URL - SQLAlchemyError: If there's a database error - RuntimeError: If the video processing fails - """ - task_logger = TaskLoggingService(session, workspace_id) - - # Log task start - log_entry = await task_logger.log_task_start( - task_name="youtube_video_document", - source="background_task", - message=f"Starting YouTube video processing for: {url}", - metadata={"url": url, "user_id": str(user_id)}, - ) - - document = None - video_id = None - is_new_document = False - - try: - # Extract video ID from URL (lightweight operation) - await task_logger.log_task_progress( - log_entry, - f"Extracting video ID from URL: {url}", - {"stage": "video_id_extraction"}, - ) - - video_id = get_youtube_video_id(url) - if not video_id: - raise ValueError(f"Could not extract video ID from URL: {url}") - - await task_logger.log_task_progress( - log_entry, - f"Video ID extracted: {video_id}", - {"stage": "video_id_extracted", "video_id": video_id}, - ) - - # Generate unique identifier hash for this YouTube video - unique_identifier_hash = generate_unique_identifier_hash( - DocumentType.YOUTUBE_VIDEO, video_id, workspace_id - ) - - # Check if document with this unique identifier already exists - await task_logger.log_task_progress( - log_entry, - f"Checking for existing video: {video_id}", - {"stage": "duplicate_check", "video_id": video_id}, - ) - - existing_document = await check_document_by_unique_identifier( - session, unique_identifier_hash - ) - - # ======================================================================= - # PHASE 1: Create pending document or prepare existing for update - # ======================================================================= - if existing_document: - document = existing_document - is_new_document = False - # Check if already being processed - if DocumentStatus.is_state( - existing_document.status, DocumentStatus.PENDING - ): - logging.info( - f"YouTube video {video_id} already pending. Returning existing." - ) - return existing_document - if DocumentStatus.is_state( - existing_document.status, DocumentStatus.PROCESSING - ): - logging.info( - f"YouTube video {video_id} already processing. Returning existing." - ) - return existing_document - else: - # Create new document with PENDING status (visible in UI immediately) - await task_logger.log_task_progress( - log_entry, - f"Creating pending document for video: {video_id}", - {"stage": "pending_document_creation"}, - ) - - document = Document( - title=f"YouTube Video: {video_id}", # Placeholder title - document_type=DocumentType.YOUTUBE_VIDEO, - document_metadata={ - "url": url, - "video_id": video_id, - }, - content="Processing video...", # Placeholder content - content_hash=unique_identifier_hash, # Temporary unique value - unique_identifier_hash=unique_identifier_hash, - embedding=None, - chunks=[], # Empty at creation - status=DocumentStatus.pending(), # PENDING status - visible in UI - workspace_id=workspace_id, - updated_at=get_current_timestamp(), - created_by_id=user_id, - ) - session.add(document) - await session.commit() # Document visible in UI now with pending status! - is_new_document = True - - # Store document_id in notification metadata so stale cleanup task - # can identify this document if the worker crashes. - if notification and notification.notification_metadata is not None: - from sqlalchemy.orm.attributes import flag_modified - - notification.notification_metadata["document_id"] = document.id - flag_modified(notification, "notification_metadata") - await session.commit() - - logging.info(f"Created pending document for YouTube video {video_id}") - - # ======================================================================= - # PHASE 2: Set to PROCESSING and do heavy work - # ======================================================================= - document.status = DocumentStatus.processing() - await session.commit() # UI shows "processing" status - - await task_logger.log_task_progress( - log_entry, - f"Fetching video metadata for: {video_id}", - {"stage": "metadata_fetch"}, - ) - - # Fetch video metadata - params = { - "format": "json", - "url": f"https://www.youtube.com/watch?v={video_id}", - } - oembed_url = "https://www.youtube.com/oembed" - - # Build residential proxy settings (if configured) - residential_proxies = get_requests_proxies() - - oembed_fetch_start = time.perf_counter() - oembed_page = await AsyncFetcher.get( - oembed_url, - params=params, - proxy=get_proxy_url(), - stealthy_headers=True, - ) - logging.info( - "[youtube][perf] source=oembed video=%s status=%s fetch_ms=%.1f", - video_id, - getattr(oembed_page, "status", None), - (time.perf_counter() - oembed_fetch_start) * 1000, - ) - video_data = oembed_page.json() - - # Update title immediately for better UX (user sees actual title sooner) - document.title = video_data.get("title", f"YouTube Video: {video_id}") - await session.commit() - - await task_logger.log_task_progress( - log_entry, - f"Video metadata fetched: {video_data.get('title', 'Unknown')}", - { - "stage": "metadata_fetched", - "title": video_data.get("title"), - "author": video_data.get("author_name"), - }, - ) - - # Get video transcript - await task_logger.log_task_progress( - log_entry, - f"Fetching transcript for video: {video_id}", - {"stage": "transcript_fetch"}, - ) - - try: - transcript_fetch_start = time.perf_counter() - ua = UserAgent() - http_client = Session() - http_client.headers.update({"User-Agent": ua.random}) - if residential_proxies: - http_client.proxies.update(residential_proxies) - ytt_api = YouTubeTranscriptApi(http_client=http_client) - - # List all available transcripts and pick the first one - # (the video's primary language) instead of defaulting to English - transcript_list = ytt_api.list(video_id) - transcript = next(iter(transcript_list)) - captions = transcript.fetch() - logging.info( - "[youtube][perf] source=transcript video=%s fetch_ms=%.1f", - video_id, - (time.perf_counter() - transcript_fetch_start) * 1000, - ) - - # Include complete caption information with timestamps - transcript_segments = [] - for line in captions: - start_time = line.start - duration = line.duration - text = line.text - timestamp = f"[{start_time:.2f}s-{start_time + duration:.2f}s]" - transcript_segments.append(f"{timestamp} {text}") - transcript_text = "\n".join(transcript_segments) - - await task_logger.log_task_progress( - log_entry, - f"Transcript fetched successfully: {len(captions)} segments ({transcript.language})", - { - "stage": "transcript_fetched", - "segments_count": len(captions), - "transcript_length": len(transcript_text), - "language": transcript.language, - "language_code": transcript.language_code, - "is_generated": transcript.is_generated, - }, - ) - except Exception as e: - transcript_text = f"No captions available for this video. Error: {e!s}" - await task_logger.log_task_progress( - log_entry, - f"No transcript available for video: {video_id}", - {"stage": "transcript_unavailable", "error": str(e)}, - ) - - # Format document - await task_logger.log_task_progress( - log_entry, - f"Processing video content: {video_data.get('title', 'YouTube Video')}", - {"stage": "content_processing"}, - ) - - # Format document metadata in a more maintainable way - metadata_sections = [ - ( - "METADATA", - [ - f"TITLE: {video_data.get('title', 'YouTube Video')}", - f"URL: {url}", - f"VIDEO_ID: {video_id}", - f"AUTHOR: {video_data.get('author_name', 'Unknown')}", - f"THUMBNAIL: {video_data.get('thumbnail_url', '')}", - ], - ), - ( - "CONTENT", - ["FORMAT: transcript", "TEXT_START", transcript_text, "TEXT_END"], - ), - ] - - # Build the document string more efficiently - document_parts = [] - document_parts.append("") - - for section_title, section_content in metadata_sections: - document_parts.append(f"<{section_title}>") - document_parts.extend(section_content) - document_parts.append(f"") - - document_parts.append("") - combined_document_string = "\n".join(document_parts) - - # Generate content hash - content_hash = generate_content_hash(combined_document_string, workspace_id) - - # For existing documents, check if content has changed - if not is_new_document and existing_document.content_hash == content_hash: - await task_logger.log_task_success( - log_entry, - f"YouTube video document unchanged: {video_data.get('title', 'YouTube Video')}", - { - "duplicate_detected": True, - "existing_document_id": existing_document.id, - "video_id": video_id, - }, - ) - logging.info( - f"Document for YouTube video {video_id} unchanged. Marking as ready." - ) - document.status = DocumentStatus.ready() - await session.commit() - return document - - summary_content = combined_document_string - summary_embedding = embed_text(summary_content) - - # Process chunks - await task_logger.log_task_progress( - log_entry, - f"Processing content chunks for video: {video_data.get('title', 'YouTube Video')}", - {"stage": "chunk_processing"}, - ) - - chunks = await create_document_chunks(combined_document_string) - - # ======================================================================= - # PHASE 3: Update document to READY with all content - # ======================================================================= - await task_logger.log_task_progress( - log_entry, - f"Finalizing document: {video_data.get('title', 'YouTube Video')}", - {"stage": "document_finalization", "chunks_count": len(chunks)}, - ) - - document.title = video_data.get("title", "YouTube Video") - document.content = summary_content - document.content_hash = content_hash - document.embedding = summary_embedding - document.document_metadata = { - "url": url, - "video_id": video_id, - "video_title": video_data.get("title", "YouTube Video"), - "author": video_data.get("author_name", "Unknown"), - "thumbnail": video_data.get("thumbnail_url", ""), - } - await safe_set_chunks(session, document, chunks) - document.source_markdown = combined_document_string - document.status = DocumentStatus.ready() # READY status - fully processed - document.updated_at = get_current_timestamp() - - await session.commit() - await session.refresh(document) - - # Log success - await task_logger.log_task_success( - log_entry, - f"Successfully processed YouTube video: {video_data.get('title', 'YouTube Video')}", - { - "document_id": document.id, - "video_id": video_id, - "title": document.title, - "content_hash": content_hash, - "chunks_count": len(chunks), - "summary_length": len(summary_content), - "has_transcript": "No captions available" not in transcript_text, - }, - ) - - return document - - except SQLAlchemyError as db_error: - # Mark document as failed if it exists - if document: - try: - document.status = DocumentStatus.failed( - f"Database error: {str(db_error)[:150]}" - ) - document.updated_at = get_current_timestamp() - await session.commit() - except Exception: - await session.rollback() - else: - await session.rollback() - - await task_logger.log_task_failure( - log_entry, - f"Database error while processing YouTube video: {url}", - str(db_error), - { - "error_type": "SQLAlchemyError", - "video_id": video_id, - }, - ) - raise db_error - - except Exception as e: - # Mark document as failed if it exists - if document: - try: - document.status = DocumentStatus.failed(str(e)[:200]) - document.updated_at = get_current_timestamp() - await session.commit() - except Exception: - await session.rollback() - else: - await session.rollback() - - await task_logger.log_task_failure( - log_entry, - f"Failed to process YouTube video: {url}", - str(e), - { - "error_type": type(e).__name__, - "video_id": video_id, - }, - ) - logging.error(f"Failed to process YouTube video: {e!s}") - raise diff --git a/surfsense_backend/tests/unit/observability/test_helpers.py b/surfsense_backend/tests/unit/observability/test_helpers.py index eafb8b626..d81a7ed50 100644 --- a/surfsense_backend/tests/unit/observability/test_helpers.py +++ b/surfsense_backend/tests/unit/observability/test_helpers.py @@ -26,7 +26,6 @@ def _disable_otel(monkeypatch: pytest.MonkeyPatch): ("delete_folder_documents_background", "delete"), ("delete_search_space_background", "delete"), ("process_extension_document", "process"), - ("process_youtube_video", "process"), ("process_file_upload", "process"), ("process_file_upload_with_document", "process"), ("process_circleback_meeting", "process"),