xiaoxia-saas/apps/api/app/dependencies.py

"""Dependency injection providers for FastAPI endpoints.

All repository and service factories are defined here as FastAPI dependencies,
ensuring proper lifecycle management and testability.
"""

from __future__ import annotations

from typing import Generator

import redis
from app.config import settings
from fastapi import Depends
from sqlalchemy.orm import Session

from packages.adapters.redis import NoopSessionStore, SessionStore
from packages.adapters.smtp import EmailConfig, EmailService, NoopEmailService, get_email_service
from packages.adapters.sqlalchemy_impl.asset_library_repository import (
    SQLAlchemyAssetLibraryRepository,
)
from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
from packages.adapters.sqlalchemy_impl.classification_job_repository import (
    SQLAlchemyClassificationJobRepository,
)
from packages.adapters.sqlalchemy_impl.cover_template_repository import (
    SQLAlchemyCoverTemplateRepository,
)
from packages.adapters.sqlalchemy_impl.duplication_repository import (
    SQLAlchemyDuplicationRecordRepository,
)
from packages.adapters.sqlalchemy_impl.generated_video_repository import (
    SQLAlchemyGeneratedVideoRepository,
)
from packages.adapters.sqlalchemy_impl.generation_task_repository import (
    SQLAlchemyGenerationTaskRepository,
)
from packages.adapters.sqlalchemy_impl.ingest_job_repository import (
    SQLAlchemyIngestJobRepository,
)
from packages.adapters.sqlalchemy_impl.job_repository import SQLAlchemyJobRepository
from packages.adapters.sqlalchemy_impl.project_repository import (
    SQLAlchemyProjectRepository,
)
from packages.adapters.sqlalchemy_impl.session import build_session_factory
from packages.adapters.sqlalchemy_impl.tag_repository import SQLAlchemyTagRepository
from packages.adapters.sqlalchemy_impl.title_library_repository import (
    SQLAlchemyTitleLibraryRepository,
)
from packages.adapters.sqlalchemy_impl.user_repository import SQLAlchemyUserRepository
from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
    SQLAlchemyVoiceCloneProfileRepository,
)
from packages.adapters.sqlalchemy_impl.voice_library_repository import (
    SQLAlchemyVoiceLibraryRepository,
)
from packages.ports.tag_repository import TagRepository
from packages.ports.user_repository import UserRepository

_engine, _SessionLocal = build_session_factory(settings.DATABASE_URL)


def get_db_session() -> Generator[Session, None, None]:
    """Provide a database session with automatic cleanup."""
    session: Session = _SessionLocal()
    try:
        yield session
    finally:
        session.close()


def get_asset_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyAssetRepository:
    """Provide the SQLAlchemy asset repository implementation."""
    return SQLAlchemyAssetRepository(session)


def get_asset_library_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyAssetLibraryRepository:
    """Provide the SQLAlchemy asset library repository implementation."""
    return SQLAlchemyAssetLibraryRepository(session)


def get_ingest_job_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyIngestJobRepository:
    """Provide the SQLAlchemy ingest job repository implementation."""
    return SQLAlchemyIngestJobRepository(session)


def get_classification_job_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyClassificationJobRepository:
    """Provide the SQLAlchemy classification job repository implementation."""
    return SQLAlchemyClassificationJobRepository(session)


def get_generation_task_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyGenerationTaskRepository:
    """Provide the SQLAlchemy generation task repository implementation."""
    return SQLAlchemyGenerationTaskRepository(session)


def get_job_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyJobRepository:
    """Provide the SQLAlchemy job repository implementation."""
    return SQLAlchemyJobRepository(session)


def get_generated_video_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyGeneratedVideoRepository:
    """Provide the SQLAlchemy generated video repository implementation."""
    return SQLAlchemyGeneratedVideoRepository(session)


def get_duplication_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyDuplicationRecordRepository:
    """Provide the SQLAlchemy duplication record repository implementation."""
    return SQLAlchemyDuplicationRecordRepository(session)


def get_project_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyProjectRepository:
    """Provide the SQLAlchemy project repository implementation."""
    return SQLAlchemyProjectRepository(session)


def get_cover_template_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyCoverTemplateRepository:
    """Provide the SQLAlchemy cover template repository implementation."""
    return SQLAlchemyCoverTemplateRepository(session)


def get_tag_repository(
    session: Session = Depends(get_db_session),
) -> TagRepository:
    """Provide the SQLAlchemy tag repository implementation."""
    return SQLAlchemyTagRepository(session)  # type: ignore[return-value]


def get_user_repository(
    session: Session = Depends(get_db_session),
) -> UserRepository:
    """Provide the SQLAlchemy user repository implementation."""
    return SQLAlchemyUserRepository(session)


def get_auth_session_store() -> SessionStore | NoopSessionStore:
    """Provide the session store based on configuration."""
    if not settings.ENABLE_REDIS_SESSIONS:
        return NoopSessionStore()
    return SessionStore(redis_client=redis.from_url(settings.REDIS_URL, decode_responses=True))


def get_auth_email_service() -> NoopEmailService | EmailService:
    """Provide the email service based on configuration."""
    if not settings.ENABLE_EMAIL_DELIVERY:
        return NoopEmailService()
    return get_email_service(
        EmailConfig(
            smtp_host=settings.SMTP_HOST,
            smtp_port=settings.SMTP_PORT,
            smtp_user=settings.SMTP_USER,
            smtp_password=settings.SMTP_PASSWORD,
            from_email=settings.SMTP_FROM_EMAIL,
            from_name=settings.SMTP_FROM_NAME,
            use_tls=settings.SMTP_USE_TLS,
        ),
        enabled=True,
    )


def get_title_library_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyTitleLibraryRepository:
    """Provide the SQLAlchemy title library repository implementation."""
    return SQLAlchemyTitleLibraryRepository(session)


def get_voice_library_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyVoiceLibraryRepository:
    """Provide the SQLAlchemy voice library repository implementation."""
    return SQLAlchemyVoiceLibraryRepository(session)


def get_voice_clone_profile_repository(
    session: Session = Depends(get_db_session),
) -> SQLAlchemyVoiceCloneProfileRepository:
    """Provide the SQLAlchemy voice clone profile repository implementation."""
    return SQLAlchemyVoiceCloneProfileRepository(session)


def get_cosyvoice_service():
    """Provide the CosyVoice service instance.

    注入 OSS 音频URL预签名函数，确保私有bucket下的参考音频
    能被 CosyVoice 服务器下载。
    """
    from app.core.storage import get_storage_service

    from packages.application.cosyvoice_service import CosyVoiceService

    storage = get_storage_service()

    def _sign_audio_url(url: str) -> str:
        """对音频URL做预签名，私有bucket下 CosyVoice 服务器才能下载."""
        return storage.get_download_url(url, expires_seconds=86400)

    return CosyVoiceService(audio_url_signer=_sign_audio_url)


def get_audio_url_signer():
    """提供音频URL预签名函数（24小时有效期）。

    用于所有 API 返回给前端的音频 URL，确保私有 OSS bucket 下可正常访问。
    空 URL、非 OSS URL 直接原样返回；签名失败时回退到原始 URL。
    """
    from app.core.storage import get_storage_service

    storage = get_storage_service()

    def sign_audio_url(url: str) -> str:
        if not url:
            return url
        return storage.get_download_url(url, expires_seconds=86400)

    return sign_audio_url