555 lines
20 KiB
Python
555 lines
20 KiB
Python
"""
|
|
keys-bridges — Headless-Chromium PWA-Bridge-Hub.
|
|
Endpoints:
|
|
GET /healthz → ok
|
|
GET /status → login-status aller Bridges
|
|
POST /mammouth/chat → {prompt, model?} → {content}
|
|
POST /suno/generate → {prompt, instrumental?} → {audio_urls}
|
|
POST /gemini/chat → {prompt} → {content}
|
|
noVNC läuft separat auf :4004 (über supervisord)
|
|
"""
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from bridges.mammouth import Mammouth
|
|
from bridges.suno import Suno
|
|
from bridges.gemini import Gemini
|
|
|
|
app = FastAPI(title="keys-bridges")
|
|
|
|
BRIDGES = {"mammouth": Mammouth(), "suno": Suno(), "gemini": Gemini()}
|
|
|
|
class ChatIn(BaseModel):
|
|
prompt: str
|
|
model: str | None = None
|
|
|
|
class SunoIn(BaseModel):
|
|
prompt: str
|
|
instrumental: bool = False
|
|
|
|
@app.get("/healthz")
|
|
async def health(): return {"ok": True}
|
|
|
|
@app.get("/status")
|
|
async def status():
|
|
out = {}
|
|
for n, b in BRIDGES.items():
|
|
try:
|
|
out[n] = await b.login_status()
|
|
except Exception as e:
|
|
out[n] = {"error": str(e), "logged_in": False}
|
|
return out
|
|
|
|
@app.post("/mammouth/chat")
|
|
async def mammouth_chat(p: ChatIn):
|
|
try:
|
|
content = await BRIDGES["mammouth"].chat(p.prompt, p.model or "auto")
|
|
return {"content": content, "model": p.model or "auto"}
|
|
except Exception as e:
|
|
raise HTTPException(500, f"mammouth: {e}")
|
|
|
|
@app.post("/gemini/chat")
|
|
async def gemini_chat(p: ChatIn):
|
|
try:
|
|
content = await BRIDGES["gemini"].chat(p.prompt)
|
|
return {"content": content}
|
|
except Exception as e:
|
|
raise HTTPException(500, f"gemini: {e}")
|
|
|
|
from bridges.qolaba import Qolaba
|
|
BRIDGES["qolaba"] = Qolaba()
|
|
|
|
@app.post("/qolaba/chat")
|
|
async def qolaba_chat(p: ChatIn):
|
|
try:
|
|
content = await BRIDGES["qolaba"].chat(p.prompt, p.model or "chat")
|
|
return {"content": content}
|
|
except Exception as e:
|
|
raise HTTPException(500, f"qolaba: {e}")
|
|
|
|
from bridges.suno import Suno as _SunoCls
|
|
BRIDGES["suno"] = _SunoCls()
|
|
|
|
@app.post("/suno/generate")
|
|
async def suno_submit(p: SunoIn):
|
|
"""Async — returns job_id sofort. Poll /suno/status/<id>."""
|
|
job_id = await BRIDGES["suno"].submit_job(p.prompt, p.instrumental)
|
|
return {"job_id": job_id, "status_url": f"/suno/status/{job_id}"}
|
|
|
|
@app.get("/suno/status/{job_id}")
|
|
async def suno_status(job_id: str):
|
|
job = BRIDGES["suno"].get_job(job_id)
|
|
if not job: raise HTTPException(404, "unknown job_id")
|
|
return job
|
|
|
|
@app.get("/suno/jobs")
|
|
async def suno_jobs():
|
|
return BRIDGES["suno"].list_jobs()
|
|
|
|
from bridges.qolaba_api import qolaba_api
|
|
|
|
class QolabaAPIIn(BaseModel):
|
|
model: str
|
|
messages: list
|
|
|
|
@app.post("/qolaba/api/chat")
|
|
async def qolaba_api_chat(p: QolabaAPIIn):
|
|
content = await qolaba_api.chat_via_ui(p.model, p.messages)
|
|
return {"content": content, "model": p.model}
|
|
|
|
|
|
# --- Nimbus/FuseBase Bridge (Baer 2026-06-24) ---
|
|
from bridges.nimbus import Nimbus
|
|
BRIDGES["nimbus"] = Nimbus()
|
|
|
|
@app.get("/nimbus/status")
|
|
async def nimbus_status(): return await BRIDGES["nimbus"].login_status()
|
|
|
|
@app.get("/nimbus/text")
|
|
async def nimbus_text(path: str = ""): return await BRIDGES["nimbus"].text(path)
|
|
|
|
@app.get("/nimbus/workspaces")
|
|
async def nimbus_workspaces(): return await BRIDGES["nimbus"].workspaces()
|
|
|
|
@app.get("/nimbus/screenshot")
|
|
async def nimbus_screenshot(path: str = ""): return await BRIDGES["nimbus"].screenshot(path)
|
|
|
|
@app.get("/nimbus/goto")
|
|
async def nimbus_goto(path: str = ""): return await BRIDGES["nimbus"].goto(path)
|
|
|
|
|
|
# --- OpenAI-compatible adapter (replaces dead code-x86 adapters) ---
|
|
import time, uuid
|
|
from fastapi import Request
|
|
from fastapi.responses import JSONResponse
|
|
|
|
BRIDGE_ROUTING = {
|
|
"mammouth": "mammouth",
|
|
"qolaba": "qolaba",
|
|
"gemini": "gemini",
|
|
}
|
|
|
|
def _openai_response(content: str, model: str) -> dict:
|
|
return {
|
|
"id": f"chatcmpl-{uuid.uuid4().hex[:12]}",
|
|
"object": "chat.completion",
|
|
"created": int(time.time()),
|
|
"model": model,
|
|
"choices": [{
|
|
"index": 0,
|
|
"message": {"role": "assistant", "content": content},
|
|
"finish_reason": "stop"
|
|
}],
|
|
"usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
|
|
}
|
|
|
|
@app.post("/v1/chat/completions")
|
|
async def openai_chat_completions(request: Request):
|
|
body = await request.json()
|
|
model = body.get("model", "mammouth/auto")
|
|
messages = body.get("messages", [])
|
|
prompt = "\n".join(
|
|
f"{m.get('role','user')}: {m.get('content','')}" for m in messages
|
|
) if messages else ""
|
|
if not prompt and messages:
|
|
prompt = messages[-1].get("content", "")
|
|
|
|
prefix = model.split("/")[0] if "/" in model else model
|
|
sub_model = "/".join(model.split("/")[1:]) if "/" in model else "auto"
|
|
|
|
bridge_name = BRIDGE_ROUTING.get(prefix)
|
|
if not bridge_name:
|
|
return JSONResponse(status_code=400, content={
|
|
"error": {"message": f"Unknown model prefix: {prefix}", "type": "invalid_request_error"}
|
|
})
|
|
|
|
bridge = BRIDGES.get(bridge_name)
|
|
if not bridge:
|
|
return JSONResponse(status_code=503, content={
|
|
"error": {"message": f"Bridge {bridge_name} not loaded", "type": "server_error"}
|
|
})
|
|
|
|
try:
|
|
if bridge_name == "mammouth":
|
|
content = await bridge.chat(prompt, sub_model)
|
|
elif bridge_name == "qolaba":
|
|
content = await bridge.chat(prompt, sub_model)
|
|
elif bridge_name == "gemini":
|
|
content = await bridge.chat(prompt)
|
|
else:
|
|
content = await bridge.chat(prompt)
|
|
except Exception as e:
|
|
return JSONResponse(status_code=500, content={
|
|
"error": {"message": str(e), "type": "server_error"}
|
|
})
|
|
|
|
return _openai_response(content, model)
|
|
|
|
@app.get("/v1/models")
|
|
async def openai_models():
|
|
model_list = []
|
|
for prefix in BRIDGE_ROUTING:
|
|
model_list.append({
|
|
"id": f"{prefix}/auto",
|
|
"object": "model",
|
|
"created": 1700000000,
|
|
"owned_by": "keys-bridges"
|
|
})
|
|
return {"object": "list", "data": model_list}
|
|
|
|
|
|
# --- Veo3 Video-Generation via Gemini PWA (Baer 2026-07-05) ---
|
|
from pydantic import Field
|
|
|
|
class Veo3In(BaseModel):
|
|
prompt: str
|
|
timeout: int = Field(default=300, ge=30, le=600)
|
|
|
|
@app.post("/gemini/veo3")
|
|
async def gemini_veo3(p: Veo3In):
|
|
try:
|
|
result = await BRIDGES["gemini"].veo3(p.prompt, p.timeout)
|
|
return result
|
|
except Exception as e:
|
|
raise HTTPException(500, f"veo3: {e}")
|
|
|
|
@app.get("/gemini/screenshot")
|
|
async def gemini_screenshot():
|
|
try:
|
|
path = await BRIDGES["gemini"].screenshot()
|
|
return {"screenshot": path}
|
|
except Exception as e:
|
|
raise HTTPException(500, f"screenshot: {e}")
|
|
|
|
@app.get("/gemini/veo3/files")
|
|
async def veo3_files():
|
|
from pathlib import Path
|
|
d = Path("/tmp/veo3")
|
|
if not d.exists():
|
|
return {"files": []}
|
|
files = sorted(d.glob("veo3_*.mp4"), key=lambda f: f.stat().st_mtime, reverse=True)
|
|
return {"files": [{"path": str(f), "size": f.stat().st_size, "name": f.name} for f in files[:20]]}
|
|
|
|
@app.get("/gemini/debug/overlay")
|
|
async def gemini_debug_overlay():
|
|
try:
|
|
bridge = BRIDGES["gemini"]
|
|
p = await bridge.get_page()
|
|
result = await p.evaluate("""() => {
|
|
const out = {overlay_text: "", buttons: [], dialogs: []};
|
|
const container = document.querySelector(".cdk-overlay-container");
|
|
if (container) {
|
|
out.overlay_text = container.innerText.substring(0, 2000);
|
|
container.querySelectorAll("button").forEach(b => {
|
|
if (b.offsetHeight > 0) out.buttons.push({text: b.innerText.trim(), ariaLabel: b.getAttribute('aria-label') || ''});
|
|
});
|
|
}
|
|
document.querySelectorAll("[role=dialog], mat-dialog-container, .mat-mdc-dialog-surface").forEach(d => {
|
|
out.dialogs.push(d.innerText.substring(0, 500));
|
|
});
|
|
return out;
|
|
}""")
|
|
return result
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
@app.get("/gemini/debug/page")
|
|
async def gemini_debug_page():
|
|
try:
|
|
bridge = BRIDGES["gemini"]
|
|
p = await bridge.get_page()
|
|
result = await p.evaluate("""() => {
|
|
const out = {
|
|
url: window.location.href,
|
|
title: document.title,
|
|
body_text: document.body ? document.body.innerText.substring(0, 3000) : '',
|
|
textbox_exists: !!document.querySelector('div.ql-editor[role="textbox"]'),
|
|
overlay_visible: false,
|
|
video_count: document.querySelectorAll('video').length,
|
|
model_response_count: document.querySelectorAll('model-response').length,
|
|
message_content_count: document.querySelectorAll('message-content').length,
|
|
};
|
|
const ov = document.querySelector('.cdk-overlay-container');
|
|
if (ov && ov.innerText.trim()) {
|
|
out.overlay_visible = true;
|
|
out.overlay_text = ov.innerText.substring(0, 500);
|
|
}
|
|
// Check last message content
|
|
const msgs = document.querySelectorAll('message-content');
|
|
if (msgs.length > 0) {
|
|
out.last_message = msgs[msgs.length - 1].innerText.substring(0, 500);
|
|
}
|
|
return out;
|
|
}""")
|
|
return result
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
@app.get("/gemini/debug/video-state")
|
|
async def gemini_debug_video_state():
|
|
try:
|
|
bridge = BRIDGES["gemini"]
|
|
p = await bridge.get_page()
|
|
result = await p.evaluate("""() => {
|
|
const out = {
|
|
url: window.location.href,
|
|
jetzt_antworten_found: false,
|
|
jetzt_antworten_is_button: false,
|
|
video_elements: [],
|
|
spinners: [],
|
|
progress_indicators: [],
|
|
};
|
|
// Check all elements for "Jetzt antworten"
|
|
const allElements = document.querySelectorAll('*');
|
|
for (const el of allElements) {
|
|
if (el.children.length === 0 || el.tagName === 'BUTTON') {
|
|
const t = el.textContent.trim();
|
|
if (t === 'Jetzt antworten' || t === 'Answer now') {
|
|
out.jetzt_antworten_found = true;
|
|
out.jetzt_antworten_is_button = el.tagName === 'BUTTON' || el.closest('button') !== null;
|
|
out.jetzt_antworten_tag = el.tagName;
|
|
out.jetzt_antworten_visible = el.offsetHeight > 0;
|
|
}
|
|
}
|
|
// Check for spinners/loading
|
|
if (el.getAttribute('role') === 'progressbar' ||
|
|
el.classList.contains('loading') ||
|
|
el.classList.contains('spinner') ||
|
|
el.tagName === 'MAT-SPINNER' ||
|
|
el.tagName === 'MAT-PROGRESS-SPINNER') {
|
|
if (el.offsetHeight > 0) {
|
|
out.spinners.push({tag: el.tagName, class: el.className.substring(0,100)});
|
|
}
|
|
}
|
|
}
|
|
// Shadow DOM video check
|
|
function findVideos(root) {
|
|
root.querySelectorAll('video').forEach(v => {
|
|
out.video_elements.push({src: v.src || v.currentSrc || 'no-src', ready: v.readyState});
|
|
});
|
|
root.querySelectorAll('*').forEach(el => {
|
|
if (el.shadowRoot) findVideos(el.shadowRoot);
|
|
});
|
|
}
|
|
findVideos(document);
|
|
return out;
|
|
}""")
|
|
return result
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
|
|
# TEMPORAER (2026-07-20): Selektor-Diagnose fuer mammouth.ai
|
|
@app.get("/debug/mammouth-dom")
|
|
async def debug_mammouth_dom():
|
|
b = BRIDGES["mammouth"]
|
|
p = await b.get_page()
|
|
try:
|
|
await p.wait_for_load_state("domcontentloaded", timeout=20000)
|
|
except Exception:
|
|
pass
|
|
return await p.evaluate("""() => {
|
|
const sel = (q) => Array.from(document.querySelectorAll(q));
|
|
const d = (e) => ({
|
|
tag: e.tagName.toLowerCase(),
|
|
platzhalter: e.getAttribute('placeholder'),
|
|
rolle: e.getAttribute('role'),
|
|
aria: e.getAttribute('aria-label'),
|
|
klasse: (e.className || '').toString().slice(0, 80),
|
|
id: e.id,
|
|
editierbar: e.isContentEditable,
|
|
sichtbar: !!(e.offsetParent || e.getClientRects().length)
|
|
});
|
|
return {
|
|
url: location.href,
|
|
titel: document.title,
|
|
textareas: sel('textarea').map(d),
|
|
editierbare: sel('[contenteditable]').map(d),
|
|
eingaben: sel('input').map(d).slice(0, 10),
|
|
knoepfe: sel('button').filter(e => e.offsetParent)
|
|
.map(e => ({text: (e.innerText||'').trim().slice(0,35), aria: e.getAttribute('aria-label')}))
|
|
.slice(0, 15)
|
|
};
|
|
}""")
|
|
|
|
|
|
# TEMPORAER (2026-07-20): Selektor-Diagnose fuer suno.com
|
|
@app.get("/debug/suno-dom")
|
|
async def debug_suno_dom():
|
|
b = BRIDGES["suno"]
|
|
p = await b.get_page()
|
|
try:
|
|
await p.wait_for_load_state("domcontentloaded", timeout=20000)
|
|
except Exception:
|
|
pass
|
|
return await p.evaluate("""() => {
|
|
const sel = (q) => Array.from(document.querySelectorAll(q));
|
|
const d = (e) => ({
|
|
tag: e.tagName.toLowerCase(),
|
|
platzhalter: e.getAttribute('placeholder'),
|
|
rolle: e.getAttribute('role'),
|
|
aria: e.getAttribute('aria-label'),
|
|
klasse: (e.className || '').toString().slice(0, 80),
|
|
id: e.id,
|
|
editierbar: e.isContentEditable,
|
|
sichtbar: !!(e.offsetParent || e.getClientRects().length)
|
|
});
|
|
return {
|
|
url: location.href,
|
|
titel: document.title,
|
|
textareas: sel('textarea').map(d),
|
|
editierbare: sel('[contenteditable]').map(d),
|
|
eingaben: sel('input').map(d).slice(0, 10),
|
|
knoepfe: sel('button').filter(e => e.offsetParent)
|
|
.map(e => ({text: (e.innerText||'').trim().slice(0,35), aria: e.getAttribute('aria-label')}))
|
|
.slice(0, 20)
|
|
};
|
|
}""")
|
|
|
|
|
|
# TEMPORAER (2026-07-20): Selektor-Diagnose fuer qolaba.ai
|
|
@app.get("/debug/qolaba-dom")
|
|
async def debug_qolaba_dom():
|
|
b = BRIDGES["qolaba"]
|
|
p = await b.get_page()
|
|
try:
|
|
await p.wait_for_load_state("domcontentloaded", timeout=20000)
|
|
except Exception:
|
|
pass
|
|
return await p.evaluate("""() => {
|
|
const sel = (q) => Array.from(document.querySelectorAll(q));
|
|
const d = (e) => ({
|
|
tag: e.tagName.toLowerCase(),
|
|
platzhalter: e.getAttribute('placeholder'),
|
|
rolle: e.getAttribute('role'),
|
|
aria: e.getAttribute('aria-label'),
|
|
klasse: (e.className || '').toString().slice(0, 80),
|
|
id: e.id,
|
|
editierbar: e.isContentEditable,
|
|
sichtbar: !!(e.offsetParent || e.getClientRects().length)
|
|
});
|
|
return {
|
|
url: location.href,
|
|
titel: document.title,
|
|
textareas: sel('textarea').map(d),
|
|
editierbare: sel('[contenteditable]').map(d),
|
|
eingaben: sel('input').map(d).slice(0, 10),
|
|
knoepfe: sel('button').filter(e => e.offsetParent)
|
|
.map(e => ({text: (e.innerText||'').trim().slice(0,35), aria: e.getAttribute('aria-label')}))
|
|
.slice(0, 15),
|
|
prose_count: sel('main .prose').length,
|
|
main_html_len: (document.querySelector('main')||{}).innerHTML ? document.querySelector('main').innerHTML.length : 0
|
|
};
|
|
}""")
|
|
|
|
|
|
@app.get("/debug/qolaba-dom2")
|
|
async def debug_qolaba_dom2():
|
|
b = BRIDGES["qolaba"]
|
|
p = await b.get_page()
|
|
return await p.evaluate("""() => {
|
|
const all = Array.from(document.querySelectorAll('main *'));
|
|
const cand = all.filter(e => {
|
|
const cls = (e.className || '').toString().toLowerCase();
|
|
const id = (e.id || '').toLowerCase();
|
|
return /message|markdown|chat|bubble|assistant|response|prose|reply/.test(cls) ||
|
|
/message|markdown|chat|bubble|assistant|response|prose|reply/.test(id);
|
|
});
|
|
const seen = new Set();
|
|
const out = [];
|
|
for (const e of cand) {
|
|
const key = e.tagName + '|' + (e.className||'').toString().slice(0,100);
|
|
if (seen.has(key)) continue;
|
|
seen.add(key);
|
|
out.push({
|
|
tag: e.tagName.toLowerCase(),
|
|
klasse: (e.className||'').toString().slice(0,100),
|
|
id: e.id,
|
|
textlen: (e.innerText||'').length,
|
|
textsnippet: (e.innerText||'').trim().slice(0,80)
|
|
});
|
|
if (out.length >= 40) break;
|
|
}
|
|
return out;
|
|
}""")
|
|
|
|
|
|
@app.get("/debug/qolaba-send")
|
|
async def debug_qolaba_send():
|
|
b = BRIDGES["qolaba"]
|
|
p = await b.get_page()
|
|
try:
|
|
await p.goto("https://www.qolaba.ai/chat", wait_until="domcontentloaded", timeout=20000)
|
|
await p.wait_for_timeout(3000)
|
|
except Exception:
|
|
pass
|
|
ta = await p.wait_for_selector('textarea[placeholder*="typing" i]', timeout=20000)
|
|
await ta.click(); await ta.fill("Sag in einem Satz: welches Datum ist heute laut deinem Wissen?")
|
|
await p.wait_for_timeout(1000)
|
|
sub = await p.query_selector('button[type="submit"]')
|
|
clicked_via = None
|
|
if sub:
|
|
await sub.scroll_into_view_if_needed()
|
|
await sub.click(force=True)
|
|
clicked_via = "submit-button"
|
|
else:
|
|
await p.keyboard.press("Enter")
|
|
clicked_via = "enter-key"
|
|
await p.wait_for_timeout(12000)
|
|
result = await p.evaluate("""() => {
|
|
const all = Array.from(document.querySelectorAll('main *'));
|
|
const cand = all.filter(e => {
|
|
const cls = (e.className || '').toString().toLowerCase();
|
|
const id = (e.id || '').toLowerCase();
|
|
return /message|markdown|chat|bubble|assistant|response|prose|reply|answer|content|text-/.test(cls) ||
|
|
/message|markdown|chat|bubble|assistant|response|prose|reply/.test(id);
|
|
});
|
|
const seen = new Set();
|
|
const out = [];
|
|
for (const e of cand) {
|
|
const key = e.tagName + '|' + (e.className||'').toString().slice(0,100);
|
|
if (seen.has(key)) continue;
|
|
seen.add(key);
|
|
const t = (e.innerText||'').trim();
|
|
if (!t) continue;
|
|
out.push({
|
|
tag: e.tagName.toLowerCase(),
|
|
klasse: (e.className||'').toString().slice(0,120),
|
|
id: e.id,
|
|
textlen: t.length,
|
|
textsnippet: t.slice(0,150)
|
|
});
|
|
if (out.length >= 40) break;
|
|
}
|
|
return out;
|
|
}""")
|
|
return {"clicked_via": clicked_via, "result": result}
|
|
|
|
|
|
@app.get("/debug/qolaba-dom3")
|
|
async def debug_qolaba_dom3():
|
|
b = BRIDGES["qolaba"]
|
|
p = await b.get_page()
|
|
await p.wait_for_timeout(2000)
|
|
return await p.evaluate("""() => {
|
|
const btns = Array.from(document.querySelectorAll('button')).filter(e=>e.offsetParent).map(e=>({text:(e.innerText||'').trim().slice(0,40), cls:(e.className||'').toString().slice(0,60)}));
|
|
const overlays = Array.from(document.querySelectorAll('[class*="cookie" i], [class*="consent" i], [id*="cookie" i], [id*="consent" i], [class*="overlay" i], [class*="modal" i]')).map(e=>({tag:e.tagName, cls:(e.className||'').toString().slice(0,80), visible: !!(e.offsetParent)}));
|
|
return {url: location.href, btns, overlays};
|
|
}""")
|
|
|
|
|
|
@app.get("/debug/qolaba-dom4")
|
|
async def debug_qolaba_dom4():
|
|
b = BRIDGES["qolaba"]
|
|
p = await b.get_page()
|
|
await p.wait_for_timeout(1500)
|
|
return await p.evaluate("""() => {
|
|
const btns = Array.from(document.querySelectorAll('button'));
|
|
return btns.map(e => ({
|
|
text: (e.innerText||'').trim().slice(0,40),
|
|
type: e.getAttribute('type'),
|
|
aria: e.getAttribute('aria-label'),
|
|
visible: !!(e.offsetParent),
|
|
disabled: e.disabled
|
|
}));
|
|
}""")
|