ARIA-AGENT/aria-brain/skill_templates.py

"""
Skill-Templates — Boilerplate fuer haeufige Skill-Pattern.

ARIA muss nicht jedes Mal einen kompletten Python-Skill aus dem Nichts
generieren. Sie ruft `skill_scaffold(name, template, params)`, Brain
expandiert das Template und legt den Skill an. Hoehere Skill-Adoption
weil niedrigere Bauh-Huerde.

Templates sind ueber Token-Replacement parametrisiert (kein f-String —
das wuerde mit dem skill-internen Python-Code kollidieren).
"""

from __future__ import annotations

import re
from typing import Callable


# ── Hilfsfunktion ────────────────────────────────────────────────────

def _replace_tokens(s: str, tokens: dict) -> str:
    """Ersetzt {{TOKEN}}-Platzhalter durch Werte. Robust gegen f-String-
    Konflikte im Python-Code des Skills."""
    out = s
    for k, v in tokens.items():
        out = out.replace("{{" + k + "}}", str(v))
    return out


# ── Template 1: oauth-api ────────────────────────────────────────────
# Wrappt eine OAuth2-API. Token kommt aus dem Brain (Auto-Refresh).

_OAUTH_API_CODE = '''"""
{{NAME}} — OAuth2-API-Wrapper fuer {{SERVICE}}.

Holt Token vom Brain (Auto-Refresh) und ruft HTTP-Endpoints der {{SERVICE}}-API.
Keine hardcoded Credentials — alles ueber das zentrale OAuth-System.

Args (alle als ENV ARG_<NAME>):
  ARG_METHOD   = GET | POST | PUT | DELETE | PATCH (Default GET)
  ARG_PATH     = API-Pfad inkl. Query-String (z.B. /v1/me/player)
  ARG_BODY     = JSON-Body als String (optional, fuer POST/PUT/PATCH)
  ARG_BASE_URL = Override der Default-Base-URL (optional)

Exit-Codes: 0 ok, 1 Fehler, 2 nicht autorisiert (Re-Login noetig)
"""
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request

BRAIN_URL = os.environ.get("BRAIN_INTERNAL_URL", "http://localhost:8080")
DEFAULT_BASE_URL = "{{DEFAULT_BASE_URL}}"
SERVICE = "{{SERVICE}}"


def get_token() -> str:
    try:
        with urllib.request.urlopen(
            f"{BRAIN_URL}/oauth/{SERVICE}/token", timeout=10,
        ) as r:
            return json.loads(r.read())["access_token"]
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", "replace")[:400]
        if e.code == 401:
            print(f"NICHT AUTORISIERT: {SERVICE}-Token abgelaufen oder nie gesetzt. "
                  f"ARIA-Tool 'oauth_authorize' nutzen. Details: {body}", file=sys.stderr)
            sys.exit(2)
        print(f"Token-Holen fehlgeschlagen: HTTP {e.code} - {body}", file=sys.stderr)
        sys.exit(1)
    except Exception as e:
        print(f"Token-Holen fehlgeschlagen: {e}", file=sys.stderr)
        sys.exit(1)


def main() -> int:
    method = (os.environ.get("ARG_METHOD") or "GET").upper()
    path = (os.environ.get("ARG_PATH") or "").strip()
    body_raw = (os.environ.get("ARG_BODY") or "").strip()
    base_url = (os.environ.get("ARG_BASE_URL") or DEFAULT_BASE_URL).rstrip("/")
    if not path:
        print(json.dumps({"ok": False, "error": "ARG_PATH erforderlich"}), file=sys.stderr)
        return 1
    if not path.startswith("/"):
        path = "/" + path
    url = base_url + path
    headers = {"Authorization": f"Bearer {get_token()}"}
    data = None
    if body_raw and method in ("POST", "PUT", "PATCH"):
        data = body_raw.encode("utf-8")
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, method=method, headers=headers)
    try:
        with urllib.request.urlopen(req, timeout=20) as r:
            txt = r.read().decode("utf-8")
            parsed = json.loads(txt) if txt and txt[:1] in "[{" else txt
            print(json.dumps({"ok": True, "status": r.status, "data": parsed},
                             ensure_ascii=False, indent=2))
            return 0
    except urllib.error.HTTPError as e:
        txt = e.read().decode("utf-8", "replace")
        try: parsed = json.loads(txt)
        except Exception: parsed = txt[:800]
        print(json.dumps({"ok": False, "status": e.code, "error": parsed},
                         ensure_ascii=False, indent=2))
        return 1


if __name__ == "__main__":
    sys.exit(main())
'''

_OAUTH_API_README = '''# {{NAME}}

OAuth2-API-Wrapper fuer **{{SERVICE}}**. Generiert via `skill_scaffold(template="oauth-api")`.

Holt den Token vom Brain (Auto-Refresh) und macht beliebige HTTP-Calls gegen
die {{SERVICE}}-API. Keine hardcoded Credentials — die Auth-Pipeline laeuft
zentral ueber das Brain-OAuth-System.

## Voraussetzung

- OAuth-App fuer **{{SERVICE}}** im Brain registriert (Diagnostic → OAuth-Apps → client_id + client_secret eintragen)
- Einmaliges `oauth_authorize {{SERVICE}}` zum Initial-Login

## Args

| Name | Default | Beschreibung |
|------|---------|--------------|
| method | GET | HTTP-Methode (GET/POST/PUT/DELETE/PATCH) |
| path | - | API-Pfad mit Query-String (z.B. `/v1/me/player`) |
| body | - | JSON-Body fuer POST/PUT/PATCH |
| base_url | {{DEFAULT_BASE_URL}} | Override der Base-URL falls Sub-API |

## Beispiele

```
method=GET path=/v1/me/player                              # Was laeuft?
method=POST path=/v1/me/player/next                        # Skip
method=PUT path=/v1/me/player/volume?volume_percent=40     # Volume 40
```

Antwort: `{ok, status, data}` als JSON. Bei Fehler `ok=false`.
'''


def _oauth_api(name: str, params: dict) -> dict:
    service = (params.get("service") or name).strip().lower()
    default_base_url = params.get("base_url") or f"https://api.{service}.com"
    tokens = {
        "NAME": name,
        "SERVICE": service,
        "DEFAULT_BASE_URL": default_base_url,
    }
    return {
        "entry_code": _replace_tokens(_OAUTH_API_CODE, tokens),
        "readme": _replace_tokens(_OAUTH_API_README, tokens),
        "pip_packages": [],
        "args": [
            {"name": "method", "type": "string", "required": False,
             "description": "HTTP-Methode (Default GET)"},
            {"name": "path", "type": "string", "required": True,
             "description": "API-Pfad inkl. Query-String, z.B. /v1/me/player"},
            {"name": "body", "type": "string", "required": False,
             "description": "JSON-Body fuer POST/PUT/PATCH"},
            {"name": "base_url", "type": "string", "required": False,
             "description": f"Override der Base-URL (Default {default_base_url})"},
        ],
        "config_schema": [],
        "description": f"OAuth2-API-Wrapper fuer {service}. Token kommt vom Brain (Auto-Refresh).",
    }


# ── Template 2: apikey-api ───────────────────────────────────────────
# Wrappt eine API die mit statischem API-Key/Bearer-Token arbeitet.
# Key liegt in skill.json::config_schema und wird via CFG_<KEY> ENV
# durchgereicht — kein hardcoden, Stefan setzt's in Diagnostic.

_APIKEY_API_CODE = '''"""
{{NAME}} — API-Wrapper fuer {{API_NAME}} mit statischem Key.

Schluessel kommt aus dem Skill-Config (CFG_{{KEY_ENV}}) — Stefan setzt
ihn im Diagnostic-UI bzw. App, NICHT hardcoded.

Args:
  ARG_METHOD   = GET | POST | PUT | DELETE (Default GET)
  ARG_PATH     = API-Pfad inkl. Query-String
  ARG_BODY     = JSON-Body (optional)
  ARG_BASE_URL = Override der Default-Base-URL

Exit-Codes: 0 ok, 1 Fehler, 2 Key nicht gesetzt
"""
import json
import os
import sys
import urllib.error
import urllib.request

DEFAULT_BASE_URL = "{{DEFAULT_BASE_URL}}"
AUTH_HEADER = "{{AUTH_HEADER}}"           # z.B. "Authorization" oder "X-Api-Key"
AUTH_PREFIX = "{{AUTH_PREFIX}}"           # z.B. "Bearer " oder leer


def main() -> int:
    key = os.environ.get("CFG_{{KEY_ENV}}", "").strip()
    if not key:
        print(json.dumps({"ok": False,
                          "error": "API-Key nicht gesetzt — in Diagnostic Skill-Config '{{KEY_ENV}}' eintragen"}),
              file=sys.stderr)
        return 2
    method = (os.environ.get("ARG_METHOD") or "GET").upper()
    path = (os.environ.get("ARG_PATH") or "").strip()
    body_raw = (os.environ.get("ARG_BODY") or "").strip()
    base_url = (os.environ.get("ARG_BASE_URL") or DEFAULT_BASE_URL).rstrip("/")
    if not path:
        print(json.dumps({"ok": False, "error": "ARG_PATH erforderlich"}), file=sys.stderr)
        return 1
    if not path.startswith("/"):
        path = "/" + path
    url = base_url + path
    headers = {AUTH_HEADER: f"{AUTH_PREFIX}{key}"}
    data = None
    if body_raw and method in ("POST", "PUT", "PATCH"):
        data = body_raw.encode("utf-8")
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, method=method, headers=headers)
    try:
        with urllib.request.urlopen(req, timeout=20) as r:
            txt = r.read().decode("utf-8")
            parsed = json.loads(txt) if txt and txt[:1] in "[{" else txt
            print(json.dumps({"ok": True, "status": r.status, "data": parsed},
                             ensure_ascii=False, indent=2))
            return 0
    except urllib.error.HTTPError as e:
        txt = e.read().decode("utf-8", "replace")
        try: parsed = json.loads(txt)
        except Exception: parsed = txt[:800]
        print(json.dumps({"ok": False, "status": e.code, "error": parsed},
                         ensure_ascii=False, indent=2))
        return 1


if __name__ == "__main__":
    sys.exit(main())
'''

_APIKEY_API_README = '''# {{NAME}}

API-Wrapper fuer **{{API_NAME}}** mit statischem API-Key. Generiert via
`skill_scaffold(template="apikey-api")`.

Schluessel ist NICHT im Code, sondern im Skill-Config (`CFG_{{KEY_ENV}}`).
Stefan setzt ihn in Diagnostic → Skills → Detail → Konfiguration.

## Args

| Name | Default | Beschreibung |
|------|---------|--------------|
| method | GET | HTTP-Methode |
| path | - | API-Pfad mit Query-String |
| body | - | JSON-Body |
| base_url | {{DEFAULT_BASE_URL}} | Override |

## Config (in Diagnostic einstellen)

| Feld | Typ | Beschreibung |
|------|-----|--------------|
| {{KEY_ENV}} | password | API-Key fuer {{API_NAME}} |
'''


def _apikey_api(name: str, params: dict) -> dict:
    api_name = params.get("api_name") or name
    key_env = (params.get("key_env") or "API_KEY").upper()
    # safe: nur Buchstaben/Zahlen/Underscore
    key_env = re.sub(r"[^A-Z0-9_]", "_", key_env)
    auth_header = params.get("auth_header") or "Authorization"
    auth_prefix = params.get("auth_prefix") if "auth_prefix" in params else "Bearer "
    default_base_url = params.get("base_url") or "https://api.example.com"
    tokens = {
        "NAME": name,
        "API_NAME": api_name,
        "KEY_ENV": key_env,
        "AUTH_HEADER": auth_header,
        "AUTH_PREFIX": auth_prefix,
        "DEFAULT_BASE_URL": default_base_url,
    }
    return {
        "entry_code": _replace_tokens(_APIKEY_API_CODE, tokens),
        "readme": _replace_tokens(_APIKEY_API_README, tokens),
        "pip_packages": [],
        "args": [
            {"name": "method", "type": "string", "required": False,
             "description": "HTTP-Methode (Default GET)"},
            {"name": "path", "type": "string", "required": True,
             "description": "API-Pfad inkl. Query-String"},
            {"name": "body", "type": "string", "required": False,
             "description": "JSON-Body fuer POST/PUT"},
            {"name": "base_url", "type": "string", "required": False,
             "description": "Override der Base-URL"},
        ],
        "config_schema": [
            {"name": key_env, "type": "password", "label": f"{api_name} API-Key",
             "secret": True, "description": f"Persoenlicher API-Key fuer {api_name}"},
        ],
        "description": f"API-Wrapper fuer {api_name} (Key aus CFG_{key_env}).",
    }


# ── Template 3: file-process ─────────────────────────────────────────
# Nimmt eine Datei aus /shared/uploads/, ruft eine User-Funktion drauf
# auf, schreibt das Resultat nach /shared/uploads/. Skelett — ARIA fuellt
# die `process()`-Funktion danach via skill_update mit dem echten Code.

_FILE_PROCESS_CODE = '''"""
{{NAME}} — File-Processing-Skelett.

Liest eine Eingabe-Datei aus /shared/uploads/, ruft process() auf,
schreibt Output zurueck nach /shared/uploads/.

Args:
  ARG_INPUT  = Pfad zur Eingabedatei (z.B. /shared/uploads/foo.pdf)
  ARG_OUTPUT = Optional Pfad fuer Output (Default: <input>.{{OUTPUT_EXT}})

ARIA-Hinweis: die process()-Funktion ist ein Stub — passe sie via
skill_update an deine Aufgabe an. pip_packages bei Bedarf via
skill_update ergaenzen (z.B. pypdf, Pillow, reportlab).
"""
import os
import shutil
import sys


def process(input_path: str, output_path: str) -> None:
    """Eigentlicher Verarbeitungs-Schritt. Hier kommt der Code rein."""
    # STUB: kopiert die Datei einfach. ARIA: ueberschreibe diese Funktion.
    shutil.copy(input_path, output_path)


def main() -> int:
    inp = (os.environ.get("ARG_INPUT") or "").strip()
    if not inp:
        print("FEHLER: ARG_INPUT erforderlich", file=sys.stderr)
        return 1
    if not os.path.exists(inp):
        print(f"FEHLER: Eingabe nicht gefunden: {inp}", file=sys.stderr)
        return 1
    out = (os.environ.get("ARG_OUTPUT") or "").strip()
    if not out:
        base, _ = os.path.splitext(inp)
        out = f"{base}.{{OUTPUT_EXT}}"
    try:
        process(inp, out)
    except Exception as e:
        print(f"FEHLER bei process(): {e}", file=sys.stderr)
        return 1
    print(out)  # stdout = Pfad zur Ausgabe-Datei, ARIA kann den dem User zurueckgeben
    return 0


if __name__ == "__main__":
    sys.exit(main())
'''

_FILE_PROCESS_README = '''# {{NAME}}

File-Processing-Skelett (`skill_scaffold(template="file-process")`).

Liest eine Datei aus `/shared/uploads/`, ruft die `process()`-Funktion auf,
schreibt das Resultat zurueck. Die `process()`-Funktion ist initial ein
Stub (kopiert nur) — ARIA passt sie via `skill_update` an die konkrete
Aufgabe an.

## Args

| Name | Default | Beschreibung |
|------|---------|--------------|
| input | - | Eingabedatei (z.B. /shared/uploads/foo.pdf) |
| output | `<input>.{{OUTPUT_EXT}}` | Ausgabepfad (optional) |

stdout = Pfad zur erzeugten Datei → ARIA kann ihn dem User zurueckgeben.
'''


def _file_process(name: str, params: dict) -> dict:
    output_ext = (params.get("output_ext") or "out").strip().lstrip(".")
    output_ext = re.sub(r"[^a-zA-Z0-9]", "", output_ext) or "out"
    tokens = {
        "NAME": name,
        "OUTPUT_EXT": output_ext,
    }
    return {
        "entry_code": _replace_tokens(_FILE_PROCESS_CODE, tokens),
        "readme": _replace_tokens(_FILE_PROCESS_README, tokens),
        "pip_packages": [],
        "args": [
            {"name": "input", "type": "string", "required": True,
             "description": "Eingabedatei (z.B. /shared/uploads/foo.pdf)"},
            {"name": "output", "type": "string", "required": False,
             "description": f"Output-Pfad (Default <input>.{output_ext})"},
        ],
        "config_schema": [],
        "description": f"File-Processing-Skelett (Input → process() → Output.{output_ext}).",
    }


# ── Registry ────────────────────────────────────────────────────────

TEMPLATES: dict[str, Callable[[str, dict], dict]] = {
    "oauth-api": _oauth_api,
    "apikey-api": _apikey_api,
    "file-process": _file_process,
}


def list_templates() -> list[dict]:
    """Liste aller verfuegbaren Templates mit Kurzbeschreibung — fuer UI/Tool-Doku."""
    return [
        {
            "name": "oauth-api",
            "description": "OAuth2-API-Wrapper (Spotify, GitHub, Reddit, Google, …). "
                           "Token kommt vom Brain mit Auto-Refresh. Args: method/path/body.",
            "params": ["service (str, OAuth-Service-Name)", "base_url (str, optional)"],
        },
        {
            "name": "apikey-api",
            "description": "API-Wrapper fuer Services mit statischem API-Key "
                           "(OpenWeather, OpenAI, Twilio, …). Key liegt im Skill-Config "
                           "und kommt als CFG_<NAME> ENV — kein hardcode.",
            "params": ["api_name (str)", "key_env (str, ENV-Name fuer den Key)",
                       "auth_header (str, default 'Authorization')",
                       "auth_prefix (str, default 'Bearer ')",
                       "base_url (str)"],
        },
        {
            "name": "file-process",
            "description": "Skelett fuer File-In/File-Out-Operationen "
                           "(PDF konvertieren, Bild bearbeiten, JSON umformen). "
                           "process()-Funktion ist Stub, ARIA fuellt sie via skill_update.",
            "params": ["output_ext (str, Datei-Endung des Outputs)"],
        },
    ]


def expand(name: str, template: str, params: dict | None = None) -> dict:
    """Expandiert ein Template zu einem fertigen Skill-Spec.

    Returns: dict mit entry_code / readme / pip_packages / args /
             config_schema / description — direkt an create_skill weitergebbar.

    Wirft ValueError wenn das Template nicht existiert.
    """
    fn = TEMPLATES.get(template)
    if not fn:
        raise ValueError(
            f"Template '{template}' unbekannt. Verfuegbar: {sorted(TEMPLATES.keys())}"
        )
    return fn(name, params or {})