File size: 2,603 Bytes
c91aa86 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | #!/usr/bin/env python3
"""
Example: after local Whisper (or any STT) produces text, POST it to Nexus for summarization.
Usage:
set NEXUS_URL=https://malek391-nexus-ai.hf.space
set NEXUS_INGEST_SECRET=your_secret_if_configured
python tools/openclaw_post_transcript.py path/to/transcript.txt --meeting-id demo1
Or pipe:
type transcript.txt | python tools/openclaw_post_transcript.py -
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import urllib.error
import urllib.request
def main() -> int:
p = argparse.ArgumentParser(description="POST transcript text to Nexus /api/integrations/openclaw-transcript")
p.add_argument("file", help="Path to UTF-8 text file, or - for stdin")
p.add_argument("--title", default="", help="Meeting title")
p.add_argument("--meeting-id", default="", dest="meeting_id")
p.add_argument("--meeting-url", default="", dest="meeting_url")
p.add_argument("--source", default="openclaw_whisper_cli")
args = p.parse_args()
base = (os.environ.get("NEXUS_URL") or "").rstrip("/")
if not base:
print("Set NEXUS_URL to your Nexus origin, e.g. https://malek391-nexus-ai.hf.space", file=sys.stderr)
return 2
if args.file == "-":
transcript = sys.stdin.read()
else:
with open(args.file, encoding="utf-8", errors="replace") as f:
transcript = f.read()
url = f"{base}/api/integrations/openclaw-transcript"
body = {
"transcript": transcript,
"title": args.title or "Meeting transcript",
"meeting_id": args.meeting_id,
"meeting_url": args.meeting_url,
"source": args.source,
}
data = json.dumps(body).encode("utf-8")
headers = {"Content-Type": "application/json"}
secret = (os.environ.get("NEXUS_INGEST_SECRET") or os.environ.get("NEXUS_OPENCLAW_INGEST_SECRET") or "").strip()
if secret:
headers["X-Nexus-Ingest-Secret"] = secret
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
try:
with urllib.request.urlopen(req, timeout=120) as resp:
out = resp.read().decode("utf-8", errors="replace")
except urllib.error.HTTPError as e:
print(e.read().decode("utf-8", errors="replace"), file=sys.stderr)
return 1
print(out)
try:
j = json.loads(out)
jid = j.get("job_id")
if jid:
print(f"\nPoll: {base}/api/integrations/openclaw/jobs/{jid}", file=sys.stderr)
except Exception:
pass
return 0
if __name__ == "__main__":
raise SystemExit(main())
|