def run_format()

in infra-as-code/modules/audio-data-format-change/function-source-code/lib.py [0:0]

• 24 lines of code
• 4 McCabe index (conditional complexity)

  def run_format(self):
    """Runs the format change when the format change is successful 
       it sends the correct payload. 

       Calls verify_file from RecordKeeper to: 
        Verify that filename has a case manager otherwise it will not process and log a warning
        Verify if file was already processed it will log a warning
        Modify the parquet file to keep track of the pipeline step status
    """
    filename, trigger_file = self.download_from_gcs()

    self.record_keeper = RecordKeeper(self.ingest_record_bucket_id, self.metadata['original_file_name'], self.storage_client)
    
    try:
      self.record_keeper.verify_file()
      print(f'New assigned filename: {filename}')
      if(self.format_audio(filename, trigger_file)):
        #Successfully changed format
        print('Changed format')
        self.upload_resources(filename)
      else: 
        #Unsuccessful format change, write on error and delete from processed
        print('Unsucessful')
        self.record_keeper.replace_row(
          self.record_keeper.create_error_record(
            f'An error ocurred while changing audio format: {self.get_ffmpeg_error()}')) 
        self.log_error()
    except Exception as e:
      match str(e):
        # case 'Repeated file with no case manager email':
        #   self.log_error(str(e), "WARNING")
        # case 'File is processing or was already processed':
        #   self.log_error(str(e), "WARNING")
        case _:
          self.record_keeper.replace_row(
            self.record_keeper.create_error_record(
              f'An error ocurred while changing audio format: {str(e)}')) 
          self.log_error(str(e))

    self.delete_tmp_files()
    print("Finished")