trade-assistant/backend/app/ai/providers/alibaba.py

from typing import Dict, Any, Optional
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.auth.credentials import AccessKeyCredential
from aliyunsdkalimt.request.v20181012 import TranslateGeneralRequest, TranslateECommerceRequest
from app.services.translation_quota import TranslationQuotaService
from app.database import AsyncSessionLocal
import asyncio
import json
import logging
import os

logger = logging.getLogger(__name__)

ALIBABA_LANG_MAP = {
    "zh": "zh", "en": "en", "ja": "ja", "ko": "ko",
    "fr": "fr", "de": "de", "es": "es", "pt": "pt",
    "ru": "ru", "ar": "ar", "th": "th", "vi": "vi",
    "id": "id", "ms": "ms", "tl": "tl", "hi": "hi",
}

ECS_METADATA_URL = "http://100.100.100.200/latest/meta-data/ram/security-credentials/"


def _fetch_ecs_ram_credentials():
    try:
        import urllib.request
        req = urllib.request.Request(ECS_METADATA_URL, method="GET")
        with urllib.request.urlopen(req, timeout=2) as resp:
            role_name = resp.read().decode().strip()
        if not role_name:
            logger.warning("ECS metadata returned empty role name")
            return None
        url = f"{ECS_METADATA_URL}{role_name}"
        req = urllib.request.Request(url, method="GET")
        with urllib.request.urlopen(req, timeout=2) as resp:
            data = json.loads(resp.read().decode())
        if data.get("Code") == "Success":
            logger.info(f"Fetched STS token for role {role_name}, expires {data.get('Expiration')}")
            return (data["AccessKeyId"], data["AccessKeySecret"], data["SecurityToken"])
        else:
            logger.warning(f"ECS metadata returned non-success: {data.get('Code')}")
    except Exception as e:
        logger.debug(f"ECS metadata fetch failed: {e}")
    return None


def _build_acs_client(access_key_id: str = "", access_key_secret: str = "",
                      region_id: str = "cn-hangzhou") -> AcsClient:
    creds = _fetch_ecs_ram_credentials()
    if creds:
        ak, sk, token = creds
        sts_cred = StsTokenCredential(ak, sk, token)
        client = AcsClient(credential=sts_cred, region_id=region_id)
        logger.info("Alibaba MT using ECS RAM role (STS token)")
        return client

    ak = access_key_id or os.getenv("ALIBABA_ACCESS_KEY_ID", "")
    sk = access_key_secret or os.getenv("ALIBABA_ACCESS_KEY_SECRET", "")
    if ak and sk:
        logger.info("Alibaba MT using AccessKey credentials")
        return AcsClient(ak, sk, region_id)

    raise ValueError("No Alibaba Cloud credentials found (neither ECS RAM role nor AccessKey)")


class AlibabaMTProvider:
    def __init__(self, access_key_id: str = "", access_key_secret: str = "",
                 region_id: str = "cn-hangzhou"):
        self.client = _build_acs_client(access_key_id, access_key_secret, region_id)
        self._name = "alibaba-mt"

    async def translate(self, text: str, source_lang: Optional[str],
                        target_lang: str, context: Optional[str] = None) -> Dict[str, Any]:
        src = source_lang if source_lang and source_lang != "auto" else "auto"
        tgt = ALIBABA_LANG_MAP.get(target_lang[:2].lower(), "en")

        async with AsyncSessionLocal() as db:
            quota_svc = TranslationQuotaService(db)

            for version in ("ecommerce", "general"):
                if not await quota_svc.check_quota(version):
                    logger.info(f"Quota [{version}] exhausted or disabled, trying next")
                    continue

                result = await self._do_translate(version, text, src, tgt)
                if result and result.get("translated_text"):
                    await quota_svc.consume(version, len(text))
                    await db.commit()
                    result["provider"] = f"{self.name}-{version}"
                    return result

        raise Exception("Alibaba MT: both versions quota exhausted or API failed")

    async def _do_translate(self, version: str, text: str, src: str,
                            tgt: str) -> Optional[Dict[str, Any]]:
        try:
            if version == "ecommerce":
                req = TranslateECommerceRequest.TranslateECommerceRequest()
            else:
                req = TranslateGeneralRequest.TranslateGeneralRequest()

            req.set_FormatType("text")
            req.set_Scene(version)
            req.set_SourceLanguage(src)
            req.set_TargetLanguage(tgt)
            req.set_SourceText(text)

            loop = asyncio.get_event_loop()
            body = await loop.run_in_executor(None, self.client.do_action_with_exception, req)
            resp = json.loads(body)
            data = resp.get("Data", {})
            translated = data.get("Translated", "")
            detected = data.get("DetectedLanguage", src)

            if translated:
                logger.info(f"Alibaba MT [{version}] ok: {len(text)} chars translated")
                return {
                    "translated_text": translated,
                    "provider": f"{self.name}-{version}",
                    "detected_source_lang": detected,
                    "char_count": len(text),
                    "cost": 0,
                }
        except Exception as e:
            logger.warning(f"Alibaba MT [{version}] failed: {e}")

        return None

    async def reply(self, *args, **kwargs) -> Dict[str, Any]:
        raise NotImplementedError("Alibaba MT does not support reply generation")

    async def generate_marketing(self, *args, **kwargs) -> Dict[str, Any]:
        raise NotImplementedError("Alibaba MT does not support marketing generation")

    async def extract_info(self, text: str, schema: Dict[str, Any]) -> Dict[str, Any]:
        raise NotImplementedError("Alibaba MT does not support info extraction")

    @property
    def name(self) -> str:
        return self._name

    @property
    def cost_per_1k_tokens(self) -> float:
        return 0