apunkt/SurfSense
1
0
Fork 0
mirror of https://github.com/MODSetter/SurfSense.git synced 2026-07-08 22:22:17 +02:00
Code Issues Projects Releases Packages Wiki Activity

refactor(youtube): remove legacy KB ingestion; use actor get_youtube_video_id

Browse source
This commit is contained in:
CREDO23 2026-07-03 12:00:22 +02:00
parent ab6be6cbda
commit 817ec0e9f4
8 changed files with 4 additions and 614 deletions
Download patch file Download diff file Expand all files Collapse all files

2
surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/tools/scrape_webpage.py
View file

@ -18,11 +18,11 @@ from requests import Session
from scrapling.fetchers import AsyncFetcher from scrapling.fetchers import AsyncFetcher
from youtube_transcript_api import YouTubeTranscriptApi from youtube_transcript_api import YouTubeTranscriptApi
from app.proprietary.platforms.youtube.url_resolver import get_youtube_video_id
from app.proprietary.web_crawler import ( from app.proprietary.web_crawler import (
CrawlOutcomeStatus, CrawlOutcomeStatus,
WebCrawlerConnector, WebCrawlerConnector,
) )
from app.tasks.document_processors.youtube_processor import get_youtube_video_id
from app.utils.proxy import get_proxy_url, get_requests_proxies from app.utils.proxy import get_proxy_url, get_requests_proxies
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)

2
surfsense_backend/app/proprietary/platforms/youtube/subtitles.py
View file

@ -2,7 +2,7 @@
Uses the library's built-in formatters (no hand-rolled srt/vtt conversion). The 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 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 from __future__ import annotations

3
surfsense_backend/app/proprietary/platforms/youtube/url_resolver.py
View file

@ -1,8 +1,7 @@
"""Classify a YouTube URL and extract its identifier. """Classify a YouTube URL and extract its identifier.
Covers the ``startUrls`` shapes the Apify spec accepts: video, channel, Covers the ``startUrls`` shapes the Apify spec accepts: video, channel,
playlist, hashtag, and search-results pages. Video-ID extraction reuses the playlist, hashtag, and search-results pages.
logic already in ``app/tasks/document_processors/youtube_processor.py``.
""" """
from __future__ import annotations from __future__ import annotations

7
surfsense_backend/app/routes/documents_routes.py
View file

@ -98,13 +98,6 @@ async def create_documents(
process_extension_document_task.delay( process_extension_document_task.delay(
document_dict, request.workspace_id, str(user.id) 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: else:
raise HTTPException(status_code=400, detail="Invalid document type") raise HTTPException(status_code=400, detail="Invalid document type")

126
surfsense_backend/app/tasks/celery_tasks/document_tasks.py
View file

@ -19,7 +19,6 @@ from app.tasks.connector_indexers.local_folder_indexer import (
) )
from app.tasks.document_processors import ( from app.tasks.document_processors import (
add_extension_received_document, add_extension_received_document,
add_youtube_video_document,
) )
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -405,131 +404,6 @@ async def _process_extension_document(
raise 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) @celery_app.task(name="process_file_upload", bind=True)
def process_file_upload_task( def process_file_upload_task(
self, file_path: str, filename: str, workspace_id: int, user_id: str self, file_path: str, filename: str, workspace_id: int, user_id: str

4
surfsense_backend/app/tasks/document_processors/__init__.py
View file

@ -3,15 +3,13 @@ Document processors module for background tasks.
Content extraction is handled by ``app.etl_pipeline.EtlPipelineService``. Content extraction is handled by ``app.etl_pipeline.EtlPipelineService``.
This package keeps orchestration (save, notify, page-limit) and 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 .extension_processor import add_extension_received_document
from .markdown_processor import add_received_markdown_file_document from .markdown_processor import add_received_markdown_file_document
from .youtube_processor import add_youtube_video_document
__all__ = [ __all__ = [
"add_extension_received_document", "add_extension_received_document",
"add_received_markdown_file_document", "add_received_markdown_file_document",
"add_youtube_video_document",
] ]

473
surfsense_backend/app/tasks/document_processors/youtube_processor.py
View file

@ -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("<DOCUMENT>")
for section_title, section_content in metadata_sections:
document_parts.append(f"<{section_title}>")
document_parts.extend(section_content)
document_parts.append(f"</{section_title}>")
document_parts.append("</DOCUMENT>")
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

1
surfsense_backend/tests/unit/observability/test_helpers.py
View file

@ -26,7 +26,6 @@ def _disable_otel(monkeypatch: pytest.MonkeyPatch):
("delete_folder_documents_background", "delete"), ("delete_folder_documents_background", "delete"),
("delete_search_space_background", "delete"), ("delete_search_space_background", "delete"),
("process_extension_document", "process"), ("process_extension_document", "process"),
("process_youtube_video", "process"),
("process_file_upload", "process"), ("process_file_upload", "process"),
("process_file_upload_with_document", "process"), ("process_file_upload_with_document", "process"),
("process_circleback_meeting", "process"), ("process_circleback_meeting", "process"),