telegram.video.summary.bot/tests/test_real_integration.py

import pytest
import os
import sys
import asyncio
import logging
from dotenv import load_dotenv

# Dodanie katalogu nadrzędnego do ścieżki dla importów
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

from src.youtube_utils import (
    extract_youtube_urls, 
    extract_video_id, 
    get_transcript, 
    NoTranscriptFound, 
    TranscriptsDisabled,
    APITokenMissing,
    AuthorizationError,
    APIConnectionError,
    APIResponseError,
    NoTranscriptLanguagesAvailable,
    YouTubeUtilsError
)
from src.openai_utils import summarize_text, check_openai_api_status, QuotaExceededError

# Konfiguracja logowania dla testów
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Ładowanie zmiennych środowiskowych z pliku .env (jeśli istnieje)
load_dotenv()

# Zmiana na bardziej popularne wideo, które powinno być bardziej stabilne
# "What makes a good life? Lessons from the longest study on happiness | Robert Waldinger"
# TED Talk - bardzo popularne wystąpienie z napisami w wielu językach (>42M wyświetleń)
TEST_VIDEO_URL = "https://www.youtube.com/watch?v=8KkKuTCFvzI"
TEST_VIDEO_ID = extract_video_id(TEST_VIDEO_URL)
TEST_LANGUAGES = ["pl", "en"]  # Preferujemy polską transkrypcję, potem angielską

# Drugi film zapasowy (na wypadek problemów z pierwszym)
BACKUP_VIDEO_URL = "https://www.youtube.com/watch?v=jNQXAC9IVRw"  # Me at the zoo - pierwszy film na YouTube
BACKUP_VIDEO_ID = extract_video_id(BACKUP_VIDEO_URL)

# Flaga określająca dostępność API OpenAI, inicjalizowana jako None (sprawdzana przed testami)
OPENAI_API_AVAILABLE = None
OPENAI_STATUS_INFO = {}

# Oznaczamy testy jako "slow", żeby można było je pominąć za pomocą --skip-slow
# lub --skip-integration podczas uruchamiania testów
pytestmark = [pytest.mark.integration, pytest.mark.slow]

# Helper do sprawdzania dostępności API OpenAI przed testami
async def ensure_openai_api_status():
    """Sprawdza status API OpenAI i ustawia globalną flagę OPENAI_API_AVAILABLE"""
    global OPENAI_API_AVAILABLE, OPENAI_STATUS_INFO
    
    # Jeśli nie ma klucza API, nie ma sensu sprawdzać
    if not os.environ.get("OPENAI_API_KEY"):
        logger.warning("Brak klucza API OpenAI, pomijam sprawdzanie statusu API")
        OPENAI_API_AVAILABLE = False
        OPENAI_STATUS_INFO = {"error": "Brak klucza API"}
        return False
    
    # Jeśli już sprawdziliśmy, używamy zapisanego wyniku
    if OPENAI_API_AVAILABLE is not None:
        return OPENAI_API_AVAILABLE
    
    try:
        logger.info("Sprawdzam dostępność API OpenAI...")
        OPENAI_API_AVAILABLE, OPENAI_STATUS_INFO = await check_openai_api_status()
        logger.info(f"Status API OpenAI: {'dostępne' if OPENAI_API_AVAILABLE else 'niedostępne'}")
        if not OPENAI_API_AVAILABLE:
            logger.warning(f"API OpenAI niedostępne: {OPENAI_STATUS_INFO.get('error', 'Nieznany błąd')}")
        return OPENAI_API_AVAILABLE
    except QuotaExceededError as e:
        logger.warning(f"Przekroczono limit zapytań API OpenAI: {e}")
        OPENAI_API_AVAILABLE = False
        OPENAI_STATUS_INFO = {"error": str(e)}
        return False
    except Exception as e:
        logger.error(f"Błąd podczas sprawdzania statusu API OpenAI: {e}", exc_info=True)
        OPENAI_API_AVAILABLE = False
        OPENAI_STATUS_INFO = {"error": str(e)}
        return False


@pytest.mark.asyncio
async def test_real_extract_youtube_urls():
    # Test na rzeczywistym tekście z linkami YouTube
    text = """Sprawdź te filmy:
    - https://www.youtube.com/watch?v=8KkKuTCFvzI
    - https://youtu.be/iCvmsMzlF7o
    - Również shorts: https://www.youtube.com/shorts/pFaEGmxQFnM"""
    
    urls = extract_youtube_urls(text)
    assert len(urls) == 3
    assert "https://www.youtube.com/watch?v=8KkKuTCFvzI" in urls
    assert "https://youtu.be/iCvmsMzlF7o" in urls
    assert "https://www.youtube.com/shorts/pFaEGmxQFnM" in urls


@pytest.mark.asyncio
async def test_real_extract_video_id():
    # Test na rzeczywistych URL-ach
    assert extract_video_id("https://www.youtube.com/watch?v=8KkKuTCFvzI") == "8KkKuTCFvzI"
    assert extract_video_id("https://youtu.be/iCvmsMzlF7o") == "iCvmsMzlF7o"
    assert extract_video_id("https://www.youtube.com/shorts/pFaEGmxQFnM") == "pFaEGmxQFnM"


@pytest.mark.asyncio
async def test_real_get_transcript_and_title():
    """Test pobierania rzeczywistej transkrypcji i tytułu"""
    # Ten test wymaga dostępu do internetu i tokenu API youtube-transcript.io
    
    # Sprawdź czy mamy token API
    from src.config import YOUTUBE_TRANSCRIPT_API_TOKEN
    if not YOUTUBE_TRANSCRIPT_API_TOKEN:
        pytest.skip("Brak tokenu API youtube-transcript.io")
    
    # Wypróbuj główne wideo
    logger.info(f"Próbuję pobrać transkrypcję dla głównego wideo: {TEST_VIDEO_ID}")
    
    try:
        # Spróbuj z pierwszym filmem
        transcript, title = await get_transcript(TEST_VIDEO_ID, TEST_LANGUAGES)
        assert len(transcript) > 100  # Upewniamy się, że transkrypcja ma sensowną długość
        assert len(title) > 0  # Tytuł powinien być niepusty
        
        # Wypiszmy tytuł i fragment transkrypcji do weryfikacji
        logger.info(f"Tytuł filmu {TEST_VIDEO_ID}: {title}")
        logger.info(f"Fragment transkrypcji: {transcript[:50]}...")
        
    except (NoTranscriptFound, TranscriptsDisabled) as e:
        logger.warning(f"Nie udało się pobrać transkrypcji dla {TEST_VIDEO_ID}: {e}")
        logger.info(f"Próbuję film zapasowy: {BACKUP_VIDEO_ID}")
        
        # Spróbuj z zapasowym filmem
        try:
            transcript, title = await get_transcript(BACKUP_VIDEO_ID, TEST_LANGUAGES)
            assert len(transcript) > 10  # Film zapasowy "Me at the zoo" ma bardzo krótką transkrypcję
            assert len(title) > 0
            
            # Wypiszmy tytuł i fragment transkrypcji do weryfikacji
            logger.info(f"Tytuł filmu zapasowego {BACKUP_VIDEO_ID}: {title}")
            logger.info(f"Fragment transkrypcji zapasowej: {transcript[:50]}...")
            
        except (NoTranscriptFound, TranscriptsDisabled) as e:
            logger.error(f"Nie można pobrać transkrypcji dla żadnego z testowych filmów: {e}")
            pytest.skip(f"Nie można pobrać transkrypcji dla żadnego z testowych filmów: {e}")
            
    except (APITokenMissing, AuthorizationError) as e:
        logger.error(f"Błąd autoryzacji: {e}")
        pytest.skip(f"Błąd autoryzacji API: {e}")
        
    except (APIConnectionError, APIResponseError) as e:
        logger.error(f"Błąd API: {e}")
        pytest.skip(f"Błąd API: {e}")
        
    except Exception as e:
        logger.error(f"Nieoczekiwany błąd podczas pobierania transkrypcji: {e}", exc_info=True)
        pytest.skip(f"Nieoczekiwany błąd podczas pobierania transkrypcji: {e}")


@pytest.mark.asyncio
async def test_real_summarize_text():
    # Test rzeczywistego streszczania tekstu z OpenAI API
    # Ten test wymaga klucza API OpenAI i dostępnego API
    
    # Sprawdź czy API OpenAI jest dostępne
    if not await ensure_openai_api_status():
        pytest.skip(f"API OpenAI niedostępne: {OPENAI_STATUS_INFO.get('error', 'Nieznany błąd')}")
    
    # Użyjmy krótkiej transkrypcji dla oszczędności tokenów
    short_transcript = """
    Chciałbym porozmawiać o tym, co sprawia, że życie jest dobre i wartościowe.
    Przeprowadziliśmy jedno z najdłuższych badań nad szczęściem - trwające ponad 75 lat.
    Badaliśmy życie tych samych osób od czasu, gdy byli nastolatkami, aż do starości.
    Wniosek? Dobre relacje z innymi ludźmi są kluczem do szczęścia i zdrowia.
    Nie pieniądze, nie sława, nie ciężka praca, ale jakość naszych relacji z bliskimi.
    Osoby, które miały dobre relacje z rodziną, przyjaciółmi i społecznością, 
    były szczęśliwsze, zdrowsze i żyły dłużej.
    """
    
    logger.info("Próbuję streszczenie transkrypcji za pomocą OpenAI API")
    
    try:
        summary = await summarize_text(short_transcript)
        assert summary is not None
        assert len(summary) > 0
        
        logger.info(f"Wygenerowane streszczenie: {summary}")
    except QuotaExceededError as e:
        logger.warning(f"Przekroczono limit zapytań API OpenAI: {e}")
        pytest.skip(f"Przekroczono limit zapytań API OpenAI: {e}")
    except Exception as e:
        logger.error(f"Problem z API OpenAI: {e}", exc_info=True)
        pytest.skip(f"Problem z API OpenAI: {e}")


@pytest.mark.asyncio
async def test_end_to_end_integration():
    """
    Ten test wykonuje pełną integrację od linku YouTube do streszczenia.
    Wymaga dostępu do internetu, tokenu API youtube-transcript.io oraz klucza API OpenAI.
    """
    # Sprawdź czy mamy token API youtube-transcript.io
    from src.config import YOUTUBE_TRANSCRIPT_API_TOKEN
    if not YOUTUBE_TRANSCRIPT_API_TOKEN:
        pytest.skip("Brak tokenu API youtube-transcript.io")
    
    # Sprawdź czy API OpenAI jest dostępne
    if not await ensure_openai_api_status():
        pytest.skip(f"API OpenAI niedostępne: {OPENAI_STATUS_INFO.get('error', 'Nieznany błąd')}")
    
    logger.info("Rozpoczynam test integracyjny end-to-end")
    
    # Testujemy na klasycznym filmie "Me at the zoo", który jest krótki i ma dostępne transkrypcje
    text_with_url = f"Sprawdź ten film: {BACKUP_VIDEO_URL}"
    
    # 1. Wydobycie URL
    urls = extract_youtube_urls(text_with_url)
    assert len(urls) == 1
    url = urls[0]
    logger.info(f"Znaleziono URL: {url}")
    
    # 2. Wydobycie ID filmu
    video_id = extract_video_id(url)
    assert video_id == BACKUP_VIDEO_ID
    logger.info(f"ID filmu: {video_id}")
    
    # 3. Pobranie transkrypcji i tytułu
    try:
        logger.info(f"Próbuję pobrać transkrypcję dla {video_id}")
        transcript, title = await get_transcript(video_id, TEST_LANGUAGES)
        
        logger.info(f"Pobrano tytuł filmu: {title}")
        logger.info(f"Długość transkrypcji: {len(transcript)} znaków")
        logger.info(f"Fragment transkrypcji: {transcript[:150]}...")
        
        # Weryfikacja tytułu dla "Me at the zoo"
        assert "Me at the zoo" in title
        
    except (NoTranscriptFound, TranscriptsDisabled, APIConnectionError, APIResponseError) as e:
        logger.warning(f"Nie udało się pobrać transkrypcji dla {video_id}: {e}")
        
        # Jeśli wystąpił wyjątek, spróbuj z głównym filmem
        logger.info(f"Próbuję główny film: {TEST_VIDEO_ID}")
        
        try:
            transcript, title = await get_transcript(TEST_VIDEO_ID, TEST_LANGUAGES)
            
            logger.info(f"Pobrano tytuł głównego filmu: {title}")
            logger.info(f"Długość transkrypcji: {len(transcript)} znaków")
            logger.info(f"Fragment transkrypcji: {transcript[:150]}...")
            
        except (NoTranscriptFound, TranscriptsDisabled, APIConnectionError, APIResponseError) as e:
            logger.error(f"Nie można pobrać transkrypcji dla żadnego z testowych filmów: {e}")
            pytest.skip(f"Nie można pobrać transkrypcji dla żadnego z testowych filmów: {e}")
    
    # 4. "Me at the zoo" ma wystarczająco krótką transkrypcję, więc możemy użyć całości
    logger.info(f"Długość oryginalnej transkrypcji: {len(transcript)} znaków")
    
    # 5. Wygenerowanie streszczenia
    try:
        logger.info("Próbuję wygenerować streszczenie")
        
        summary = await summarize_text(transcript)
        assert summary is not None
        assert len(summary) > 10
        
        logger.info(f"Wygenerowane streszczenie: {summary}")
        
    except QuotaExceededError as e:
        logger.warning(f"Przekroczono limit zapytań API OpenAI: {e}")
        pytest.skip(f"Przekroczono limit zapytań API OpenAI: {e}")
    except Exception as e:
        logger.error(f"Błąd podczas generowania streszczenia: {e}", exc_info=True)
        pytest.skip(f"Błąd podczas generowania streszczenia: {e}")


if __name__ == "__main__":
    # Możliwość uruchomienia testów bezpośrednio z tego pliku
    asyncio.run(test_real_extract_youtube_urls())
    asyncio.run(test_real_extract_video_id())
    
    # Testy wymagające połączenia z YouTube
    try:
        asyncio.run(test_real_get_transcript_and_title())
    except Exception as e:
        logger.error(f"Problemy z połączeniem z YouTube API: {e}", exc_info=True)
    
    # Testy wymagające klucza API OpenAI
    if os.environ.get("OPENAI_API_KEY"):
        asyncio.run(ensure_openai_api_status())  # Sprawdź status API przed uruchomieniem testów
        if OPENAI_API_AVAILABLE:
            try:
                asyncio.run(test_real_summarize_text())
                if os.environ.get("YOUTUBE_TRANSCRIPT_API_TOKEN"):
                    asyncio.run(test_end_to_end_integration())
                else:
                    logger.warning("Pominięto test end-to-end (brak tokenu API youtube-transcript.io)")
            except Exception as e:
                logger.error(f"Problemy z OpenAI API: {e}", exc_info=True)
        else:
            logger.warning(f"Pominięto testy OpenAI (API niedostępne: {OPENAI_STATUS_INFO.get('error')})")
    else:
        logger.warning("Pominięto testy OpenAI (brak klucza API)")