muse-chat-api.py: multi-account support (--account)
This commit is contained in:
operator
+92
-92
@@ -1,105 +1,105 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Muse.ai chat API via CDP (headless smoke profile).
|
||||
Usage:
|
||||
muse-chat-api.py send "hello" # send message to current chat
|
||||
muse-chat-api.py messages # print recent messages
|
||||
muse-chat-api.py wait [timeout] # wait for new response\n muse-chat-api.py navigate <url> # go to thread URL
|
||||
Requires: websocket-client (pip install --break-system-packages websocket-client)
|
||||
CDP relay must be up: http://10.201.87.2:9410/json/list
|
||||
"""
|
||||
import json, sys, time, urllib.request
|
||||
import websocket
|
||||
Multi-account muse.ai chat API.
|
||||
|
||||
CDP_URL = "http://10.201.87.2:9410/json/list"
|
||||
Usage:
|
||||
muse-chat-api.py --account <name> send "message"
|
||||
muse-chat-api.py --account <name> messages [n]
|
||||
muse-chat-api.py --account <name> wait [timeout]
|
||||
|
||||
def get_page():
|
||||
with urllib.request.urlopen(CDP_URL, timeout=5) as r:
|
||||
targets = json.load(r)
|
||||
pages = [t for t in targets if t.get('type')=='page' and 'muse.ai' in t.get('url','')]
|
||||
Accounts (configured on bl):
|
||||
muse-agent : CDP 9410 (smoke node) - ltd.pixels.ltd@gmail.com
|
||||
pip : CDP 9420 (phone-test node) - piparada (phone OTP)
|
||||
646 : (not currently logged in - needs OTP)
|
||||
|
||||
The account maps to a CDP endpoint. The browser must be running
|
||||
with the account logged in.
|
||||
"""
|
||||
import json, urllib.request, websocket, time, sys, argparse
|
||||
|
||||
ACCOUNTS = {
|
||||
"muse-agent": "http://127.0.0.1:9410/json/list",
|
||||
"pip": "http://127.0.0.1:9420/json/list",
|
||||
# "646": "http://127.0.0.1:9430/json/list", # TODO: set up 646 node
|
||||
}
|
||||
|
||||
def get_page(cdp_url):
|
||||
with urllib.request.urlopen(cdp_url, timeout=5) as r:
|
||||
ts = json.load(r)
|
||||
pages = [t for t in ts if t.get('type') == 'page']
|
||||
if not pages:
|
||||
raise RuntimeError("no muse.ai page found")
|
||||
print("ERROR: No page found", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
return pages[0]
|
||||
|
||||
def connect():
|
||||
page = get_page()
|
||||
ws = websocket.create_connection(page['webSocketDebuggerUrl'], timeout=10)
|
||||
return ws
|
||||
def ev(ws, expr, await_p=False):
|
||||
ws.send(json.dumps({
|
||||
"id": 1, "method": "Runtime.evaluate",
|
||||
"params": {"expression": expr, "returnByValue": True, "awaitPromise": await_p}
|
||||
}))
|
||||
resp = json.loads(ws.recv())
|
||||
return resp.get('result', {}).get('result', {}).get('value')
|
||||
|
||||
def ev(ws, expr, await_promise=False):
|
||||
ws.send(json.dumps({"id":1,"method":"Runtime.evaluate",
|
||||
"params":{"expression":expr,"returnByValue":True,"awaitPromise":await_promise}}))
|
||||
r = json.loads(ws.recv())
|
||||
return r.get('result',{}).get('result',{}).get('value')
|
||||
def cmd_send(ws, message):
|
||||
# Escape for JS
|
||||
msg_esc = message.replace('\\', '\\\\').replace('`', '\\`').replace('$', '\\$')
|
||||
result = ev(ws, f"""(async()=>{{
|
||||
const input = document.querySelector('[contenteditable="true"]') ||
|
||||
document.querySelector('textarea[placeholder*="Message"]') ||
|
||||
[...document.querySelectorAll('div[role="textbox"]')][0];
|
||||
if (!input) return 'NOINPUT';
|
||||
input.focus();
|
||||
document.execCommand('insertText', false, `{msg_esc}`);
|
||||
await new Promise(r=>setTimeout(r,500));
|
||||
const send = [...document.querySelectorAll('button')].find(b=>
|
||||
b.getAttribute('aria-label')&&b.getAttribute('aria-label').toLowerCase().includes('send')
|
||||
);
|
||||
if (send) {{ send.click(); return 'sent'; }}
|
||||
const ke = new KeyboardEvent('keydown', {{key:'Enter', code:'Enter', bubbles:true}});
|
||||
input.dispatchEvent(ke);
|
||||
return 'enter-sent';
|
||||
}})()""", True)
|
||||
print(result)
|
||||
|
||||
def cmd_messages(ws, n=5):
|
||||
result = ev(ws, f"""(() => {{
|
||||
const ps = [...document.querySelectorAll('p')].slice(-{n*2}).map(p=>p.innerText.slice(0,200));
|
||||
return ps.join('\\n---\\n');
|
||||
}})()""")
|
||||
print(result)
|
||||
|
||||
def cmd_wait(ws, timeout=30):
|
||||
print(f"Waiting {timeout}s for response...")
|
||||
# Simple: wait and then get messages
|
||||
time.sleep(timeout)
|
||||
cmd_messages(ws, 2)
|
||||
|
||||
def main():
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument('--account', required=True, choices=list(ACCOUNTS.keys()),
|
||||
help='Account to use')
|
||||
p.add_argument('command', choices=['send', 'messages', 'wait'])
|
||||
p.add_argument('arg', nargs='?', default=None)
|
||||
args = p.parse_args()
|
||||
|
||||
cdp_url = ACCOUNTS[args.account]
|
||||
page = get_page(cdp_url)
|
||||
ws = websocket.create_connection(page['webSocketDebuggerUrl'], timeout=15)
|
||||
|
||||
def send_message(text):
|
||||
ws = connect()
|
||||
try:
|
||||
result = ev(ws, f"""
|
||||
(async () => {{
|
||||
const ta = document.querySelector('textarea[aria-label="Message"]');
|
||||
if (!ta) return 'ERROR: no composer';
|
||||
ta.focus();
|
||||
document.execCommand('insertText', false, {json.dumps(text)});
|
||||
await new Promise(r => setTimeout(r, 200));
|
||||
ta.dispatchEvent(new KeyboardEvent('keydown', {{key:'Enter', code:'Enter', keyCode:13, bubbles:true}}));
|
||||
return 'sent';
|
||||
}})()
|
||||
""", await_promise=True)
|
||||
return result
|
||||
if args.command == 'send':
|
||||
if not args.arg:
|
||||
print("ERROR: send requires a message", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
cmd_send(ws, args.arg)
|
||||
elif args.command == 'messages':
|
||||
n = int(args.arg) if args.arg else 5
|
||||
cmd_messages(ws, n)
|
||||
elif args.command == 'wait':
|
||||
t = int(args.arg) if args.arg else 30
|
||||
cmd_wait(ws, t)
|
||||
finally:
|
||||
ws.close()
|
||||
|
||||
def get_messages(limit=10):
|
||||
ws = connect()
|
||||
try:
|
||||
return ev(ws, f"""
|
||||
Array.from(document.querySelectorAll('p')).slice(-{limit}).map(e=>e.innerText).join('\\n---\\n')
|
||||
""")
|
||||
finally:
|
||||
ws.close()
|
||||
|
||||
def navigate(url):
|
||||
"""Navigate to a URL (e.g., a thread: https://muse.ai/thread/<id>)."""
|
||||
ws = connect()
|
||||
try:
|
||||
ws.send(json.dumps({"id":1,"method":"Page.navigate","params":{"url":url}}))
|
||||
r = json.loads(ws.recv())
|
||||
return r.get('result',{}).get('frameId','navigated')
|
||||
finally:
|
||||
ws.close()
|
||||
|
||||
def wait_for_response(timeout=60, poll=2):
|
||||
"""Wait for a new assistant message after sending."""
|
||||
ws = connect()
|
||||
try:
|
||||
before = ev(ws, "document.body.innerText.length")
|
||||
start = time.time()
|
||||
while time.time() - start < timeout:
|
||||
time.sleep(poll)
|
||||
# Check if there's a "stop" button (indicates generating)
|
||||
generating = ev(ws, """
|
||||
!!document.querySelector('button[aria-label*="Stop"], button[aria-label*="stop"]')
|
||||
""")
|
||||
if not generating:
|
||||
# Give it a moment to finish rendering
|
||||
time.sleep(2)
|
||||
return get_messages(5)
|
||||
return "TIMEOUT waiting for response"
|
||||
finally:
|
||||
ws.close()
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2:
|
||||
print(__doc__)
|
||||
sys.exit(1)
|
||||
cmd = sys.argv[1]
|
||||
if cmd == "send" and len(sys.argv) > 2:
|
||||
print(send_message(sys.argv[2]))
|
||||
elif cmd == "messages":
|
||||
print(get_messages(int(sys.argv[2]) if len(sys.argv)>2 else 10))
|
||||
elif cmd == "navigate" and len(sys.argv) > 2:
|
||||
print(navigate(sys.argv[2]))
|
||||
elif cmd == "wait":
|
||||
print(wait_for_response(int(sys.argv[2]) if len(sys.argv)>2 else 60))
|
||||
else:
|
||||
print("unknown command")
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
||||
Reference in New Issue
Block a user