| """Luna writes both words and direction; manual prompts bypass the director.""" |
| import json |
| import time |
|
|
| import httpx |
|
|
| import settings |
| from actor_prompt import SYSTEM |
|
|
|
|
| class Director: |
| def __init__(self): |
| self.client = httpx.AsyncClient(timeout=httpx.Timeout(30, connect=10)) |
|
|
| async def close(self): |
| await self.client.aclose() |
|
|
| async def direct(self, message, fmt, language, mode, history): |
| key = settings.luna_key() |
| if not key: |
| raise ValueError("Luna API-Key fehlt. Manuellen Prompt verwenden oder SMALL_LUNA_API_KEY konfigurieren.") |
| context = [{"role": "system", "content": SYSTEM}] |
| for item in history[-16:]: |
| if item.get("role") in {"user", "assistant"} and isinstance(item.get("content"), str): |
| context.append({"role": item["role"], "content": item["content"][:3000]}) |
| task = ("The entire input below is fixed spoken text. Perform these words EXACTLY, without any extra words." |
| if mode == "perform" else |
| "Answer conversationally, or actually enact the user's acting brief. Interpret the scene; do not echo the assignment or describe your performance. Preserve words explicitly supplied to be spoken.") |
| context.append({"role": "user", "content": f"Format: {fmt}. Spoken language: {language}. {task}\n{message}"}) |
| started = time.monotonic() |
| for attempt in range(2): |
| try: |
| response = await self.client.post(settings.LUNA_BASE.rstrip("/") + "/v1/chat/completions", |
| headers={"Authorization": f"Bearer {key}"}, |
| json={"model": settings.LUNA_MODEL, "messages": context, |
| "reasoning_effort": "none", "max_completion_tokens": 2400, |
| "response_format": {"type": "json_object"}}) |
| except httpx.TimeoutException as exc: |
| raise RuntimeError("Luna hat nicht rechtzeitig geantwortet. Bitte erneut senden oder einen manuellen Prompt verwenden.") from exc |
| except httpx.RequestError as exc: |
| raise RuntimeError("Luna API ist momentan nicht erreichbar. Bitte erneut versuchen.") from exc |
| if response.status_code >= 400: |
| raise RuntimeError(f"Luna API antwortet mit HTTP {response.status_code}.") |
| raw = response.json()["choices"][0]["message"].get("content") or "" |
| try: |
| result = json.loads(raw) |
| if not all(k in result for k in ("text", "general", "script", "caption", "tags", "language")): |
| raise ValueError("Unvollständige Regie-Antwort") |
| if not all(isinstance(result[k], str) for k in ("text", "general", "script", "caption", "language")) or not isinstance(result["tags"], list) or not all(isinstance(tag, str) for tag in result["tags"]): |
| raise ValueError("Ungültige Regie-Antwort") |
| from prompting import plain_script |
| spoken = plain_script(result["script"]) if fmt == "structured" else " ".join(result["text"].split()) |
| if not spoken: |
| raise ValueError("Leeres gesprochenes Skript") |
| if mode == "perform" and spoken != " ".join(message.split()): |
| raise ValueError("Luna hat die vorgegebenen Worte geändert") |
| |
| |
| |
| result["text"] = spoken |
| return result, round((time.monotonic() - started) * 1000) |
| except (ValueError, TypeError): |
| if attempt: |
| raise ValueError("Luna lieferte zweimal keine gültige, wortgetreue Regie. Bitte erneut versuchen.") |
| context.append({"role": "assistant", "content": raw[:5000]}) |
| context.append({"role": "user", "content": "Correct the JSON. Keep text and spoken script identical; preserve any exact requested transcript."}) |
|
|
