Transcribe new recordings in S3

import json
import boto3
from clickruntime import on_trigger, credentials, notify, state, report_fetched, ai

REGION = "us-east-1"

def aws(service):
    return boto3.client(service, region_name=REGION, aws_access_key_id=credentials.secret("AWS_ACCESS_KEY_ID"),
                        aws_secret_access_key=credentials.secret("AWS_SECRET_ACCESS_KEY"))

BUCKET = "acme-recordings"
MEDIA = (".mp3", ".mp4", ".wav", ".m4a")

@on_trigger
def run(event):
    s3, transcribe = aws("s3"), aws("transcribe")
    seen, pending = set(state.get("seen") or []), state.get("pending") or []
    new = [o["Key"] for o in s3.list_objects_v2(Bucket=BUCKET).get("Contents", [])
           if o["Key"].lower().endswith(MEDIA) and o["Key"] not in seen]
    report_fetched(len(new))
    if not new and not pending:
        return {"new": 0}
    for key in new:
        job = "rec-" + "".join(c if c.isalnum() else "-" for c in key)[-180:]
        transcribe.start_transcription_job(TranscriptionJobName=job, Media={"MediaFileUri": f"s3://{BUCKET}/{key}"},
                                           IdentifyLanguage=True, OutputBucketName=BUCKET, OutputKey=f"transcripts/{job}.json")
        pending.append({"job": job, "key": key})
    for p in list(pending):
        status = transcribe.get_transcription_job(TranscriptionJobName=p["job"])["TranscriptionJob"]["TranscriptionJobStatus"]
        if status == "COMPLETED":
            text = json.load(s3.get_object(Bucket=BUCKET, Key=f"transcripts/{p['job']}.json")["Body"])["results"]["transcripts"][0]["transcript"]
            summary = ai.generate(text[:20000], system="Summarise this recording in 3 bullet points.", max_output_tokens=200).text
            notify.email_me(f"Transcript: {p['key']}", f"{summary}\n\n{text}")
        if status in ("COMPLETED", "FAILED"):
            pending.remove(p)
    state.set("seen", list(seen | set(new)))
    state.set("pending", pending)
    return {"started": len(new), "pending": len(pending)}