def get_audiofile_metadata()

in infra-as-code/modules/ingest-pipeline/cf-ccai-conversation-upload/insights_uploader.py [0:0]

• 22 lines of code
• 2 McCabe index (conditional complexity)

    def get_audiofile_metadata(self, bucket_name, object_name):
        """
        Retrieves metadata from a Google Cloud Storage blob.

        Args:
            bucket_name (str): The bucket name.
            object_name (str): The object name.

        Returns:
            dict: The metadata extracted from the blob.

        Raises:
            Exception: If unable to retrieve metadata.
        """
        creds = self.get_client_credentials()
        storage_client = storage.Client(credentials=creds)
        bucket = storage_client.bucket(bucket_name)
        blob = bucket.get_blob(object_name)

        print("Bucket name: {}".format(bucket))
        print("Blob: {}".format(blob))

        if blob.metadata:
            metadata = dict()
            #TODO define agent ID with corresponding value in case it's needed
            metadata['qualityMetadata'] = {"agentInfo":[{"agentId": "Undefined"}]}
            metadata['agentId'] = "Undefined"
            metadata['labels'] = dict()
            if 'original_file_name' in blob.metadata: 
                metadata['labels']['original_file_name'] = blob.metadata['original_file_name']
            if 'patient_id' in blob.metadata:
                metadata['labels']['patient_phone_number'] = blob.metadata['patient_id']
            if 'categories' in blob.metadata:
                metadata['labels']['categories'] = blob.metadata['categories']

            print("Retrieved metadata from file")
            return metadata
        else: 
            raise Exception("Unable to retrieve metadata of agent")