| |
| """ |
| 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()) |
|
|
