from fastapi import APIRouter, Depends, HTTPException, status, UploadFile, File, BackgroundTasks, Form, Request from fastapi.responses import Response from pydantic import BaseModel import threading import subprocess import asyncio import sys import random import uuid import requests import hashlib import json from fastapi.security import OAuth2PasswordBearer from sqlalchemy.orm import Session from typing import List, Optional from uuid import UUID from ..database import get_db from ..models.stream import Stream from ..models.user import User from ..models.processing_job import ProcessingJob from ..schemas.stream import StreamCreate, StreamUpdate, StreamResponse, StreamManualCreate from ..auth import get_current_active_user, get_current_admin_user, get_current_creator_user, get_current_user, get_current_user_optional, verify_password, require_permission, require_any_permission from ..config import settings from ..services.twitch_service import TwitchService from ..services.chat_service import ChatService from ..services.chat_json_service import ChatJsonService import os import shutil import re from datetime import datetime, timezone from ..redis_client import redis_client router = APIRouter(prefix="/api/streams", tags=["streams"]) # Global dictionary to track thumbnail generation progress thumbnail_progress: dict[str, dict] = {} thumbnail_progress_lock = threading.Lock() def generate_etag(data: List) -> str: """Generate ETag hash from stream data.""" data_str = json.dumps([{ 'id': str(s.id), 'updated_at': s.updated_at.isoformat() if s.updated_at else None, 'status': s.status } for s in data], sort_keys=True) return hashlib.md5(data_str.encode()).hexdigest() def check_etag(request: Request, data: List) -> Optional[Response]: """Check ETag and return 304 if not modified.""" etag = generate_etag(data) if_none_match = request.headers.get("if-none-match") if if_none_match == etag: return Response(status_code=304) return None def stream_to_response(stream: Stream) -> dict: """Convert SQLAlchemy Stream to dict for JSON serialization.""" return { 'id': str(stream.id), 'title': stream.title, 'description': stream.description, 'status': stream.status, 'created_by': str(stream.created_by) if stream.created_by else None, 'created_at': stream.created_at.isoformat() if stream.created_at else None, 'updated_at': stream.updated_at.isoformat() if stream.updated_at else None, 'duration_seconds': stream.duration_seconds, 'thumbnail_url': stream.thumbnail_url, 'hls_path': stream.hls_path, 'hls_url': stream.hls_url, 'hls_available': stream.hls_available, 'thumbnails_available': stream.thumbnails_available, 'dash_path': stream.dash_path, 'file_path': stream.file_path, 'view_count': stream.view_count, 'is_public': stream.is_public, 'stream_date': stream.stream_date.isoformat() if stream.stream_date else None, 'twitch_vod_id': stream.twitch_vod_id, 'streamer': stream.streamer, 'stream_metadata': stream.stream_metadata, 'thumbnail_progress_generated': stream.thumbnail_progress_generated, 'thumbnail_progress_total': stream.thumbnail_progress_total, 'thumbnail_progress_message': stream.thumbnail_progress_message, 'thumbnail_progress_error': stream.thumbnail_progress_error, 'thumbnail_progress_updated_at': stream.thumbnail_progress_updated_at.isoformat() if stream.thumbnail_progress_updated_at else None, 'stream_number': stream.stream_number } def _build_hls_url(hls_path: str) -> str: """Build consistent HLS URL from path. Handles both directory paths (appends /index.m3u8) and direct .m3u8 files. Converts to absolute HTTPS URL if DOMAIN is configured. """ if not hls_path: return None # Convert cache path to URL path url_path = hls_path.replace(settings.FFMPEG_CACHE_PATH, "/hls") # Add /index.m3u8 if it's a directory if not url_path.endswith('.m3u8'): url_path = f"{url_path.rstrip('/')}/index.m3u8" # Convert to absolute HTTPS URL if not url_path.startswith('http'): url_path = f"https://{settings.DOMAIN}{url_path}" if settings.DOMAIN else url_path return url_path def _is_hls_available(stream) -> bool: """Check if HLS playlist is available for playback.""" if not stream.hls_path: return False # Allow HLS availability for ready, segmenting, and resegmenting status if stream.status not in ["ready", "segmenting", "resegmenting"]: return False # Check if the HLS directory or file exists hls_path = stream.hls_path # Convert absolute path to filesystem path if hls_path.startswith('/uploads/'): hls_path = hls_path.replace('/uploads/', '/app/uploads/') elif hls_path.startswith('/hls/'): hls_path = hls_path.replace('/hls/', settings.FFMPEG_CACHE_PATH + '/') # Check if directory exists if os.path.isdir(hls_path): # Check if index.m3u8 exists m3u8_path = os.path.join(hls_path, 'index.m3u8') if os.path.exists(m3u8_path): return True # Check if it's a direct .m3u8 file elif hls_path.endswith('.m3u8') and os.path.exists(hls_path): return True return False def _is_thumbnails_available(stream) -> bool: """Check if thumbnail sprite(s) are available (legacy single sprite or multi-sprite via JSON).""" if not stream.id: return False video_dir = os.path.join(settings.UPLOAD_PATH, 'videos', str(stream.id)) # Check for legacy single sprite if os.path.exists(os.path.join(video_dir, 'thumbnails_sprite.jpg')): return True # Check for multi-sprite metadata (JSON) if os.path.exists(os.path.join(video_dir, 'thumbnails_sprite.json')): return True return False @router.get("", response_model=List[StreamResponse]) async def list_streams( skip: int = 0, limit: int = 50, db: Session = Depends(get_db), current_user: Optional[User] = Depends(get_current_user_optional) ): if current_user: streams = db.query(Stream).filter( (Stream.is_public == True) | (Stream.created_by == current_user.id) ).offset(skip).limit(limit).all() else: streams = db.query(Stream).filter(Stream.is_public == True).offset(skip).limit(limit).all() # Add hls_url to each stream for stream in streams: stream.hls_url = _build_hls_url(stream.hls_path) stream.hls_available = _is_hls_available(stream) stream.thumbnails_available = _is_thumbnails_available(stream) # Convert thumbnail_url to absolute HTTPS URL if stream.thumbnail_url and not stream.thumbnail_url.startswith('http'): stream.thumbnail_url = f"https://{settings.DOMAIN}{stream.thumbnail_url}" if settings.DOMAIN else stream.thumbnail_url # Convert file_path to absolute HTTPS URL if stream.file_path and not stream.file_path.startswith('http'): file_path = stream.file_path.replace('/app/uploads', '/uploads') if not file_path.startswith('http'): file_path = f"https://{settings.DOMAIN}{file_path}" if settings.DOMAIN else file_path stream.file_path = file_path return streams @router.get("/latest") async def get_latest_streams( request: Request, limit: int = 12, db: Session = Depends(get_db), current_user: Optional[User] = Depends(get_current_user_optional) ): """Get latest streams by stream_date (or created_at as fallback)""" if current_user: streams = db.query(Stream).filter( (Stream.is_public == True) | (Stream.created_by == current_user.id) ).order_by(Stream.stream_date.desc().nulls_last(), Stream.created_at.desc()).limit(limit).all() else: streams = db.query(Stream).filter(Stream.is_public == True).order_by( Stream.stream_date.desc().nulls_last(), Stream.created_at.desc() ).limit(limit).all() for stream in streams: stream.hls_url = _build_hls_url(stream.hls_path) stream.hls_available = _is_hls_available(stream) stream.thumbnails_available = _is_thumbnails_available(stream) # Convert thumbnail_url to absolute HTTPS URL if stream.thumbnail_url and not stream.thumbnail_url.startswith('http'): stream.thumbnail_url = f"https://{settings.DOMAIN}{stream.thumbnail_url}" if settings.DOMAIN else stream.thumbnail_url # Convert file_path to absolute HTTPS URL if stream.file_path and not stream.file_path.startswith('http'): file_path = stream.file_path.replace('/app/uploads', '/uploads') if not file_path.startswith('http'): file_path = f"https://{settings.DOMAIN}{file_path}" if settings.DOMAIN else file_path stream.file_path = file_path # Check ETag etag_response = check_etag(request, streams) if etag_response: return etag_response # Add ETag header to response etag = generate_etag(streams) response_data = [stream_to_response(s) for s in streams] response = Response(content=json.dumps(response_data), media_type="application/json") response.headers["ETag"] = etag return response @router.get("/popular") async def get_popular_streams( request: Request, limit: int = 12, db: Session = Depends(get_db), current_user: Optional[User] = Depends(get_current_user_optional) ): """Get popular streams by view_count""" if current_user: streams = db.query(Stream).filter( (Stream.is_public == True) | (Stream.created_by == current_user.id) ).order_by(Stream.view_count.desc()).limit(limit).all() else: streams = db.query(Stream).filter(Stream.is_public == True).order_by( Stream.view_count.desc() ).limit(limit).all() for stream in streams: stream.hls_url = _build_hls_url(stream.hls_path) stream.hls_available = _is_hls_available(stream) stream.thumbnails_available = _is_thumbnails_available(stream) # Convert thumbnail_url to absolute HTTPS URL if stream.thumbnail_url and not stream.thumbnail_url.startswith('http'): stream.thumbnail_url = f"https://{settings.DOMAIN}{stream.thumbnail_url}" if settings.DOMAIN else stream.thumbnail_url # Convert file_path to absolute HTTPS URL if stream.file_path and not stream.file_path.startswith('http'): file_path = stream.file_path.replace('/app/uploads', '/uploads') if not file_path.startswith('http'): file_path = f"https://{settings.DOMAIN}{file_path}" if settings.DOMAIN else file_path stream.file_path = file_path # Check ETag etag_response = check_etag(request, streams) if etag_response: return etag_response # Add ETag header to response etag = generate_etag(streams) response_data = [stream_to_response(s) for s in streams] response = Response(content=json.dumps(response_data), media_type="application/json") response.headers["ETag"] = etag return response @router.get("/random") async def get_random_streams( request: Request, limit: int = 12, db: Session = Depends(get_db), current_user: Optional[User] = Depends(get_current_user_optional) ): """Get random streams for discovery""" if current_user: all_streams = db.query(Stream).filter( (Stream.is_public == True) | (Stream.created_by == current_user.id) ).all() else: all_streams = db.query(Stream).filter(Stream.is_public == True).all() # Randomly select streams streams = random.sample(all_streams, min(limit, len(all_streams))) for stream in streams: stream.hls_url = _build_hls_url(stream.hls_path) stream.hls_available = _is_hls_available(stream) stream.thumbnails_available = _is_thumbnails_available(stream) # Convert thumbnail_url to absolute HTTPS URL if stream.thumbnail_url and not stream.thumbnail_url.startswith('http'): stream.thumbnail_url = f"https://{settings.DOMAIN}{stream.thumbnail_url}" if settings.DOMAIN else stream.thumbnail_url # Convert file_path to absolute HTTPS URL if stream.file_path and not stream.file_path.startswith('http'): file_path = stream.file_path.replace('/app/uploads', '/uploads') if not file_path.startswith('http'): file_path = f"https://{settings.DOMAIN}{file_path}" if settings.DOMAIN else file_path stream.file_path = file_path # Check ETag etag_response = check_etag(request, streams) if etag_response: return etag_response # Add ETag header to response etag = generate_etag(streams) response_data = [stream_to_response(s) for s in streams] response = Response(content=json.dumps(response_data), media_type="application/json") response.headers["ETag"] = etag return response @router.get("/admin/all", response_model=List[StreamResponse]) async def list_all_streams_admin( skip: int = 0, limit: int = 50, db: Session = Depends(get_db), current_user: User = Depends(get_current_admin_user) ): """Admin-only endpoint to see all streams for dashboard""" streams = db.query(Stream).offset(skip).limit(limit).all() for stream in streams: stream.hls_url = _build_hls_url(stream.hls_path) stream.hls_available = _is_hls_available(stream) stream.thumbnails_available = _is_thumbnails_available(stream) # Convert thumbnail_url to absolute HTTPS URL if stream.thumbnail_url and not stream.thumbnail_url.startswith('http'): stream.thumbnail_url = f"https://{settings.DOMAIN}{stream.thumbnail_url}" if settings.DOMAIN else stream.thumbnail_url # Convert file_path to absolute HTTPS URL if stream.file_path and not stream.file_path.startswith('http'): file_path = stream.file_path.replace('/app/uploads', '/uploads') if not file_path.startswith('http'): file_path = f"https://{settings.DOMAIN}{file_path}" if settings.DOMAIN else file_path stream.file_path = file_path return streams @router.get("/search", response_model=List[StreamResponse]) async def search_streams( q: str, limit: int = 20, db: Session = Depends(get_db), current_user: Optional[User] = Depends(get_current_user_optional) ): """Search streams by title, streamer, or game (from chapters or metadata)""" if not q or len(q.strip()) < 2: return [] search_term = f"%{q.strip().lower()}%" # Build base query with visibility filter if current_user: query = db.query(Stream).filter( (Stream.is_public == True) | (Stream.created_by == current_user.id) ) else: query = db.query(Stream).filter(Stream.is_public == True) # Search in title and streamer query = query.filter( (Stream.title.ilike(search_term)) | (Stream.streamer.ilike(search_term)) ) streams = query.limit(limit).all() # Also search in chapters table for game names from ..models.chapter import Chapter if current_user: chapter_query = db.query(Stream).join(Chapter).filter( (Stream.is_public == True) | (Stream.created_by == current_user.id) ).filter(Chapter.game_display_name.ilike(search_term)) else: chapter_query = db.query(Stream).join(Chapter).filter( Stream.is_public == True ).filter(Chapter.game_display_name.ilike(search_term)) chapter_streams = chapter_query.limit(limit).all() # Merge results (avoid duplicates) stream_ids = {s.id for s in streams} for stream in chapter_streams: if stream.id not in stream_ids: streams.append(stream) stream_ids.add(stream.id) # Fallback: Search in stream_metadata JSON field for game/chapters # This is for streams that haven't been migrated yet all_streams_query = db.query(Stream) if current_user: all_streams_query = all_streams_query.filter( (Stream.is_public == True) | (Stream.created_by == current_user.id) ) else: all_streams_query = all_streams_query.filter(Stream.is_public == True) all_streams = all_streams_query.all() for stream in all_streams: if stream.id in stream_ids: continue # Check metadata for game or chapters if stream.stream_metadata: metadata = stream.stream_metadata game_match = False chapter_match = False # Check game field if metadata.get('game', '').lower().find(search_term.replace('%', '')) != -1: game_match = True # Check chapters if not game_match and metadata.get('chapters'): for chapter in metadata['chapters']: if chapter.get('gameDisplayName', '').lower().find(search_term.replace('%', '')) != -1: chapter_match = True break if game_match or chapter_match: streams.append(stream) stream_ids.add(stream.id) # Limit final results streams = streams[:limit] for stream in streams: stream.hls_url = _build_hls_url(stream.hls_path) stream.hls_available = _is_hls_available(stream) stream.thumbnails_available = _is_thumbnails_available(stream) # Convert thumbnail_url to absolute HTTPS URL if stream.thumbnail_url and not stream.thumbnail_url.startswith('http'): stream.thumbnail_url = f"https://{settings.DOMAIN}{stream.thumbnail_url}" if settings.DOMAIN else stream.thumbnail_url # Convert file_path to absolute HTTPS URL if stream.file_path and not stream.file_path.startswith('http'): file_path = stream.file_path.replace('/app/uploads', '/uploads') if not file_path.startswith('http'): file_path = f"https://{settings.DOMAIN}{file_path}" if settings.DOMAIN else file_path stream.file_path = file_path return streams @router.get("/{stream_id}", response_model=StreamResponse) async def get_stream( stream_id: UUID, db: Session = Depends(get_db), current_user: Optional[User] = Depends(get_current_user_optional) ): stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Stream not found" ) # Check access - public streams can be accessed without authentication if not stream.is_public: if not current_user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required for private streams" ) if stream.created_by != current_user.id and current_user.role.name != "admin": raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) # Add hls_url with absolute HTTPS URL stream.hls_url = _build_hls_url(stream.hls_path) stream.hls_available = _is_hls_available(stream) stream.thumbnails_available = _is_thumbnails_available(stream) # Convert thumbnail_url to absolute HTTPS URL if stream.thumbnail_url and not stream.thumbnail_url.startswith('http'): stream.thumbnail_url = f"https://{settings.DOMAIN}{stream.thumbnail_url}" if settings.DOMAIN else stream.thumbnail_url # Convert file_path to absolute HTTPS URL if stream.file_path and not stream.file_path.startswith('http'): file_path = stream.file_path.replace('/app/uploads', '/uploads') if not file_path.startswith('http'): file_path = f"https://{settings.DOMAIN}{file_path}" if settings.DOMAIN else file_path stream.file_path = file_path return stream @router.post("/{stream_id}/play", response_model=StreamResponse) async def increment_view_count( stream_id: UUID, db: Session = Depends(get_db), current_user: Optional[User] = Depends(get_current_user_optional) ): """Increment view count when a video is actually played""" stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Stream not found" ) # Check access if not stream.is_public: if not current_user: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) if stream.created_by != current_user.id and current_user.role.name != "admin": raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) # Increment view count stream.view_count += 1 db.commit() db.refresh(stream) # Add hls_url with absolute HTTPS URL stream.hls_url = _build_hls_url(stream.hls_path) stream.hls_available = _is_hls_available(stream) stream.thumbnails_available = _is_thumbnails_available(stream) # Convert thumbnail_url to absolute HTTPS URL if stream.thumbnail_url and not stream.thumbnail_url.startswith('http'): stream.thumbnail_url = f"https://{settings.DOMAIN}{stream.thumbnail_url}" if settings.DOMAIN else stream.thumbnail_url # Convert file_path to absolute HTTPS URL if stream.file_path and not stream.file_path.startswith('http'): file_path = stream.file_path.replace('/app/uploads', '/uploads') if not file_path.startswith('http'): file_path = f"https://{settings.DOMAIN}{file_path}" if settings.DOMAIN else file_path stream.file_path = file_path return stream @router.post("", response_model=StreamResponse, status_code=status.HTTP_201_CREATED) async def create_stream( stream_data: StreamCreate, db: Session = Depends(get_db), current_user: User = Depends(require_permission("stream:create")) ): new_stream = Stream( title=stream_data.title, description=stream_data.description, created_by=current_user.id, is_public=stream_data.is_public ) db.add(new_stream) db.commit() # Renumber all streams after creation try: from ..services.stream_numbering_service import renumber_all_streams renumber_all_streams() except Exception as e: sys.stderr.write(f"[NUMBERING ERROR] Failed to renumber after create: {e}\n") @router.put("/{stream_id}", response_model=StreamResponse) async def update_stream( stream_id: UUID, stream_data: StreamUpdate, db: Session = Depends(get_db), current_user: User = Depends(require_any_permission(["stream:update", "stream:update:own"])) ): stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Stream not found" ) # Check ownership: Creator can only edit their own streams, Admins can edit all if current_user.role.name not in ["admin", "creator"] and stream.created_by != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) # Creators can only edit their own streams if current_user.role.name == "creator" and stream.created_by != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) # Update fields if stream_data.title is not None: stream.title = stream_data.title if stream_data.description is not None: stream.description = stream_data.description if stream_data.is_public is not None: stream.is_public = stream_data.is_public if stream_data.twitch_vod_id is not None: # Extract VOD ID if full URL is provided twitch_vod_id = stream_data.twitch_vod_id if 'twitch.tv/videos/' in twitch_vod_id: from ..services.twitch_service import TwitchService twitch_service = TwitchService() extracted_id = twitch_service.extract_vod_id(twitch_vod_id) if extracted_id: twitch_vod_id = extracted_id stream.twitch_vod_id = twitch_vod_id if stream_data.streamer is not None: stream.streamer = stream_data.streamer if stream_data.view_count is not None: stream.view_count = stream_data.view_count # Update HLS path and status if provided if stream_data.hls_path is not None: stream.hls_path = stream_data.hls_path # Update status based on HLS path presence stream.status = "ready" if stream_data.hls_path else "pending" if stream.status == "ready": stream.thumbnail_progress_generated = 0 stream.thumbnail_progress_total = 0 stream.thumbnail_progress_message = None stream.thumbnail_progress_error = None stream.thumbnail_progress_updated_at = None # Update thumbnail_url if provided if stream_data.thumbnail_url is not None: timestamp = int(datetime.now(timezone.utc).timestamp()) stream.thumbnail_url = f"/uploads/thumbnails/{stream_id}.jpg?t={timestamp}" stream.thumbnail_updated_at = datetime.now(timezone.utc) db.commit() # Regenerate master ffconcat if HLS path changed (adds, updates or removes stream) if stream_data.hls_path is not None: try: from ..services.ffconcat_service import regenerate_master_ffconcat regenerate_master_ffconcat() except Exception as e: sys.stderr.write(f"[FFCONCAT ERROR] Failed to regenerate after update: {e}\n") db.refresh(stream) # Renumber if stream_date may have changed try: from ..services.stream_numbering_service import renumber_all_streams renumber_all_streams() except Exception as e: sys.stderr.write(f"[NUMBERING ERROR] Failed to renumber after update: {e}\n") return stream @router.put("/{stream_id}/metadata", response_model=StreamResponse) async def update_stream_metadata( stream_id: UUID, metadata: dict, db: Session = Depends(get_db), current_user: User = Depends(get_current_admin_user) ): stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Stream not found" ) # Update metadata stream.stream_metadata = metadata stream.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(stream) # Renumber after metadata change try: from ..services.stream_numbering_service import renumber_all_streams renumber_all_streams() except Exception as e: sys.stderr.write(f"[NUMBERING ERROR] Failed to renumber after metadata update: {e}\n") return stream @router.delete("/{stream_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_stream( stream_id: UUID, db: Session = Depends(get_db), current_user: User = Depends(require_any_permission(["stream:delete", "stream:delete:own"])), background_tasks: BackgroundTasks = None ): stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Stream not found" ) # Admins can delete all streams if current_user.role.name == "admin": pass # Allow deletion # Creators can only delete their own streams elif current_user.role.name == "creator": if stream.created_by != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) # Viewers cannot delete streams else: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) # Store file paths for background deletion file_path = stream.file_path hls_path = stream.hls_path twitch_vod_id = stream.twitch_vod_id stream_id = str(stream.id) # Delete from database immediately db.delete(stream) db.commit() # Regenerate master ffconcat try: from ..services.ffconcat_service import regenerate_master_ffconcat regenerate_master_ffconcat() except Exception as e: sys.stderr.write(f"[FFCONCAT ERROR] Failed to regenerate after delete: {e}\n") # Delete files in background to avoid blocking if background_tasks: background_tasks.add_task(delete_stream_files, file_path, hls_path, twitch_vod_id, stream_id) # Renumber after deletion try: from ..services.stream_numbering_service import renumber_all_streams renumber_all_streams() except Exception as e: sys.stderr.write(f"[NUMBERING ERROR] Failed to renumber after delete: {e}\n") def delete_stream_files(file_path: str, hls_path: str, twitch_vod_id: str, stream_id: str): """Delete stream files in background""" import os import shutil import sys from ..config import settings delete_error = None # 1. Delete main video file (MP4) if file_path and os.path.exists(file_path): try: os.remove(file_path) except Exception as e: sys.stderr.write(f"[DELETE] Failed to delete file_path {file_path}: {e}\n") sys.stderr.flush() delete_error = f"Failed to delete video file: {str(e)}" # 2. Delete HLS directory (contains .m3u8, .ts files, and thumbnails) if hls_path and os.path.exists(hls_path): try: shutil.rmtree(hls_path, ignore_errors=True) except Exception as e: sys.stderr.write(f"[DELETE] Failed to delete hls_path {hls_path}: {e}\n") sys.stderr.flush() if not delete_error: delete_error = f"Failed to delete HLS directory: {str(e)}" # 3. Delete chat JSON file from cache directory if twitch_vod_id: chat_file = os.path.join("/app/twitch_cache", f"{twitch_vod_id}_chat.json") if os.path.exists(chat_file): try: os.remove(chat_file) except Exception as e: sys.stderr.write(f"[DELETE] Failed to delete chat file {chat_file}: {e}\n") sys.stderr.flush() if not delete_error: delete_error = f"Failed to delete chat file: {str(e)}" # 4. Delete temporary download files (if they exist) if twitch_vod_id: temp_files = [ os.path.join("/app/twitch_cache", f"{twitch_vod_id}.mp4"), os.path.join("/app/twitch_cache", f"{twitch_vod_id}.part"), os.path.join("/app/twitch_cache", f"download_{twitch_vod_id}"), ] for temp_file in temp_files: if os.path.exists(temp_file): try: if os.path.isdir(temp_file): shutil.rmtree(temp_file, ignore_errors=True) else: os.remove(temp_file) except Exception as e: sys.stderr.write(f"[DELETE] Failed to delete temp file {temp_file}: {e}\n") sys.stderr.flush() if not delete_error: delete_error = f"Failed to delete temp files: {str(e)}" # 5. Delete any stream-specific directories in uploads stream_upload_dir = os.path.join(settings.UPLOAD_PATH, stream_id) if os.path.exists(stream_upload_dir): try: shutil.rmtree(stream_upload_dir, ignore_errors=True) except Exception as e: sys.stderr.write(f"[DELETE] Failed to delete upload dir {stream_upload_dir}: {e}\n") sys.stderr.flush() if not delete_error: delete_error = f"Failed to delete upload directory: {str(e)}" # 6. Delete uploads/videos/{stream_id} directory stream_videos_dir = os.path.join(settings.UPLOAD_PATH, "videos", stream_id) if os.path.exists(stream_videos_dir): try: shutil.rmtree(stream_videos_dir, ignore_errors=True) except Exception as e: sys.stderr.write(f"[DELETE] Failed to delete videos dir {stream_videos_dir}: {e}\n") sys.stderr.flush() if not delete_error: delete_error = f"Failed to delete videos directory: {str(e)}" if delete_error: sys.stderr.write(f"[DELETE] Stream deletion completed with errors: {delete_error}\n") sys.stderr.flush() else: sys.stderr.write("[DELETE] Stream deletion completed successfully\n") sys.stderr.flush() @router.post("/{stream_id}/upload") async def upload_video( stream_id: UUID, file: UploadFile = File(...), db: Session = Depends(get_db), current_user: User = Depends(require_permission("stream:create")) ): stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Stream not found" ) # Check ownership: Creator can only upload to their own streams, Admins can upload to all if current_user.role.name == "creator" and stream.created_by != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) # Create storage directory if not exists os.makedirs(settings.STORAGE_PATH, exist_ok=True) # Save file file_extension = file.filename.split(".")[-1] file_path = os.path.join(settings.STORAGE_PATH, f"{stream_id}.{file_extension}") with open(file_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) # Update stream stream.file_path = file_path stream.status = "processing" stream.updated_at = datetime.now(timezone.utc) db.commit() # Create processing job processing_job = ProcessingJob( stream_id=stream_id, job_type="transcode", status="pending" ) db.add(processing_job) db.commit() return {"message": "File uploaded successfully", "file_path": file_path} @router.post("/twitch-import", response_model=StreamResponse, status_code=status.HTTP_201_CREATED) async def import_from_twitch( twitch_url: str, title: Optional[str] = None, description: Optional[str] = None, download_video: bool = True, # Changed default to True embed_emotes: bool = True, db: Session = Depends(get_db), current_user: User = Depends(require_permission("stream:create")) ): """Import a stream from Twitch using Twitch API""" twitch_service = TwitchService() chat_service = ChatService(db) try: # Extract VOD ID vod_id = twitch_service.extract_vod_id(twitch_url) if not vod_id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid Twitch URL or ID" ) # Get stream info stream_info = await twitch_service.get_stream_info(twitch_url) # Create stream entry # Parse stream_date from Twitch created_at from datetime import datetime stream_date = None if stream_info.get('created_at'): try: stream_date = datetime.fromisoformat(stream_info['created_at'].replace('Z', '+00:00')) except: pass new_stream = Stream( title=title or stream_info.get('title', f'Twitch Stream {vod_id}'), description=description or stream_info.get('description', ''), created_by=current_user.id, is_public=True, # Make public immediately status="downloading", # Status: downloading -> processing -> ready duration_seconds=stream_info.get('duration_seconds', 0), view_count=0, # Always start with 0 views thumbnail_url=stream_info.get('thumbnail_url'), twitch_vod_id=vod_id, streamer=stream_info.get('streamer', ''), stream_date=stream_date ) db.add(new_stream) db.commit() db.refresh(new_stream) # Download chat synchronously try: chat_data = await twitch_service.download_chat(twitch_url, embed_emotes=embed_emotes, include_metadata=True) chat_messages = chat_data.get('messages', []) metadata = chat_data.get('metadata') sys.stderr.write(f"[CHAT] Downloaded {len(chat_messages)} chat messages\n") sys.stderr.flush() # Save chat messages for msg in chat_messages: chat_service.create_chat_message( stream_id=new_stream.id, username=msg['username'], message=msg['message'], timestamp_offset=msg['timestamp_offset'], color=msg['color'], emotes=msg['emotes'], badges=msg['badges'] ) # Save metadata if available if metadata: new_stream.stream_metadata = metadata # Save chapters to dedicated table if metadata.get('chapters'): twitch_service.save_chapters_to_db(str(new_stream.id), metadata['chapters'], db) db.commit() except Exception as e: sys.stderr.write(f"[CHAT ERROR] Chat download failed: {e}\n") sys.stderr.flush() # Download VOD if requested if download_video: try: # Define output path (HLS format for Twitch VODs) # Create subdirectory for each stream to keep things organized stream_dir = os.path.join(settings.UPLOAD_PATH, "videos", str(new_stream.id)) os.makedirs(stream_dir, exist_ok=True) output_path = os.path.join(stream_dir, f"{new_stream.id}.m3u8") # Download VOD in background thread def download_and_process(): import asyncio loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: # Download VOD result = loop.run_until_complete( twitch_service.download_video(twitch_url, output_path, str(new_stream.id)) ) # Update stream with HLS path (no FFmpeg processing needed for Twitch VODs) from ..database import SessionLocal db_thread = SessionLocal() try: stream = db_thread.query(Stream).filter(Stream.id == new_stream.id).first() if stream: # Convert absolute path to relative path for frontend hls_path = result['file_path'].replace('/app/uploads', '/uploads') stream.hls_path = hls_path stream.status = "ready" db_thread.commit() finally: db_thread.close() # Regenerate master ffconcat try: from ..services.ffconcat_service import regenerate_master_ffconcat regenerate_master_ffconcat() except Exception as e: sys.stderr.write(f"[FFCONCAT ERROR] Failed to regenerate: {e}\n") except Exception as e: sys.stderr.write(f"[DOWNLOAD ERROR] VOD download/processing failed: {e}\n") sys.stderr.flush() # Update status to error from ..database import SessionLocal db_thread = SessionLocal() try: stream = db_thread.query(Stream).filter(Stream.id == new_stream.id).first() if stream: stream.status = "error" db_thread.commit() finally: db_thread.close() finally: loop.close() # Start download in background thread = threading.Thread(target=download_and_process) thread.daemon = True thread.start() sys.stderr.write(f"[DOWNLOAD] Started VOD download in background for stream {new_stream.id}\n") sys.stderr.flush() except Exception as e: sys.stderr.write(f"[DOWNLOAD ERROR] Failed to start VOD download: {e}\n") sys.stderr.flush() else: # No video download, mark as ready new_stream.status = "ready" db.commit() # Regenerate master ffconcat try: from ..services.ffconcat_service import regenerate_master_ffconcat regenerate_master_ffconcat() except Exception as e: sys.stderr.write(f"[FFCONCAT ERROR] Failed to regenerate: {e}\n") # Renumber after twitch import try: from ..services.stream_numbering_service import renumber_all_streams renumber_all_streams() except Exception as e: sys.stderr.write(f"[NUMBERING ERROR] Failed to renumber after twitch import: {e}\n") return new_stream except ValueError as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=str(e) ) except RuntimeError as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e) ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Unexpected error: {str(e)}" ) class ReDownloadRequest(BaseModel): download_video: bool = False download_chat: bool = True twitch_vod_id: Optional[str] = None @router.post("/{stream_id}/re-download", response_model=StreamResponse) async def re_download_stream( stream_id: UUID, request: ReDownloadRequest, db: Session = Depends(get_db), current_user: User = Depends(get_current_active_user) ): """Re-download chat and optionally video for an existing stream with metadata""" stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Stream not found" ) # Check ownership: Creator can only re-download their own streams, Admins can re-download all if current_user.role.name not in ["admin", "creator"] and stream.created_by != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) # Creators can only re-download their own streams if current_user.role.name == "creator" and stream.created_by != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) # Use provided twitch_vod_id or fall back to stream's twitch_vod_id vod_id_to_use = request.twitch_vod_id or stream.twitch_vod_id if not vod_id_to_use: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="No Twitch VOD ID provided. Please enter a Twitch VOD ID or URL in the form and save first." ) twitch_service = TwitchService() chat_service = ChatService(db) try: # Extract VOD ID if full URL provided if 'twitch.tv/videos/' in vod_id_to_use: vod_id_to_use = twitch_service.extract_vod_id(vod_id_to_use) # Reconstruct Twitch URL twitch_url = f"https://www.twitch.tv/videos/{vod_id_to_use}" # Download chat with metadata if request.download_chat: try: chat_data = await twitch_service.download_chat(twitch_url, embed_emotes=True, include_metadata=True) chat_messages = chat_data.get('messages', []) metadata = chat_data.get('metadata') sys.stderr.write(f"[CHAT] Downloaded {len(chat_messages)} chat messages with metadata\n") sys.stderr.flush() # Delete old chat messages chat_service.delete_chat_messages(stream_id) # Save new chat messages for msg in chat_messages: chat_service.create_chat_message( stream_id=stream_id, username=msg['username'], message=msg['message'], timestamp_offset=msg['timestamp_offset'], color=msg['color'], emotes=msg['emotes'], badges=msg['badges'] ) # Update stream metadata if available if metadata: stream.stream_metadata = metadata # Save chapters to dedicated table if metadata.get('chapters'): twitch_service.save_chapters_to_db(str(stream.id), metadata['chapters'], db) db.commit() # Update twitch_vod_id if provided if request.twitch_vod_id and request.twitch_vod_id != stream.twitch_vod_id: stream.twitch_vod_id = vod_id_to_use db.commit() except Exception as e: sys.stderr.write(f"[CHAT ERROR] Chat re-download failed: {e}\n") sys.stderr.flush() # Re-download video if requested if request.download_video: try: # Clean up entire HLS directory before re-download stream_dir = os.path.join(settings.UPLOAD_PATH, "videos", str(stream_id)) if os.path.exists(stream_dir): import shutil sys.stderr.write(f"[CLEANUP] Removing old HLS directory: {stream_dir}\n") sys.stderr.flush() try: shutil.rmtree(stream_dir) except Exception as e: sys.stderr.write(f"[CLEANUP WARNING] Failed to remove directory: {e}\n") sys.stderr.flush() # Create subdirectory for each stream to keep things organized os.makedirs(stream_dir, exist_ok=True) output_path = os.path.join(stream_dir, f"{stream_id}.m3u8") # Download VOD in background thread def download_and_process(): import asyncio loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: # Download VOD result = loop.run_until_complete( twitch_service.download_video(twitch_url, output_path, str(stream_id)) ) # Update stream with HLS path (no FFmpeg processing needed for Twitch VODs) from ..database import SessionLocal db_thread = SessionLocal() try: stream = db_thread.query(Stream).filter(Stream.id == stream_id).first() if stream: # Convert absolute path to relative path for frontend hls_path = result['file_path'].replace('/app/uploads', '/uploads') stream.hls_path = hls_path stream.status = "ready" db_thread.commit() finally: db_thread.close() # Regenerate master ffconcat try: from ..services.ffconcat_service import regenerate_master_ffconcat regenerate_master_ffconcat() except Exception as e: sys.stderr.write(f"[FFCONCAT ERROR] Failed to regenerate: {e}\n") except Exception as e: sys.stderr.write(f"[DOWNLOAD ERROR] VOD re-download/processing failed: {e}\n") sys.stderr.flush() from ..database import SessionLocal db_thread = SessionLocal() try: stream = db_thread.query(Stream).filter(Stream.id == stream_id).first() if stream: stream.status = "error" db_thread.commit() finally: db_thread.close() finally: loop.close() # Start download in background thread = threading.Thread(target=download_and_process) thread.daemon = True thread.start() stream.status = "downloading" db.commit() except Exception as e: sys.stderr.write(f"[DOWNLOAD ERROR] Failed to start VOD re-download: {e}\n") sys.stderr.flush() db.refresh(stream) # Renumber after re-download try: from ..services.stream_numbering_service import renumber_all_streams renumber_all_streams() except Exception as e: sys.stderr.write(f"[NUMBERING ERROR] Failed to renumber after re-download: {e}\n") return stream except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Re-download failed: {str(e)}" ) @router.post("/{stream_id}/import-chat-json", response_model=StreamResponse) async def import_chat_json_for_stream( stream_id: UUID, chat_json_file: UploadFile = File(...), db: Session = Depends(get_db), current_user: User = Depends(get_current_active_user) ): """Import chat.json (chapters and chat messages) for an existing stream""" stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Stream not found" ) # Check ownership: Creator can only import their own streams, Admins can import all if current_user.role.name not in ["admin", "creator"] and stream.created_by != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) # Creators can only import their own streams if current_user.role.name == "creator" and stream.created_by != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) try: # Parse chat.json chat_json_content = await chat_json_file.read() chat_json_data = json.loads(chat_json_content.decode('utf-8')) chat_json_service = ChatJsonService(db) parsed_data = chat_json_service.parse_chat_json(chat_json_data) # Update stream_metadata with chapters metadata = { "game": parsed_data.get("game"), "streamer_login": parsed_data.get("streamer_login"), "streamer_id": parsed_data.get("streamer_id"), "chapters": parsed_data.get("chapters") } stream.stream_metadata = metadata db.commit() # Import chapters to dedicated table if parsed_data.get("chapters"): # Delete existing chapters for this stream from ..models.chapter import Chapter db.query(Chapter).filter(Chapter.stream_id == stream.id).delete() db.commit() # Import new chapters chat_json_service.import_chapters_from_chat_json( stream_id=stream.id, chapters_data=parsed_data["chapters"] ) # Import chat messages if parsed_data.get("comments"): # Delete existing chat messages for this stream from ..models.chat_message import ChatMessage db.query(ChatMessage).filter(ChatMessage.stream_id == stream.id).delete() db.commit() # Import new chat messages chat_json_service.import_chat_messages_from_chat_json( stream_id=stream.id, comments_data=parsed_data["comments"], embedded_data=parsed_data.get("embedded_data", {}) ) db.refresh(stream) # Renumber after manual import try: from ..services.stream_numbering_service import renumber_all_streams renumber_all_streams() except Exception as e: sys.stderr.write(f"[NUMBERING ERROR] Failed to renumber after manual import: {e}\n") return stream except json.JSONDecodeError: raise HTTPException(status_code=400, detail="Invalid chat.json file") except Exception as e: sys.stderr.write(f"[IMPORT CHAT JSON ERROR] Error importing chat.json: {e}\n") sys.stderr.flush() raise HTTPException(status_code=500, detail=f"Error importing chat.json: {str(e)}") @router.get("/{stream_id}/download-progress") async def get_download_progress( stream_id: UUID, db: Session = Depends(get_db) ): """Get download progress for a stream""" try: progress_key = f"download_progress:{stream_id}" progress_data = redis_client.get(progress_key) if not progress_data: return { 'status': 'not_found', 'percentage': 0 } return json.loads(progress_data) except Exception as e: return { 'status': 'error', 'percentage': 0, 'error': str(e) } @router.get("/{stream_id}/processing-progress") async def get_processing_progress( stream_id: UUID, db: Session = Depends(get_db) ): """Get processing progress for a stream""" try: progress_key = f"stream_progress:{stream_id}" progress_data = redis_client.get(progress_key) if not progress_data: return { 'status': 'not_found', 'percentage': 0 } return json.loads(progress_data) except Exception as e: return { 'status': 'error', 'percentage': 0, 'error': str(e) } @router.post("/{stream_id}/process") async def trigger_processing( stream_id: UUID, force: bool = False, db: Session = Depends(get_db), current_user = Depends(get_current_user) ): """Manually trigger HLS processing for a stream""" try: sys.stderr.write(f"[PROCESSING] Triggering processing for stream {stream_id}, force={force}\n") sys.stderr.flush() # Get stream stream = db.query(Stream).filter(Stream.id == str(stream_id)).first() if not stream: raise HTTPException(status_code=404, detail="Stream not found") sys.stderr.write(f"[PROCESSING] Stream found: {stream.title}, status: {stream.status}, file_path: {stream.file_path}\n") sys.stderr.flush() if not stream.file_path: raise HTTPException(status_code=400, detail="No video file available for processing") if stream.status == 'processing' and not force: sys.stderr.write(f"[PROCESSING] Stream is already being processed\n") sys.stderr.flush() return {'message': 'Stream is already being processed', 'use_force': 'Set force=true to restart'} # Check if processing was interrupted (backend restart) state_key = f"processing_state:{stream_id}" state = redis_client.get(state_key) if state and json.loads(state).get('status') == 'interrupted': sys.stderr.write(f"[PROCESSING] Processing was interrupted, auto-restarting\n") sys.stderr.flush() # Clear interrupted state redis_client.delete(state_key) if stream.status == 'ready': sys.stderr.write(f"[PROCESSING] Stream is already processed\n") sys.stderr.flush() return {'message': 'Stream is already processed', 'use_force': 'Set force=true to reprocess'} # Update status to processing stream.status = 'processing' db.commit() sys.stderr.write(f"[PROCESSING] Status updated to processing\n") sys.stderr.flush() # Trigger HLS processing in background from ..services.video_processor import VideoProcessor import threading def process_in_background(): try: sys.stderr.write(f"[PROCESSING] Background thread started for stream {stream_id}\n") sys.stderr.flush() processor = VideoProcessor() processor.process_video(str(stream_id)) sys.stderr.write(f"[PROCESSING] Background thread completed for stream {stream_id}\n") sys.stderr.flush() except Exception as e: sys.stderr.write(f"[PROCESSING ERROR] Background processing failed: {e}\n") sys.stderr.flush() import traceback traceback.print_exc() thread = threading.Thread(target=process_in_background) thread.start() sys.stderr.write(f"[PROCESSING] Thread started, returning response\n") sys.stderr.flush() return {'message': 'Processing started', 'stream_id': str(stream_id)} except HTTPException: raise except Exception as e: sys.stderr.write(f"[PROCESSING ERROR] Error: {e}\n") sys.stderr.flush() import traceback traceback.print_exc() raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to trigger processing: {str(e)}" ) @router.get("/{stream_id}/thumbnail/{segment_number}") async def get_segment_thumbnail( stream_id: str, segment_number: int, db: Session = Depends(get_db) ): """Get thumbnail (first frame) of an HLS segment""" try: # Validate segment number if segment_number < 0: raise HTTPException(status_code=400, detail="Segment number must be non-negative") # Check if stream exists stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: raise HTTPException(status_code=404, detail="Stream not found") # Get HLS directory (segments are in uploads/videos/{stream_id}/) hls_dir = f"{settings.UPLOAD_PATH}/videos/{stream_id}" # Try standard HLS format first: {stream_id}-N.ts segment_file = f"{hls_dir}/{stream_id}-{segment_number}.ts" # Fallback to Twitch format: {stream_id}.ts-FragN if not os.path.exists(segment_file): segment_file = f"{hls_dir}/{stream_id}.ts-Frag{segment_number + 1}" # Twitch starts at 1 # Check if segment exists if not os.path.exists(segment_file): raise HTTPException(status_code=404, detail="Segment not found") # Extract first frame using ffmpeg thumbnail_path = f"{hls_dir}/thumb_{segment_number:03d}.jpg" # Generate thumbnail if not cached if not os.path.exists(thumbnail_path): cmd = [ 'ffmpeg', '-i', segment_file, '-frames:v', '1', '-q:v', '2', '-strict', '-2', '-y', thumbnail_path ] subprocess.run(cmd, capture_output=True, check=True) # Read and return thumbnail with open(thumbnail_path, 'rb') as f: thumbnail_data = f.read() return Response(content=thumbnail_data, media_type="image/jpeg") except HTTPException: raise except Exception as e: sys.stderr.write(f"[THUMBNAIL ERROR] Error generating thumbnail: {e}\n") sys.stderr.flush() raise HTTPException(status_code=500, detail=f"Failed to generate thumbnail: {str(e)}") @router.get("/{stream_id}/thumbnail-progress") async def get_thumbnail_progress( stream_id: str, db: Session = Depends(get_db) ): """Get thumbnail generation progress for a stream""" try: stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: return None # Check if there's any recent progress if stream.thumbnail_progress_updated_at and stream.thumbnail_progress_total > 0: return { 'generated': stream.thumbnail_progress_generated, 'total': stream.thumbnail_progress_total, 'message': stream.thumbnail_progress_message, 'error': stream.thumbnail_progress_error, 'updated_at': stream.thumbnail_progress_updated_at.isoformat() if stream.thumbnail_progress_updated_at else None } else: return None except Exception as e: sys.stderr.write(f"[THUMBNAIL ERROR] Error getting thumbnail progress: {e}\n") sys.stderr.flush() raise HTTPException(status_code=500, detail=f"Failed to get thumbnail progress: {str(e)}") @router.get("/{stream_id}/hls-files") async def get_hls_files( stream_id: UUID, db: Session = Depends(get_db), current_user: User = Depends(get_current_admin_user) ): """Get available HLS files for a stream (admin only)""" try: stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: raise HTTPException(status_code=404, detail="Stream not found") hls_dir = f"{settings.UPLOAD_PATH}/videos/{stream_id}" # Check which files exist available_files = { 'm3u8': None, 'mp4': None, 'ts_segments': 0, 'directory': hls_dir } # Check for .m3u8 playlist m3u8_path = os.path.join(hls_dir, f"{stream_id}.m3u8") if os.path.exists(m3u8_path): available_files['m3u8'] = f"/uploads/videos/{stream_id}/{stream_id}.m3u8" # Check for .mp4 file mp4_path = os.path.join(hls_dir, f"{stream_id}.mp4") if os.path.exists(mp4_path): available_files['mp4'] = f"/uploads/videos/{stream_id}/{stream_id}.mp4" # Count .ts segments if os.path.exists(hls_dir) and os.path.isdir(hls_dir): ts_files = [f for f in os.listdir(hls_dir) if f.endswith('.ts')] available_files['ts_segments'] = len(ts_files) return available_files except Exception as e: sys.stderr.write(f"[HLS ERROR] Error getting HLS files: {e}\n") sys.stderr.flush() raise HTTPException(status_code=500, detail=f"Failed to get HLS files: {str(e)}") async def generate_thumbnails_background(stream_id: str, hls_dir: str): """Background function to generate thumbnails and sprite sheet""" sys.stderr.write(f"[THUMBNAILS] Starting thumbnail generation for stream {stream_id}\n") sys.stderr.write(f"[THUMBNAILS] HLS path: {hls_dir}\n") sys.stderr.flush() import json progress_key = f"stream_progress:{stream_id}" db = next(get_db()) try: # Check if hls_dir is a .m3u8 file or a directory if os.path.isfile(hls_dir) and hls_dir.endswith('.m3u8'): # Parse m3u8 playlist to get segment files sys.stderr.write(f"[THUMBNAILS] Parsing m3u8 playlist\n") sys.stderr.flush() # Get directory from m3u8 file path m3u8_dir = os.path.dirname(hls_dir) # Read m3u8 playlist with open(hls_dir, 'r') as f: m3u8_content = f.read() # Extract segment file names from m3u8 segment_files = [] for line in m3u8_content.split('\n'): line = line.strip() if line and not line.startswith('#') and line.endswith('.ts'): segment_files.append(os.path.basename(line)) # Use the directory for file operations hls_dir = m3u8_dir sys.stderr.write(f"[THUMBNAILS] Found {len(segment_files)} segments in m3u8 playlist\n") sys.stderr.flush() else: # Find all segment files supporting multiple naming patterns: # 1. Twitch format: {stream_id}.ts-Frag1, {stream_id}.ts-Frag2, etc. # 2. Standard HLS: segment_000.ts, segment_001.ts, etc. # 3. Manual playlists: segment_01127.ts, etc. # 4. Any other .ts files in the directory all_ts_files = [f for f in os.listdir(hls_dir) if f.endswith('.ts') or '.ts-Frag' in f] sys.stderr.write(f"[THUMBNAILS] HLS directory: {hls_dir}\n") sys.stderr.write(f"[THUMBNAILS] Stream ID: {stream_id}\n") sys.stderr.write(f"[THUMBNAILS] All TS files: {len(all_ts_files)}\n") sys.stderr.flush() # Try to detect the naming pattern and filter accordingly standard_hls = [f for f in all_ts_files if f.startswith(f'{stream_id}-') and f.endswith('.ts')] twitch_segments = [f for f in all_ts_files if f.startswith(f'{stream_id}.ts-Frag')] segment_segments = [f for f in all_ts_files if f.startswith('segment_')] sys.stderr.write(f"[THUMBNAILS] Standard HLS: {len(standard_hls)}\n") sys.stderr.write(f"[THUMBNAILS] Twitch segments: {len(twitch_segments)}\n") sys.stderr.write(f"[THUMBNAILS] Segment segments: {len(segment_segments)}\n") sys.stderr.flush() # Use standard HLS format if available, otherwise Twitch format, otherwise segment_ format, otherwise all .ts files if standard_hls: segment_files = standard_hls sys.stderr.write(f"[THUMBNAILS] Using standard HLS format segments\n") elif twitch_segments: segment_files = twitch_segments sys.stderr.write(f"[THUMBNAILS] Using Twitch format segments\n") elif segment_segments: segment_files = segment_segments sys.stderr.write(f"[THUMBNAILS] Using standard segment_ format\n") else: segment_files = all_ts_files sys.stderr.write(f"[THUMBNAILS] Using all .ts files\n") sys.stderr.flush() sys.stderr.write(f"[THUMBNAILS] Found {len(segment_files)} segment files\n") sys.stderr.flush() if not segment_files: stream = db.query(Stream).filter(Stream.id == stream_id).first() if stream: stream.thumbnail_progress_error = 'No segments found' stream.thumbnail_progress_updated_at = datetime.now(timezone.utc) db.commit() return # Sort segment files numerically based on their naming pattern def extract_segment_number(filename): import re # Try standard HLS format: {stream_id}-123.ts standard_match = re.search(r'-(\d+)\.ts$', filename) if standard_match: return int(standard_match.group(1)) # Try Twitch format: {stream_id}.ts-Frag123 twitch_match = re.search(r'-Frag(\d+)', filename) if twitch_match: return int(twitch_match.group(1)) # Try segment_ format: segment_000.ts or segment_01127.ts segment_match = re.search(r'segment[_-](\d+)', filename) if segment_match: return int(segment_match.group(1)) # Try any numeric pattern in filename numeric_match = re.search(r'(\d+)', filename) if numeric_match: return int(numeric_match.group(1)) # Fallback: use filename as string sort return filename segment_files.sort(key=extract_segment_number) # Apply sampling rate to reduce number of thumbnails sampling_rate = settings.THUMBNAIL_SAMPLING_RATE if sampling_rate > 1: segment_files = segment_files[::sampling_rate] # Initialize progress in database stream = db.query(Stream).filter(Stream.id == stream_id).first() if stream: stream.thumbnail_progress_total = len(segment_files) stream.thumbnail_progress_generated = 0 stream.thumbnail_progress_message = 'Starting thumbnail generation' stream.thumbnail_progress_error = None stream.thumbnail_progress_updated_at = datetime.now(timezone.utc) db.commit() generated_count = 0 semaphore = asyncio.Semaphore(8) # Limit to 8 concurrent FFmpeg processes async def generate_thumbnail(i: int, segment_file: str): nonlocal generated_count thumbnail_path = f"{hls_dir}/thumb_{i+1:03d}.jpg" # Skip if thumbnail already exists if os.path.exists(thumbnail_path): generated_count += 1 # Update progress in database stream = db.query(Stream).filter(Stream.id == stream_id).first() if stream: stream.thumbnail_progress_generated = generated_count stream.thumbnail_progress_message = f'Processing segment {i + 1} of {len(segment_files)} (skipped, already exists)' stream.thumbnail_progress_updated_at = datetime.now(timezone.utc) db.commit() return async with semaphore: # Generate thumbnail cmd = [ 'ffmpeg', '-i', f"{hls_dir}/{segment_file}", '-frames:v', '1', '-q:v', '2', '-strict', '-2', '-threads', '1', # Use single thread per FFmpeg process '-y', thumbnail_path ] process = await asyncio.create_subprocess_exec(*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) stdout, stderr = await process.communicate() if process.returncode != 0: sys.stderr.write(f"[THUMBNAILS ERROR] FFmpeg failed for segment {i+1}: {stderr.decode()}\n") sys.stderr.flush() raise Exception(f"FFmpeg thumbnail generation failed: {stderr.decode()}") generated_count += 1 # Update progress via Redis if len(segment_files) > 0: thumbnail_progress = 60 + (generated_count / len(segment_files) * 10) # 60-70% eta_seconds = int((len(segment_files) - generated_count) * 0.5) # Estimate 0.5s per thumbnail progress_data = { 'status': 'processing', 'percentage': thumbnail_progress, 'step': 'generating_thumbnails', 'step_name': 'Generating thumbnails', 'details': { 'current_thumbnail': generated_count, 'total_thumbnails': len(segment_files), 'eta': f"{eta_seconds}s" } } redis_client.setex(progress_key, 3600, json.dumps(progress_data)) # Only log at 10% milestones if generated_count == 1 or generated_count == len(segment_files) or (generated_count % max(1, len(segment_files) // 10) == 0): sys.stderr.write(f"[THUMBNAILS] Generated thumbnail {generated_count}/{len(segment_files)}\n") sys.stderr.flush() # Update progress in database stream = db.query(Stream).filter(Stream.id == stream_id).first() if stream: stream.thumbnail_progress_generated = generated_count stream.thumbnail_progress_message = f'Processing segment {i + 1} of {len(segment_files)}' stream.thumbnail_progress_updated_at = datetime.now(timezone.utc) db.commit() # Generate thumbnails in parallel with limited concurrency tasks = [generate_thumbnail(i, segment_file) for i, segment_file in enumerate(segment_files)] try: await asyncio.gather(*tasks) sys.stderr.write(f"[THUMBNAILS] Completed thumbnail generation: {generated_count}/{len(segment_files)}\n") sys.stderr.flush() except Exception as e: sys.stderr.write(f"[THUMBNAILS ERROR] Failed to generate thumbnails: {e}\n") sys.stderr.flush() import traceback traceback.print_exc() raise # Generate sprite sheet(s) - split into multiple if too many thumbnails MAX_THUMBNAILS_PER_SPRITE = 2000 existing_thumbs = [f for f in os.listdir(hls_dir) if f.startswith('thumb_') and f.endswith('.jpg')] def extract_thumb_number(filename): match = re.search(r'thumb_(\d+)\.jpg', filename) return int(match.group(1)) if match else 0 existing_thumbs.sort(key=extract_thumb_number) total_thumbs = len(existing_thumbs) sprite_count = (total_thumbs + MAX_THUMBNAILS_PER_SPRITE - 1) // MAX_THUMBNAILS_PER_SPRITE sprite_meta = { 'sprite_count': sprite_count, 'total_thumbnails': total_thumbs, 'thumbnails_per_sprite': MAX_THUMBNAILS_PER_SPRITE, 'columns': 50, 'thumbnail_width': 160, 'thumbnail_height': 90 } for sprite_idx in range(sprite_count): start_idx = sprite_idx * MAX_THUMBNAILS_PER_SPRITE end_idx = min(start_idx + MAX_THUMBNAILS_PER_SPRITE, total_thumbs) chunk_thumbs = existing_thumbs[start_idx:end_idx] # Create a temporary directory for this chunk chunk_dir = f"{hls_dir}/sprite_chunk_{sprite_idx}" os.makedirs(chunk_dir, exist_ok=True) # Create hardlinks in sequential order for ffmpeg for i, thumb_file in enumerate(chunk_thumbs, 1): src = f"{hls_dir}/{thumb_file}" dst = f"{chunk_dir}/thumb_{i:03d}.jpg" if os.path.exists(dst): os.remove(dst) os.link(src, dst) # Calculate grid dimensions max_columns = 50 columns = min(max_columns, len(chunk_thumbs)) rows = (len(chunk_thumbs) + columns - 1) // columns # Determine sprite path: always use indexed naming for consistency sprite_path = f"{hls_dir}/thumbnails_sprite_{sprite_idx + 1}.jpg" cmd = [ 'ffmpeg', '-i', f"{chunk_dir}/thumb_%03d.jpg", '-vf', f'scale=160:90,tile={columns}x{rows}', '-q:v', '2', '-y', sprite_path ] process = await asyncio.create_subprocess_exec(*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) stdout, stderr = await process.communicate() # Cleanup temporary chunk directory shutil.rmtree(chunk_dir, ignore_errors=True) if process.returncode != 0: sys.stderr.write(f"[THUMBNAILS ERROR] Failed to generate sprite sheet {sprite_idx + 1}: {stderr.decode()}\n") sys.stderr.flush() raise RuntimeError(f"Sprite sheet {sprite_idx + 1} generation failed: {stderr.decode()}") sys.stderr.write(f"[THUMBNAILS] Generated sprite sheet {sprite_idx + 1}/{sprite_count}: {sprite_path}\n") sys.stderr.flush() # Write metadata JSON (always, even for single sprite, for consistency) meta_path = f"{hls_dir}/thumbnails_sprite.json" with open(meta_path, 'w') as f: json.dump(sprite_meta, f) # Clear progress in database stream = db.query(Stream).filter(Stream.id == stream_id).first() if stream: stream.thumbnail_progress_generated = len(segment_files) stream.thumbnail_progress_total = len(segment_files) stream.thumbnail_progress_message = 'Thumbnail generation completed' stream.thumbnail_progress_error = None stream.thumbnail_progress_updated_at = datetime.now(timezone.utc) db.commit() except Exception as e: sys.stderr.write(f"[THUMBNAILS ERROR] Error generating thumbnails in background: {e}\n") sys.stderr.flush() # Update progress on error in database stream = db.query(Stream).filter(Stream.id == stream_id).first() if stream: stream.thumbnail_progress_error = str(e) stream.thumbnail_progress_message = f'Error: {str(e)}' stream.thumbnail_progress_updated_at = datetime.now(timezone.utc) db.commit() finally: db.close() @router.post("/{stream_id}/generate-thumbnails") async def generate_thumbnails( stream_id: UUID, request: Request, force: bool = False, db: Session = Depends(get_db), current_user: User = Depends(get_current_creator_user) ): """Generate all thumbnails for a stream (admin only) - runs in background""" try: # Check if stream exists stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: raise HTTPException(status_code=404, detail="Stream not found") # Parse request body for custom HLS path try: body = await request.json() custom_hls_path = body.get('hls_path') except: custom_hls_path = None # Get HLS directory if custom_hls_path: # Use custom HLS path from frontend selection # Remove /uploads prefix if present to avoid double /uploads if custom_hls_path.startswith('/uploads'): custom_hls_path = custom_hls_path[len('/uploads'):] hls_dir = f"{settings.UPLOAD_PATH}{custom_hls_path}" sys.stderr.write(f"[THUMBNAILS] Using custom HLS path: {hls_dir}\n") sys.stderr.flush() else: # Use HLS path from stream if available if stream.hls_path: hls_path = stream.hls_path # Remove /uploads prefix if present to avoid double /uploads if hls_path.startswith('/uploads'): hls_path = hls_path[len('/uploads'):] hls_dir = f"{settings.UPLOAD_PATH}{hls_path}" sys.stderr.write(f"[THUMBNAILS] Using HLS path from stream: {hls_dir}\n") sys.stderr.flush() else: # Default: use stream directory hls_dir = f"{settings.UPLOAD_PATH}/videos/{stream_id}" # Check if m3u8 playlist exists and use it, otherwise use directory with .ts segments m3u8_path = os.path.join(hls_dir, f"{stream_id}.m3u8") if os.path.exists(m3u8_path): hls_dir = m3u8_path sys.stderr.write(f"[THUMBNAILS] Using m3u8 playlist: {hls_dir}\n") sys.stderr.flush() else: sys.stderr.write(f"[THUMBNAILS] No m3u8 playlist found, using directory with .ts segments\n") sys.stderr.flush() # Check if already generating (check database for recent progress) if not force and stream.thumbnail_progress_updated_at and stream.thumbnail_progress_total > 0: # If progress was updated in the last 5 minutes, consider it still running if (datetime.now(timezone.utc) - stream.thumbnail_progress_updated_at).total_seconds() < 300: return { 'message': 'Thumbnail generation already in progress. Use force=true to restart.', 'stream_id': stream_id } # Reset progress if force=true if force: stream.thumbnail_progress_generated = 0 stream.thumbnail_progress_total = 0 stream.thumbnail_progress_error = None stream.thumbnail_progress_message = 'Forced restart' stream.thumbnail_progress_updated_at = None db.commit() sys.stderr.write(f"[THUMBNAILS] Force restart for stream {stream_id} - progress reset\n") sys.stderr.flush() # Start background thread with async function def run_async(): asyncio.run(generate_thumbnails_background(stream_id, hls_dir)) thread = threading.Thread(target=run_async) thread.daemon = True thread.start() return { 'message': 'Thumbnail generation started in background', 'stream_id': stream_id } except HTTPException: raise except Exception as e: sys.stderr.write(f"[THUMBNAILS ERROR] Error generating thumbnails: {e}\n") sys.stderr.flush() raise HTTPException(status_code=500, detail=f"Failed to generate thumbnails: {str(e)}") @router.post("/{stream_id}/thumbnail") async def generate_single_thumbnail( stream_id: UUID, request: Request, db: Session = Depends(get_db), current_user: User = Depends(get_current_creator_user) ): """Generate a single thumbnail from a specific timestamp""" try: # Parse request body body = await request.json() timestamp = body.get('timestamp') if timestamp is None: raise HTTPException(status_code=400, detail="Timestamp is required") # Check if stream exists stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: raise HTTPException(status_code=404, detail="Stream not found") if not stream.hls_path: raise HTTPException(status_code=400, detail="Stream has no HLS path") # Import thumbnail service from ..services.thumbnail_service import ThumbnailService thumbnail_service = ThumbnailService() # Convert HLS path to absolute path for FFmpeg # HLS path is already relative (e.g., /uploads/videos/...), so just prepend /app hls_path = stream.hls_path if hls_path.startswith('/'): hls_path = f"/app{hls_path}" # Generate thumbnail from timestamp output_path = f"{settings.UPLOAD_PATH}/thumbnails/{stream_id}.jpg" thumbnail_path = thumbnail_service.generate_thumbnail_from_timestamp( video_path=hls_path, timestamp_seconds=float(timestamp), output_path=output_path ) # Update thumbnail URL in database with cache-busting parameter timestamp_param = int(datetime.now(timezone.utc).timestamp()) stream.thumbnail_url = f"/uploads/thumbnails/{stream_id}.jpg?t={timestamp_param}" stream.thumbnail_updated_at = datetime.now(timezone.utc) db.commit() # Invalidate cache try: redis_client.delete(f"stream:{stream_id}") except: pass return {"thumbnail_url": stream.thumbnail_url} except HTTPException: raise except Exception as e: sys.stderr.write(f"[THUMBNAIL ERROR] Error generating thumbnail from timestamp: {e}\n") sys.stderr.flush() raise HTTPException(status_code=500, detail=f"Failed to generate thumbnail: {str(e)}") @router.post("/{stream_id}/thumbnail-upload") async def upload_stream_thumbnail( stream_id: UUID, file: UploadFile = File(...), db: Session = Depends(get_db), current_user: User = Depends(get_current_creator_user) ): """Upload a client-side generated thumbnail""" stream = db.query(Stream).filter(Stream.id == stream_id).first() if not stream: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Stream not found" ) # Check ownership if current_user.role.name == "creator" and stream.created_by != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied" ) try: # Ensure thumbnails directory exists os.makedirs(f"{settings.UPLOAD_PATH}/thumbnails", exist_ok=True) output_path = f"{settings.UPLOAD_PATH}/thumbnails/{stream_id}.jpg" # Save uploaded file with open(output_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) # Update thumbnail URL in database with cache-busting parameter timestamp_param = int(datetime.now(timezone.utc).timestamp()) stream.thumbnail_url = f"/uploads/thumbnails/{stream_id}.jpg?t={timestamp_param}" stream.thumbnail_updated_at = datetime.now(timezone.utc) db.commit() # Invalidate cache try: redis_client.delete(f"stream:{stream_id}") except: pass return {"thumbnail_url": stream.thumbnail_url} except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to upload thumbnail: {str(e)}" ) @router.post("/manual", response_model=StreamResponse) async def create_manual_stream( title: Optional[str] = Form(None), description: Optional[str] = Form(None), hls_path: Optional[str] = Form(None), thumbnail_timestamp: Optional[float] = Form(None), is_public: bool = Form(False), chat_json_file: Optional[UploadFile] = File(None), db: Session = Depends(get_db), current_user: User = Depends(require_permission("stream:create")) ): """Create a stream manually without upload, optionally with chat.json data""" try: # Auto-create HLS directory if no path provided resolved_path = hls_path if not hls_path: # Create HLS directory - will use stream ID after stream is created hls_dir = None else: # If HLS path is provided, only allow for admins if current_user.role.name != 'admin': raise HTTPException(status_code=403, detail="Only admins can specify custom HLS path") # Resolve HLS path - handle relative paths and check both storage locations if not os.path.isabs(hls_path): # Try relative paths in storage locations possible_paths = [ os.path.join('/app/storage', hls_path.lstrip('/')), os.path.join('/app/uploads', hls_path.lstrip('/')), hls_path # try as-is ] for path in possible_paths: if os.path.exists(path): resolved_path = path break else: raise HTTPException(status_code=400, detail=f"HLS path does not exist: {hls_path} (tried: {possible_paths})") elif not os.path.exists(hls_path): raise HTTPException(status_code=400, detail=f"HLS path does not exist: {hls_path}") # Parse chat.json if provided parsed_data = None if chat_json_file: try: chat_json_content = await chat_json_file.read() chat_json_data = json.loads(chat_json_content.decode('utf-8')) chat_json_service = ChatJsonService(db) parsed_data = chat_json_service.parse_chat_json(chat_json_data) # Override stream data with chat.json if not provided if not title and parsed_data.get("title"): title = parsed_data["title"] if not description and parsed_data.get("description"): description = parsed_data["description"] except json.JSONDecodeError: raise HTTPException(status_code=400, detail="Invalid chat.json file") except Exception as e: sys.stderr.write(f"[MANUAL IMPORT ERROR] Error parsing chat.json: {e}\n") sys.stderr.flush() raise HTTPException(status_code=400, detail=f"Error parsing chat.json: {str(e)}") # Use default title if still not provided if not title: title = "Manual Stream" # Create stream (without HLS path if not provided) chat_json_service = ChatJsonService(db) stream = chat_json_service.create_stream_from_chat_json( user_id=current_user.id, hls_path=None if not hls_path else resolved_path, parsed_data=parsed_data or {}, thumbnail_timestamp=thumbnail_timestamp, is_public=is_public, is_twitch_import=False ) # Create HLS directory after stream is created (if no HLS path was provided) if not hls_path: hls_dir = f"/app/uploads/videos/{stream.id}" os.makedirs(hls_dir, exist_ok=True) # Update stream with relative HLS path (without /app prefix) stream.hls_path = f"/uploads/videos/{stream.id}" stream.status = "ready" # Update status to ready since HLS path is now set db.commit() db.refresh(stream) # Regenerate master ffconcat try: from ..services.ffconcat_service import regenerate_master_ffconcat regenerate_master_ffconcat() except Exception as e: sys.stderr.write(f"[FFCONCAT ERROR] Failed to regenerate after manual import: {e}\n") # Download thumbnail if video_id is available, otherwise use default thumbnail thumbnail_downloaded = False if parsed_data and parsed_data.get("video_id"): try: video_id = parsed_data.get("video_id") twitch_thumbnail_url = f"https://static-cdn.jtvnw.net/cf_vods/d2nvs3xx597ldy/{video_id}/thumb/thumb-640-360.jpg" response = requests.get(twitch_thumbnail_url, timeout=10) if response.status_code == 200: thumbnail_path = os.path.join(hls_dir, "thumbnail.jpg") with open(thumbnail_path, 'wb') as f: f.write(response.content) # Update thumbnail URL to local HTTPS path stream.thumbnail_url = f"https://{settings.DOMAIN}/uploads/videos/{stream.id}/thumbnail.jpg" if settings.DOMAIN else f"/uploads/videos/{stream.id}/thumbnail.jpg" thumbnail_downloaded = True db.commit() db.refresh(stream) except Exception as e: sys.stderr.write(f"[MANUAL IMPORT ERROR] Error downloading thumbnail: {e}\n") # If thumbnail download failed, use default thumbnail if not thumbnail_downloaded: try: default_thumbnail_path = "/app/app/static/default_thumbnail.svg" thumbnail_dest = os.path.join(hls_dir, "thumbnail.svg") if os.path.exists(default_thumbnail_path): shutil.copy(default_thumbnail_path, thumbnail_dest) stream.thumbnail_url = f"https://{settings.DOMAIN}/uploads/videos/{stream.id}/thumbnail.svg" if settings.DOMAIN else f"/uploads/videos/{stream.id}/thumbnail.svg" db.commit() db.refresh(stream) except Exception as e: sys.stderr.write(f"[MANUAL IMPORT ERROR] Error copying default thumbnail: {e}\n") # Import chapters and chat messages if chat.json was provided if parsed_data: try: # Update stream_metadata with chapters (like Twitch import) metadata = { "game": parsed_data.get("game"), "streamer_login": parsed_data.get("streamer_login"), "streamer_id": parsed_data.get("streamer_id"), "chapters": parsed_data.get("chapters") } stream.stream_metadata = metadata db.commit() # Import chapters to dedicated table if parsed_data.get("chapters"): chat_json_service.import_chapters_from_chat_json( stream_id=stream.id, chapters_data=parsed_data["chapters"] ) # Import chat messages if parsed_data.get("comments"): chat_json_service.import_chat_messages_from_chat_json( stream_id=stream.id, comments_data=parsed_data["comments"], embedded_data=parsed_data.get("embedded_data", {}) ) except Exception as e: sys.stderr.write(f"[MANUAL IMPORT ERROR] Error importing chat data: {e}\n") sys.stderr.flush() # Don't fail the stream creation if chat import fails # Generate thumbnail if timestamp provided if thumbnail_timestamp and stream.hls_path: try: from ..services.thumbnail_service import ThumbnailService thumbnail_service = ThumbnailService() # Generate thumbnail from timestamp thumbnail_path = thumbnail_service.generate_thumbnail_from_timestamp( video_path=stream.hls_path, timestamp_seconds=thumbnail_timestamp, output_path=f"{settings.UPLOAD_PATH}/thumbnails/{stream.id}.jpg" ) # Update stream with thumbnail URL (HTTPS) stream.thumbnail_url = f"https://{settings.DOMAIN}/uploads/thumbnails/{stream.id}.jpg" if settings.DOMAIN else f"/uploads/thumbnails/{stream.id}.jpg" db.commit() # Invalidate cache for this stream to ensure frontend gets updated data redis_client.delete(f"stream:{stream.id}") redis_client.delete("streams:latest") redis_client.delete("streams:popular") redis_client.delete("streams:random") except Exception as e: sys.stderr.write(f"[MANUAL IMPORT ERROR] Error generating thumbnail: {e}\n") sys.stderr.flush() # Don't fail the stream creation if thumbnail generation fails # Add hls_url to response stream.hls_url = _build_hls_url(stream.hls_path) stream.hls_available = _is_hls_available(stream) # Convert thumbnail_url to absolute HTTPS URL if stream.thumbnail_url and not stream.thumbnail_url.startswith('http'): stream.thumbnail_url = f"https://{settings.DOMAIN}{stream.thumbnail_url}" if settings.DOMAIN else stream.thumbnail_url return stream except HTTPException: raise except Exception as e: sys.stderr.write(f"[MANUAL IMPORT ERROR] Error creating manual stream: {e}\n") sys.stderr.flush() raise HTTPException(status_code=500, detail=f"Failed to create manual stream: {str(e)}")