services_config.py 4.56 KB
"""
Unified service configuration accessors.

This module is now a thin adapter over ``config.loader.get_app_config`` and
contains no independent parsing or precedence logic.
"""

from __future__ import annotations

import os
from typing import Any, Dict, Tuple

from config.loader import get_app_config
from config.schema import (
    EmbeddingServiceConfig,
    RerankServiceConfig,
    RerankServiceInstanceConfig,
    TranslationServiceConfig,
)


def get_translation_config() -> Dict[str, Any]:
    return get_app_config().services.translation.as_dict()


def get_embedding_config() -> EmbeddingServiceConfig:
    return get_app_config().services.embedding


def get_rerank_config() -> RerankServiceConfig:
    return get_app_config().services.rerank


def get_rerank_instance_config(profile: str | None = None) -> RerankServiceInstanceConfig:
    cfg = get_app_config().services.rerank
    instance_name = str(
        profile
        or os.getenv("RERANK_INSTANCE")
        or cfg.default_instance
    ).strip() or cfg.default_instance
    return cfg.get_instance(instance_name)


def get_translation_base_url() -> str:
    return get_app_config().services.translation.endpoint


def get_translation_cache_config() -> Dict[str, Any]:
    return dict(get_app_config().services.translation.cache)


def get_embedding_text_base_url() -> str:
    provider_cfg = get_app_config().services.embedding.get_provider_config()
    base = provider_cfg.get("text_base_url")
    if not base:
        raise ValueError("Embedding text base_url is not configured")
    return str(base).rstrip("/")


def get_embedding_image_base_url() -> str:
    provider_cfg = get_app_config().services.embedding.get_provider_config()
    base = provider_cfg.get("image_base_url")
    if not base:
        raise ValueError("Embedding image base_url is not configured")
    return str(base).rstrip("/")


def get_embedding_backend_config() -> Tuple[str, Dict[str, Any]]:
    cfg = get_app_config().services.embedding
    return cfg.backend, cfg.get_backend_config()


def get_embedding_image_backend_config() -> Tuple[str, Dict[str, Any]]:
    cfg = get_app_config().services.embedding
    return cfg.image_backend, cfg.get_image_backend_config()


def get_rerank_backend_config(profile: str | None = None) -> Tuple[str, Dict[str, Any]]:
    cfg = get_app_config().services.rerank
    instance = get_rerank_instance_config(profile)
    backend = str(os.getenv("RERANK_BACKEND") or instance.backend).strip()
    if backend != instance.backend:
        backend_cfg = cfg.backends.get(backend)
        if backend_cfg is None:
            raise ValueError(f"Unknown rerank backend override from RERANK_BACKEND: {backend!r}")
        return backend, dict(backend_cfg)
    return instance.backend, cfg.get_backend_config(profile)


def get_rerank_base_url(profile: str | None = None) -> str:
    cfg = get_app_config().services.rerank
    instance = get_rerank_instance_config(profile)
    provider_cfg = cfg.get_provider_config()
    profile_name = str(profile or os.getenv("RERANK_INSTANCE") or cfg.default_instance).strip() or cfg.default_instance

    base = None
    provider_instances = provider_cfg.get("instances")
    if isinstance(provider_instances, dict):
        instance_provider_cfg = provider_instances.get(profile_name)
        if isinstance(instance_provider_cfg, dict):
            base = instance_provider_cfg.get("base_url") or instance_provider_cfg.get("service_url")
    if not base:
        base = instance.base_url or instance.service_url
    if not base:
        base = provider_cfg.get("service_url") or provider_cfg.get("base_url")
    if not base:
        raise ValueError("Rerank service URL is not configured")
    base = str(base).rstrip("/")
    if base.endswith("/rerank"):
        base = base[: -len("/rerank")]
    return base


def get_rerank_service_url(profile: str | None = None) -> str:
    cfg = get_app_config().services.rerank
    instance = get_rerank_instance_config(profile)
    provider_cfg = cfg.get_provider_config()
    profile_name = str(profile or os.getenv("RERANK_INSTANCE") or cfg.default_instance).strip() or cfg.default_instance

    service_url = None
    provider_instances = provider_cfg.get("instances")
    if isinstance(provider_instances, dict):
        instance_provider_cfg = provider_instances.get(profile_name)
        if isinstance(instance_provider_cfg, dict):
            service_url = instance_provider_cfg.get("service_url")
    if not service_url:
        service_url = instance.service_url
    if not service_url:
        service_url = f"{get_rerank_base_url(profile=profile)}/rerank"
    return str(service_url).rstrip("/")