def transcript_word_fix()

in infra-as-code/modules/ingest-pipeline/cf-transcript-correction/lib.py [0:0]

• 34 lines of code
• 6 McCabe index (conditional complexity)

  def transcript_word_fix(self):
    """Initializes the VertexAI client to start transcript word fix

    Raises:
        e: Could fail when loading the json string or response does not include valid items
    """
    print('Starting Vertex Client')
    creds = self.get_credentials()
    vertexai.init(project=self.project_id, location=self.location_id, credentials=creds)

    model = GenerativeModel(self.model_name)

    prompt = self.prompt

    audio_file_uri = self.get_gcs_uri(self.formatted_audio_bucket_id, self.formatted_audio_file_name)
    audio_file = Part.from_uri(audio_file_uri, mime_type="audio/flac")

    contents = [audio_file, prompt]
    
    token_number = model.count_tokens(prompt)

    if token_number.total_tokens > 8193:
      print('Too many tokens')
      return 

    print('Calling API')
    response = model.generate_content(contents,
                                      generation_config=GenerationConfig(
                                        temperature=0,
                                        response_mime_type="application/json",
                                        response_schema=self.response_schema
                                        )
                                      )
   
    fixed_transcript = rf'{response.text}'
    print(fixed_transcript)

    try:
      fixed_transcript = json.loads(fixed_transcript)
      if response.text != "I am not able to answer this question" or response.candidate.finish_reason != "MAX_TOKENS":
        if len(self.transcript) == len(fixed_transcript):
          self.update_transcript(self.original_transcript, fixed_transcript)
          self.upload_json_to_gcs(self.transcript_bucket_id, self.transcript_file_name, self.gemini_transcript)
      else: 
        print('Will use original transcript')
        return
    except Exception as e:
        raise e