AppSync Event API を使用したリアルタイム文字起こしをブラウザに表示させてみた！ #lambda

"""
Kinesis(Contact Lens Streams) → Lambda → AppSync Event API へ API キー認証でイベントを Publish する関数
環境変数
  HTTP_ENDPOINT = ".appsync-api..amazonaws.com"
  API_KEY       = "da2-xxxxxxxxxxxxxxxxxxxx"     # AWS AppSync で発行したキー
  CHANNEL       = "/default/message"             # チャンネル名
"""
import botocore.awsrequest, botocore.httpsession, json, os, base64

HTTP_ENDPOINT = os.environ["HTTP_ENDPOINT"].rstrip("/")
API_KEY       = os.environ["API_KEY"]
CHANNEL       = os.environ["CHANNEL"]

session = botocore.httpsession.URLLib3Session()

def lambda_handler(event, context):
    try:
      print(f"Received event: {event}")
      segments = []
      for record in event.get("Records", []):
        payload = record["kinesis"]["data"]
        decoded = base64.b64decode(payload)
        data = json.loads(decoded)
        print(f"Decoded data: {data}")
        if "Segments" in data:
          segments.extend(data["Segments"])
        else:
          print("No Segments found in the record.")
      transcript = correct_transcript(segments)

      events = [
          json.dumps({"role": t["role"], "content": t["content"]})
          for t in transcript
      ]
      print(f"AppSync Post Event: ", events)

      body = _http_post({"channel": CHANNEL, "events": events})
      return {"statusCode": 200, "body": body}
    except Exception as e:
      print(f"Error processing event: {e}")
      return {"statusCode": 500, "body": str(e)}

def correct_transcript(segments: list):
    corrected_segments = []
    role_mapping = {
        "AGENT": "エージェント",
        "CUSTOMER": "顧客",
    }
    for s in segments:
        if "Transcript" in s:
            transcript = s["Transcript"]
            content = transcript.get("Content", "")
            participant_role = transcript.get("ParticipantRole", "")
            role = role_mapping.get(participant_role)
            begin_offset = transcript.get("BeginOffsetMillis", 0)
            if role and content:
                corrected_segments.append({
                    "role": role,
                    "content": content,
                    "beginOffsetMillis": begin_offset
                })
    corrected_segments.sort(key=lambda x: x["beginOffsetMillis"])
    return corrected_segments

def _http_post(payload: dict) -> str:
    req = botocore.awsrequest.AWSRequest(
        method="POST",
        url=f"https://{HTTP_ENDPOINT}/event",
        data=json.dumps(payload).encode(),
        headers={
            "content-type": "application/json",
            "x-api-key": API_KEY
        }
    )
    resp = session.send(req.prepare())
    if resp.status_code >= 400:
        raise RuntimeError(f"Publish failed: {resp.status_code} {resp.text[:200]}")
    return resp.text or "OK"

 lang="ja">
  
    
リアルタイムの文字起こし
    
  









Source link 

Views: 0
共有:

				
				Facebook
			

				
				X
			

				
				LinkedIn
			

				
				Tumblr
			

				
				Pinterest
			

				
				Threads
			

				
				Bluesky
			
いいね:
いいね 読み込み中…


	
関連