translate.py

import json
import uuid
from typing import Any

from config import ANTI_P5JS_PROMPT, DEFAULT_MODEL, SUPPORTED_MODELS
from tools import (
    build_tools_system_prompt,
    extract_tool_results_from_content,
    extract_tool_uses_from_content,
    render_assistant_tool_uses_as_xml,
    render_tool_results_as_xml,
    normalize_content,
)


def pick_model(requested: str | None) -> str:
    if not requested:
        return DEFAULT_MODEL
    if requested in SUPPORTED_MODELS:
        return requested
    lowered = requested.lower()
    for m in SUPPORTED_MODELS:
        if m in lowered or lowered in m:
            return m
    return DEFAULT_MODEL


def build_payload(model: str, upstream_messages: list[dict]) -> dict:
    return {
        "messages": upstream_messages,
        "provider": "anthropic",
        "model": model,
        "deviceId": str(uuid.uuid4()),
        "sessionId": str(uuid.uuid4()),
    }


def build_upstream_messages_anthropic(system: Any, messages: list[dict], tools: list[dict] | None) -> list[dict]:
    upstream: list[dict] = []
    system_text = normalize_content(system) if system else ""
    tools_prompt = build_tools_system_prompt(tools)
    combined = [ANTI_P5JS_PROMPT]
    if system_text:
        combined.append(system_text)
    if tools_prompt:
        combined.append(tools_prompt)
    upstream.append({"role": "system", "content": "\n\n".join(combined)})

    for m in messages:
        role = m.get("role", "user")
        content = m.get("content", "")
        if role == "assistant" and isinstance(content, list):
            text, tool_uses = extract_tool_uses_from_content(content)
            upstream.append({"role": "assistant", "content": render_assistant_tool_uses_as_xml(text, tool_uses)})
        elif role == "user" and isinstance(content, list):
            text, tool_results = extract_tool_results_from_content(content)
            body = []
            if text:
                body.append(text)
            if tool_results:
                body.append(render_tool_results_as_xml(tool_results))
            upstream.append({"role": "user", "content": "\n".join(body)})
        else:
            if role not in ("user", "assistant", "system"):
                role = "user"
            upstream.append({"role": role, "content": normalize_content(content)})
    return upstream


def build_upstream_messages_openai(messages: list[dict], tools: list[dict] | None) -> list[dict]:
    upstream: list[dict] = []
    pending_system_tools = build_tools_system_prompt(tools)

    for m in messages:
        role = m.get("role", "user")
        if role == "system":
            base = normalize_content(m.get("content", ""))
            upstream.append({"role": "system", "content": base})
        elif role == "tool":
            tool_call_id = m.get("tool_call_id", "")
            content = normalize_content(m.get("content", ""))
            upstream.append({
                "role": "user",
                "content": render_tool_results_as_xml([{"id": tool_call_id, "content": content}]),
            })
        elif role == "assistant":
            text = normalize_content(m.get("content", ""))
            tool_calls = m.get("tool_calls") or []
            if tool_calls:
                tus = []
                for tc in tool_calls:
                    fn = tc.get("function", {})
                    args_raw = fn.get("arguments", "{}")
                    try:
                        args = json.loads(args_raw) if isinstance(args_raw, str) else args_raw
                    except Exception:
                        args = {"_raw": args_raw}
                    tus.append({"id": tc.get("id", ""), "name": fn.get("name", ""), "input": args})
                upstream.append({"role": "assistant", "content": render_assistant_tool_uses_as_xml(text, tus)})
            else:
                upstream.append({"role": "assistant", "content": text})
        else:
            upstream.append({"role": "user", "content": normalize_content(m.get("content", ""))})

    extras = [ANTI_P5JS_PROMPT]
    if pending_system_tools:
        extras.append(pending_system_tools)
    extras_text = "\n\n".join(extras)

    has_system = any(x["role"] == "system" for x in upstream)
    if has_system:
        for x in upstream:
            if x["role"] == "system":
                x["content"] = (extras_text + "\n\n" + x["content"]).strip()
                break
    else:
        upstream.insert(0, {"role": "system", "content": extras_text})

    return upstream