337 lines
12 KiB
Python
337 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Minimal Fireflies.ai GraphQL client for the fireflies-relanguage skill.
|
|
|
|
Handles the four network steps of the workflow: pull a transcript you can
|
|
view (owned or shared) into the Markdown shape transcript-speaker-fill
|
|
expects, kick off a re-upload of its audio under the correct language,
|
|
poll until that re-upload shows up as a finished transcript, and delete
|
|
the resulting duplicate meeting once you're done with it.
|
|
|
|
No third-party dependencies - stdlib only (urllib), so this runs with a
|
|
bare `python3` on any machine that has the transcript-speaker-fill skill
|
|
installed.
|
|
|
|
Requires FIREFLIES_API_KEY in the environment (Settings > API in the
|
|
Fireflies web app to generate one).
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
API_URL = "https://api.fireflies.ai/graphql"
|
|
GENERIC_RE = re.compile(r'^Speaker\s+\d+$')
|
|
|
|
|
|
def parse_transcript_ref(ref):
|
|
"""Accept a bare id, or a shared link like
|
|
https://app.fireflies.ai/view/Some-Title::abcDEF123?channelSource=mine-shared
|
|
(query string / fragment and the ::title part are both optional) and
|
|
return just the transcript id."""
|
|
ref = ref.strip()
|
|
if '://' in ref:
|
|
ref = ref.split('?', 1)[0].split('#', 1)[0]
|
|
ref = ref.rstrip('/').rsplit('/', 1)[-1]
|
|
if '::' in ref:
|
|
ref = ref.rsplit('::', 1)[-1]
|
|
return ref
|
|
|
|
|
|
def gql(api_key, query, variables=None):
|
|
body = json.dumps({"query": query, "variables": variables or {}}).encode('utf-8')
|
|
req = urllib.request.Request(
|
|
API_URL,
|
|
data=body,
|
|
method='POST',
|
|
headers={
|
|
'Content-Type': 'application/json',
|
|
'Authorization': f'Bearer {api_key}',
|
|
},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=60) as resp:
|
|
payload = json.loads(resp.read().decode('utf-8'))
|
|
except urllib.error.HTTPError as e:
|
|
raw = e.read().decode('utf-8', errors='replace')
|
|
sys.exit(f"HTTP {e.code} from Fireflies API: {raw}")
|
|
|
|
if payload.get('errors'):
|
|
msgs = []
|
|
for err in payload['errors']:
|
|
ext = err.get('extensions') or {}
|
|
code = ext.get('code', '')
|
|
msg = err.get('message', '')
|
|
extra = f" (code={code}" + (f", retryAfter={ext['retryAfter']}" if 'retryAfter' in ext else '') + ')' if code else ''
|
|
msgs.append(f"{msg}{extra}")
|
|
sys.exit("Fireflies API returned error(s): " + " | ".join(msgs))
|
|
return payload.get('data') or {}
|
|
|
|
|
|
def require_api_key():
|
|
key = os.environ.get('FIREFLIES_API_KEY')
|
|
if not key:
|
|
sys.exit(
|
|
"FIREFLIES_API_KEY is not set. Generate a key in the Fireflies web "
|
|
"app under Settings > API, then export it, e.g.:\n"
|
|
" export FIREFLIES_API_KEY=your-key-here"
|
|
)
|
|
return key
|
|
|
|
|
|
def format_ts(seconds):
|
|
seconds = float(seconds)
|
|
total = int(round(seconds))
|
|
h, rem = divmod(total, 3600)
|
|
m, s = divmod(rem, 60)
|
|
if h:
|
|
return f"{h}:{m:02d}:{s:02d}"
|
|
return f"{m}:{s:02d}"
|
|
|
|
|
|
TRANSCRIPT_FIELDS = """
|
|
id
|
|
title
|
|
dateString
|
|
date
|
|
duration
|
|
audio_url
|
|
video_url
|
|
sentences {
|
|
speaker_name
|
|
speaker_id
|
|
start_time
|
|
end_time
|
|
text
|
|
raw_text
|
|
}
|
|
"""
|
|
|
|
|
|
def cmd_fetch(args):
|
|
api_key = require_api_key()
|
|
tid = parse_transcript_ref(args.ref)
|
|
data = gql(
|
|
api_key,
|
|
f"query Transcript($id: String!) {{ transcript(id: $id) {{ {TRANSCRIPT_FIELDS} }} }}",
|
|
{"id": tid},
|
|
)
|
|
t = data.get('transcript')
|
|
if not t:
|
|
sys.exit(
|
|
f"No transcript returned for id '{tid}'. Either the id is wrong, or this "
|
|
f"API key's account doesn't have access to it (not shared with you, or "
|
|
f"the workspace/plan doesn't expose it via API)."
|
|
)
|
|
|
|
sentences = t.get('sentences') or []
|
|
if not sentences:
|
|
sys.exit(
|
|
f"Transcript '{t.get('title')}' ({tid}) has no sentences yet - it may "
|
|
f"still be processing, or your account tier doesn't return transcript "
|
|
f"content via the API for this meeting."
|
|
)
|
|
|
|
out_path = Path(args.out)
|
|
lines = []
|
|
for s in sentences:
|
|
speaker = s.get('speaker_name') or f"Speaker {s.get('speaker_id', '?')}"
|
|
text = s.get('text') or s.get('raw_text') or ''
|
|
ts = format_ts(s.get('start_time', 0))
|
|
lines.append(f"**{speaker}** *[{ts}]*: {text}")
|
|
out_path.write_text("\n".join(lines) + "\n", encoding='utf-8')
|
|
|
|
meta = {
|
|
'id': t['id'],
|
|
'title': t.get('title'),
|
|
'dateString': t.get('dateString'),
|
|
'date': t.get('date'),
|
|
'duration': t.get('duration'),
|
|
'audio_url': t.get('audio_url'),
|
|
'video_url': t.get('video_url'),
|
|
'sentence_count': len(sentences),
|
|
}
|
|
meta_path = out_path.with_suffix(out_path.suffix + '.meta.json')
|
|
meta_path.write_text(json.dumps(meta, indent=2), encoding='utf-8')
|
|
|
|
print(f"Wrote {len(lines)} lines to {out_path}")
|
|
print(f"Wrote metadata to {meta_path}")
|
|
print(f"Title: {meta['title']}")
|
|
print(f"audio_url present: {bool(meta['audio_url'])}")
|
|
print(f"video_url present: {bool(meta['video_url'])}")
|
|
if not meta['audio_url']:
|
|
print(
|
|
"\nWARNING: audio_url is null. This is the expected failure mode when the "
|
|
"querying account isn't on a Fireflies Pro+ seat, or when audio access "
|
|
"isn't granted to a merely-shared (non-owned) meeting. Re-uploading this "
|
|
"meeting's audio via the API is NOT possible until this is resolved - "
|
|
"either use an API key belonging to a Pro+ seat, ask the meeting owner to "
|
|
"reprocess it directly, or manually download the audio via the Fireflies "
|
|
"web UI (if the share settings allow it) and host it at a public HTTPS URL "
|
|
"yourself before using the `upload` command's --audio-url override.",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
|
|
def cmd_upload(args):
|
|
api_key = require_api_key()
|
|
meta = json.loads(Path(args.meta).read_text(encoding='utf-8'))
|
|
|
|
audio_url = args.audio_url or meta.get('audio_url')
|
|
if not audio_url:
|
|
sys.exit(
|
|
"No audio_url available (neither in the meta file nor via --audio-url). "
|
|
"See the WARNING printed by the `fetch` command for why, and how to work "
|
|
"around it."
|
|
)
|
|
|
|
original_id = meta['id']
|
|
title = args.title or f"{meta.get('title', 'Meeting')} [relang-{original_id}]"
|
|
|
|
variables = {
|
|
"input": {
|
|
"url": audio_url,
|
|
"title": title,
|
|
"custom_language": args.language,
|
|
"client_reference_id": original_id,
|
|
}
|
|
}
|
|
if args.bypass_size_check:
|
|
variables["input"]["bypass_size_check"] = True
|
|
|
|
data = gql(
|
|
api_key,
|
|
"""
|
|
mutation UploadAudio($input: AudioUploadInput!) {
|
|
uploadAudio(input: $input) {
|
|
success
|
|
title
|
|
message
|
|
}
|
|
}
|
|
""",
|
|
variables,
|
|
)
|
|
result = data.get('uploadAudio') or {}
|
|
echoed_title = result.get('title') or title
|
|
print(json.dumps({
|
|
"submitted_title": title,
|
|
"language": args.language,
|
|
"source_transcript_id": original_id,
|
|
"api_response": result,
|
|
}, indent=2))
|
|
if not result.get('success'):
|
|
sys.exit("uploadAudio did not report success - check the message above.")
|
|
if echoed_title != title:
|
|
print(
|
|
f"\nNOTE: Fireflies echoed back a different title than submitted "
|
|
f"(likely stripped/altered some character) - use the ECHOED one below "
|
|
f"for `wait`, not what you submitted."
|
|
)
|
|
print(
|
|
f"\nQueued. Use the `wait` command with --title '{echoed_title}' to find the "
|
|
f"new transcript once processing finishes (this can take several minutes for "
|
|
f"a long recording)."
|
|
)
|
|
|
|
|
|
def cmd_wait(args):
|
|
api_key = require_api_key()
|
|
deadline = time.time() + args.max_wait
|
|
attempt = 0
|
|
while True:
|
|
attempt += 1
|
|
data = gql(
|
|
api_key,
|
|
"""
|
|
query Transcripts($limit: Int) {
|
|
transcripts(mine: true, limit: $limit) {
|
|
id
|
|
title
|
|
dateString
|
|
sentences { speaker_name }
|
|
}
|
|
}
|
|
""",
|
|
{"limit": args.list_limit},
|
|
)
|
|
candidates = [t for t in (data.get('transcripts') or []) if t.get('title') == args.title]
|
|
ready = [t for t in candidates if t.get('sentences')]
|
|
if ready:
|
|
t = ready[0]
|
|
print(json.dumps({"id": t['id'], "title": t['title'], "dateString": t.get('dateString'), "ready": True}, indent=2))
|
|
return
|
|
if candidates:
|
|
print(f"[attempt {attempt}] Found the meeting but it's still processing (no sentences yet)...", file=sys.stderr)
|
|
else:
|
|
print(f"[attempt {attempt}] Not found yet...", file=sys.stderr)
|
|
if time.time() >= deadline:
|
|
sys.exit(
|
|
f"Gave up after {args.max_wait}s without finding a ready transcript "
|
|
f"titled '{args.title}'. Long recordings can take longer than that to "
|
|
f"process - re-run `wait` again with a fresh --max-wait, or check the "
|
|
f"Fireflies web UI directly for a meeting with that title."
|
|
)
|
|
time.sleep(args.interval)
|
|
|
|
|
|
def cmd_delete(args):
|
|
if not args.yes:
|
|
sys.exit("Refusing to delete without --yes (this is irreversible).")
|
|
api_key = require_api_key()
|
|
data = gql(
|
|
api_key,
|
|
"""
|
|
mutation DeleteTranscript($id: String!) {
|
|
deleteTranscript(id: $id) {
|
|
id
|
|
title
|
|
date
|
|
duration
|
|
}
|
|
}
|
|
""",
|
|
{"id": args.id},
|
|
)
|
|
print(json.dumps(data.get('deleteTranscript') or {}, indent=2))
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
sub = ap.add_subparsers(dest='command', required=True)
|
|
|
|
p = sub.add_parser('fetch', help="Fetch a transcript (owned or shared) into transcript-speaker-fill-compatible Markdown + a metadata sidecar.")
|
|
p.add_argument('--ref', required=True, help='Transcript id, or a Fireflies share URL/link')
|
|
p.add_argument('--out', required=True, help='Output .md path')
|
|
p.set_defaults(func=cmd_fetch)
|
|
|
|
p = sub.add_parser('upload', help="Re-upload a transcript's audio under a different language via uploadAudio.")
|
|
p.add_argument('--meta', required=True, help="Path to the .meta.json produced by `fetch` for the ORIGINAL (broken-language) transcript")
|
|
p.add_argument('--language', required=True, help='Target language code, e.g. "en", "pl", "es"')
|
|
p.add_argument('--title', help='Override the title used for the new upload (default: original title + a [relang-<id>] tag)')
|
|
p.add_argument('--audio-url', help='Override the audio URL instead of using the one from --meta (e.g. a self-hosted fallback URL)')
|
|
p.add_argument('--bypass-size-check', action='store_true')
|
|
p.set_defaults(func=cmd_upload)
|
|
|
|
p = sub.add_parser('wait', help='Poll until the re-uploaded transcript shows up as fully processed.')
|
|
p.add_argument('--title', required=True, help='Exact title echoed back by `upload` (not necessarily what you submitted - Fireflies can alter it, e.g. stripping colons)')
|
|
p.add_argument('--interval', type=int, default=30, help='Seconds between polls (default 30)')
|
|
p.add_argument('--max-wait', type=int, default=900, help='Give up after this many seconds (default 900 = 15 min)')
|
|
p.add_argument('--list-limit', type=int, default=20, help="How many of your most recent transcripts to scan each poll for an exact title match (default 20; server 'keyword' search was found unreliable against bracket-tagged titles, so this lists recent transcripts client-side instead of filtering server-side)")
|
|
p.set_defaults(func=cmd_wait)
|
|
|
|
p = sub.add_parser('delete', help='Delete a transcript by id (irreversible).')
|
|
p.add_argument('--id', required=True)
|
|
p.add_argument('--yes', action='store_true', help='Required confirmation flag')
|
|
p.set_defaults(func=cmd_delete)
|
|
|
|
args = ap.parse_args()
|
|
args.func(args)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|