office_translator/services/providers/minimax_provider.py

"""
Minimax Provider - Cloud LLM translation via the Minimax public API.

Minimax exposes an OpenAI-compatible Chat Completions API at
``https://api.minimax.io/v1/chat/completions`` (default model ``MiniMax-M3``).
Note: ``api.minimax.chat`` is NOT a reachable public host.
"""

import threading
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

import requests
from requests.exceptions import Timeout, ConnectionError as RequestsConnectionError

from core.logging import get_logger
from .base import TranslationProvider
from .schemas import ProviderHealthStatus, TranslationRequest, TranslationResponse

logger = get_logger(__name__)

MINIMAX_RATE_LIMITED = "MINIMAX_RATE_LIMITED"
MINIMAX_INVALID_KEY = "MINIMAX_INVALID_KEY"
MINIMAX_TIMEOUT = "MINIMAX_TIMEOUT"
MINIMAX_SERVICE_ERROR = "MINIMAX_SERVICE_ERROR"

_RETRYABLE_ERRORS = {MINIMAX_RATE_LIMITED, MINIMAX_TIMEOUT, MINIMAX_SERVICE_ERROR}

DEFAULT_TRANSLATION_PROMPT = """You are a professional translator. Translate the following text from {source_lang} to {target_lang}.

Rules:
- Translate ONLY the text, do not add explanations or notes
- Preserve the original formatting, line breaks, and structure
- Maintain the original tone and style
- Translate technical terms, jargon, and labels using the standard target-language equivalent
- Chart elements (titles, axis labels, legend entries, category labels, series names) MUST always be translated, even when they look like a title or a proper noun
- Translate month and weekday abbreviations (Jan → janvier, Mon → lundi) to the target language
- Keep ONLY real proper nouns unchanged: people's names, place names, company names (Google, Microsoft), and product names (GitHub, HuggingFace)
- Do not invent content; if a term is an acronym with no translation (API, URL, HTTP, JSON), keep it as-is"""


def _get_language_name(code: str) -> str:
    """Convert language code to full name (all supported languages)."""
    from core.languages import language_name

    return language_name(code)


def _build_system_prompt(
    source_lang: str, target_lang: str, custom_prompt: Optional[str] = None
) -> str:
    """Build system prompt for translation.

    The base translation instructions are ALWAYS present — a custom prompt
    (glossary, tone, context) is appended as additional directives, never a
    replacement. Same contract as the OpenAI provider.
    """
    base = DEFAULT_TRANSLATION_PROMPT.format(
        source_lang=source_lang, target_lang=target_lang
    )
    if custom_prompt and custom_prompt.strip():
        return f"{base}\n\nADDITIONAL CONTEXT AND INSTRUCTIONS:\n{custom_prompt.strip()}"
    return base


class MinimaxProviderError(Exception):
    def __init__(self, code: str, message: str, details: Optional[Dict[str, Any]] = None):
        self.code = code
        self.message = message
        self.details = details or {}
        super().__init__(message)


class MinimaxTranslationProvider(TranslationProvider):
    """
    Minimax translation provider using OpenAI-compatible API.

    Default model: MiniMax-M3 (latest public OpenAI-compatible model).
    The public endpoint is https://api.minimax.io/v1 (NOT api.minimax.chat,
    which is not a reachable host on the public API).
    """

    def __init__(
        self,
        api_key: str,
        model: str = "MiniMax-M3",
        timeout: int = 60,
        max_retries: int = 3,
        retry_delay: float = 1.0,
        base_url: str = "https://api.minimax.io/v1",
        group_id: str = "",
    ):
        if not api_key or not api_key.strip():
            raise ValueError("Minimax API key cannot be empty")

        self._api_key = api_key
        self._model = model
        self._base_url = base_url.rstrip("/")
        self._group_id = group_id
        self._provider_name = "minimax"
        self._timeout = timeout
        self._max_retries = max_retries
        self._retry_delay = retry_delay
        self._health_cache: Dict[str, Any] = {}
        self._health_cache_ttl = 60
        self._health_cache_lock = threading.Lock()

    def _make_api_request(self, text: str, system_prompt: str) -> tuple:
        if not text or not text.strip():
            return text, {}

        url = f"{self._base_url}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self._api_key}",
            "Content-Type": "application/json",
        }
        payload = {
            "model": self._model,
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": text},
            ],
            "temperature": 0.3,
            "max_tokens": 4096,
        }

        try:
            response = requests.post(url, headers=headers, json=payload, timeout=self._timeout)

            if response.status_code == 401:
                raise MinimaxProviderError(MINIMAX_INVALID_KEY, "Cle API Minimax invalide.")
            if response.status_code == 429:
                raise MinimaxProviderError(MINIMAX_RATE_LIMITED, "Limite de requetes Minimax atteinte.")
            if response.status_code >= 500:
                raise MinimaxProviderError(MINIMAX_SERVICE_ERROR, "Service Minimax temporairement indisponible.")
            if response.status_code != 200:
                raise MinimaxProviderError(MINIMAX_SERVICE_ERROR, f"Erreur Minimax: {response.text[:200]}")

            data = response.json()
            choices = data.get("choices", [])
            if not choices:
                raise MinimaxProviderError(MINIMAX_SERVICE_ERROR, "Reponse Minimax vide")

            content = choices[0].get("message", {}).get("content", "")
            if not content:
                raise MinimaxProviderError(MINIMAX_SERVICE_ERROR, "Reponse Minimax vide")

            return content.strip(), data.get("usage", {})

        except Timeout:
            raise MinimaxProviderError(MINIMAX_TIMEOUT, "Delai d'attente Minimax depasse.")
        except RequestsConnectionError:
            raise MinimaxProviderError(MINIMAX_SERVICE_ERROR, "Service Minimax indisponible.")
        except MinimaxProviderError:
            raise
        except Exception as e:
            raise MinimaxProviderError(MINIMAX_SERVICE_ERROR, f"Erreur Minimax: {str(e)[:100]}")

    def get_name(self) -> str:
        return self._provider_name

    def _probe_available(self) -> tuple[bool, int]:
        """Probe the Minimax API. Returns (available, status_code).

        Minimax does not document a public ``GET /models`` endpoint, so a 404/405
        on that path does NOT mean the provider is down — it only means the path
        is absent. We treat any non-401 response as "available": the host is
        reachable and the API key was not rejected. Only 401 (and network
        errors) mark the provider unavailable.
        """
        try:
            headers = {"Authorization": f"Bearer {self._api_key}"}
            response = requests.get(f"{self._base_url}/models", headers=headers, timeout=5)
            return response.status_code != 401, response.status_code
        except Exception:
            return False, 0

    def is_available(self) -> bool:
        return self._probe_available()[0]

    def translate_text(self, request: TranslationRequest) -> TranslationResponse:
        text = request.text
        target_language = request.target_language
        source_language = request.source_language or "auto"

        if not text or not text.strip():
            return TranslationResponse(translated_text=text, provider_name=self._provider_name, from_cache=False)

        source_lang_name = _get_language_name(source_language)
        target_lang_name = _get_language_name(target_language)
        custom_prompt = request.metadata.get("custom_prompt") if request.metadata else None
        # Base translation instructions always present; the custom prompt
        # (glossary/tone/context) is appended, never a replacement.
        system_prompt = _build_system_prompt(
            source_lang_name, target_lang_name, custom_prompt
        )

        last_error = None
        for attempt in range(self._max_retries + 1):
            try:
                start_time = time.time()
                result, usage = self._make_api_request(text, system_prompt)
                latency = time.time() - start_time

                logger.info("minimax_translation_success",
                    chars=len(text), source_lang=source_language, target_lang=target_language,
                    model=self._model, latency_ms=round(latency * 1000, 2), retries=attempt)

                return TranslationResponse(
                    translated_text=result, provider_name=self._provider_name,
                    from_cache=False, source_language=source_language)

            except MinimaxProviderError as e:
                last_error = e
                if e.code not in _RETRYABLE_ERRORS or attempt >= self._max_retries:
                    break
                delay = self._retry_delay * (2 ** attempt)
                time.sleep(delay)

        return TranslationResponse(
            translated_text=text, provider_name=self._provider_name, from_cache=False,
            error=last_error.message if last_error else "Unknown error",
            error_code=last_error.code if last_error else MINIMAX_SERVICE_ERROR)

    def _make_batch_api_request(
        self, texts: List[str], system_prompt: str
    ) -> Optional[List[str]]:
        """Translate a whole chunk in ONE request via a numbered JSON list.

        The user message is a JSON array; the model must answer with a JSON
        array of the same length. Returns None when the answer cannot be
        parsed confidently — callers then fall back to per-item calls
        (correctness over latency). Mirrors the OpenAI provider batch mode.
        """
        import json as _json

        numbered = _json.dumps(
            [{"id": i, "text": t} for i, t in enumerate(texts)],
            ensure_ascii=False,
        )
        batch_system = (
            system_prompt
            + "\n\nBATCH MODE: the user message is a JSON array of items with "
            "unique ids. Answer with ONLY a JSON array of objects "
            '[{"id": <same id>, "translation": "<translated text>"}], same '
            "length and same ids, in the same order. Translate every item; "
            "keep ids unchanged; no comments, no markdown fence."
        )

        try:
            content, _usage = self._make_api_request(numbered, batch_system)
            raw = content.strip()
            # Strip an optional markdown fence
            if raw.startswith("```"):
                raw = raw.strip("`")
                if raw.lower().startswith("json"):
                    raw = raw[4:]
                raw = raw.strip()
            parsed = _json.loads(raw)
            if not isinstance(parsed, list) or len(parsed) != len(texts):
                return None
            out: List[str] = [""] * len(texts)
            for item in parsed:
                if not isinstance(item, dict):
                    return None
                idx = item.get("id")
                translation = item.get("translation")
                if not isinstance(idx, int) or not 0 <= idx < len(texts):
                    return None
                if not isinstance(translation, str) or not translation.strip():
                    return None
                out[idx] = translation.strip()
            return out
        except MinimaxProviderError:
            raise
        except Exception:
            return None

    def translate_batch(
        self, requests: List[TranslationRequest]
    ) -> List[TranslationResponse]:
        """
        Translate multiple texts.

        Chunks arrive from the translators as ~15 texts. When every request
        shares the same language pair and metadata, they are sent in ONE
        call (numbered JSON list — ~15x fewer requests, better contextual
        consistency across neighbouring segments). Any parse/API doubt falls
        back to the per-item path so a batch failure never corrupts output.
        """
        if not requests:
            return []

        same_pair = len({(r.source_language, r.target_language) for r in requests}) == 1
        same_meta = len(
            {tuple(sorted((r.metadata or {}).items())) for r in requests}
        ) == 1

        if same_pair and same_meta and len(requests) > 1:
            try:
                source_lang_name = _get_language_name(
                    requests[0].source_language or "auto"
                ) or "the source language (auto-detect)"
                target_lang_name = _get_language_name(requests[0].target_language)
                custom_prompt = None
                if requests[0].metadata:
                    custom_prompt = requests[0].metadata.get("custom_prompt")
                system_prompt = _build_system_prompt(
                    source_lang_name, target_lang_name, custom_prompt
                )
                texts = [r.text for r in requests]
                translations = self._make_batch_api_request(texts, system_prompt)
                if translations is not None:
                    logger.info(
                        "minimax_batch_translation_success",
                        items=len(requests),
                        model=self._model,
                    )
                    return [
                        TranslationResponse(
                            translated_text=t,
                            provider_name=self._provider_name,
                            from_cache=False,
                        )
                        for t in translations
                    ]
                logger.warning(
                    "minimax_batch_translation_fallback",
                    reason="unparseable_response",
                    items=len(requests),
                )
            except Exception as e:
                logger.warning(
                    "minimax_batch_translation_fallback",
                    reason=type(e).__name__,
                    items=len(requests),
                )

        return [self.translate_text(req) for req in requests]

    def health_check(self) -> ProviderHealthStatus:
        start_time = time.time()
        available, status_code = self._probe_available()
        latency_ms = (time.time() - start_time) * 1000
        if available:
            return ProviderHealthStatus(
                name=self._provider_name, available=True,
                latency_ms=round(latency_ms, 2), last_check=datetime.now(timezone.utc).isoformat(),
                model=self._model)
        return ProviderHealthStatus(
            name=self._provider_name, available=False,
            latency_ms=round(latency_ms, 2),
            error=f"probe failed (status={status_code})"[:100],
            last_check=datetime.now(timezone.utc).isoformat(),
            model=self._model)


_provider_instance: Optional[MinimaxTranslationProvider] = None
_provider_lock = threading.Lock()


def get_minimax_provider() -> MinimaxTranslationProvider:
    global _provider_instance
    if _provider_instance is None:
        with _provider_lock:
            if _provider_instance is None:
                from .config import ProvidersConfig
                _provider_instance = MinimaxTranslationProvider(
                    api_key=ProvidersConfig.MINIMAX_API_KEY,
                    model=ProvidersConfig.MINIMAX_MODEL,
                    timeout=ProvidersConfig.MINIMAX_TIMEOUT,
                    max_retries=ProvidersConfig.MINIMAX_MAX_RETRIES,
                    retry_delay=ProvidersConfig.MINIMAX_RETRY_DELAY,
                    base_url=ProvidersConfig.MINIMAX_BASE_URL,
                    group_id=ProvidersConfig.MINIMAX_GROUP_ID,
                )
    return _provider_instance


def register_minimax_provider():
    from .registry import registry
    provider = get_minimax_provider()
    registry.register("minimax", provider)
    return provider


def reset_minimax_provider():
    global _provider_instance
    with _provider_lock:
        _provider_instance = None