reallife_asset/reallife_asset/rl_volcengine_client.py

"""
Volcengine Ark API Client for Real Person Portrait Asset Management.
Implements V4 HMAC-SHA256 signing.
Uses StreamHttpClient for HTTP requests.
Reads AK/SK from rl_vendor_config table.
"""
import json
import hashlib
import hmac
import datetime
from appPublic.log import debug, error
from appPublic.streamhttpclient import StreamHttpClient
from ahserver.serverenv import ServerEnv

SERVICE = "ark"
REGION = "cn-beijing"
VERSION = "2024-01-01"
HOST = "open.volcengineapi.com"
BASE_URL = f"https://{HOST}"


def _sign(key, msg):
    """HMAC-SHA256 sign."""
    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()


def _get_signature_key(secret_key, date_stamp, region, service):
    """Derive the signing key."""
    k_date = _sign(secret_key.encode("utf-8"), date_stamp)
    k_region = _sign(k_date, region)
    k_service = _sign(k_region, service)
    k_signing = _sign(k_service, "request")
    return k_signing


async def call_volcengine_api(vendor, operation, params, api_mapping):
    """
    Call Volcengine Ark API with V4 signing.
    
    Args:
        vendor: vendor identifier (e.g., "volcengine")
        operation: internal operation name (e.g., "create_session")
        params: API parameters dict
        api_mapping: dict mapping operation to API action name
    
    Returns:
        dict: API response or error
    """
    # Get action name from mapping
    action = api_mapping.get(operation)
    if not action:
        return {"error": f"未配置操作: {operation}"}
    
    # Read AK/SK from vendor config
    from sqlor.dbpools import DBPools
    dbname = ServerEnv().get_module_dbname("reallife_asset")
    db = DBPools()
    async with db.sqlorContext(dbname) as sor:
        recs = await sor.R("rl_vendor_config", {"vendor": vendor})
        if not recs:
            return {"error": f"供应商配置不存在: {vendor}"}
        rec = recs[0]
        ak = getattr(rec, "ak", "")
        sk_encrypted = getattr(rec, "sk", "")
        
        if not ak or not sk_encrypted:
            return {"error": "AK/SK未配置"}
        
        # Decrypt SK
        env = ServerEnv()
        sk = env.password_decode(sk_encrypted)
    
    # Build signed request
    now = datetime.datetime.utcnow()
    date_stamp = now.strftime("%Y%m%d")
    amz_date = now.strftime("%Y%m%dT%H%M%SZ")
    
    # Query string with Action and Version
    query_params = f"Action={action}&Version={VERSION}"
    
    # Headers
    content_type = "application/json"
    payload = json.dumps(params, ensure_ascii=False)
    payload_hash = hashlib.sha256(payload.encode("utf-8")).hexdigest()
    
    headers_to_sign = {
        "host": HOST,
        "x-date": amz_date,
        "x-content-sha256": payload_hash,
        "content-type": content_type,
    }
    signed_headers = ";".join(sorted(headers_to_sign.keys()))
    canonical_headers = "".join(
        f"{k}:{v}\n" for k, v in sorted(headers_to_sign.items())
    )
    
    # Canonical request
    canonical_request = "\n".join([
        "POST",
        "/",
        query_params,
        canonical_headers,
        signed_headers,
        payload_hash,
    ])
    
    # String to sign
    credential_scope = f"{date_stamp}/{REGION}/{SERVICE}/request"
    string_to_sign = "\n".join([
        "HMAC-SHA256",
        amz_date,
        credential_scope,
        hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(),
    ])
    
    # Signing key and signature
    signing_key = _get_signature_key(sk, date_stamp, REGION, SERVICE)
    signature = hmac.new(
        signing_key, string_to_sign.encode("utf-8"), hashlib.sha256
    ).hexdigest()
    
    # Authorization header
    authorization = (
        f"HMAC-SHA256 "
        f"Credential={ak}/{credential_scope}, "
        f"SignedHeaders={signed_headers}, "
        f"Signature={signature}"
    )
    
    # Build request
    url = f"{BASE_URL}/?{query_params}"
    req_headers = {
        "Host": HOST,
        "X-Date": amz_date,
        "X-Content-Sha256": payload_hash,
        "Content-Type": content_type,
        "Authorization": authorization,
    }
    
    # Make HTTP request
    try:
        hc = StreamHttpClient()
        raw = await hc.request("POST", url, headers=req_headers, data=payload.encode("utf-8"))
        
        # Parse response
        if isinstance(raw, bytes):
            raw = raw.decode("utf-8")
        result = json.loads(raw) if raw else {}
        debug(f"volcengine {operation} response: {result}")
        return result
    except Exception as e:
        error(f"volcengine {operation} error: {e}")
        return {"error": str(e)}