def batch_translate_text()

in projects/speech2srt/translate_txt.py [0:0]

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

def batch_translate_text(input_uri, output_uri, project_id,
                         location, source_lang, target_lang, glossary_id=None):
    # call batch translate against orig.txt

    client = translate.TranslationServiceClient()

    target_language_codes = target_lang.split(",")
    gcs_source = {"input_uri": input_uri}
    mime_type = "text/plain"
    input_configs_element = {"gcs_source": gcs_source, "mime_type": mime_type}
    input_configs = [input_configs_element]
    gcs_destination = {"output_uri_prefix": output_uri}
    output_config = {"gcs_destination": gcs_destination}
    parent = f"projects/{project_id}/locations/{location}"

    # Add glossary_config if glossary_id is provided
    if glossary_id:
        print(f"Using glossary: {glossary_id}")
        glossary_resource_name = client.glossary_path(project_id, location,
                                                      glossary_id)

        # Create TranslateTextGlossaryConfig using the glossary resource name
        translate_text_glossary_config = translate.TranslateTextGlossaryConfig(
            glossary=glossary_resource_name,
        )
        # Modify the request to include translate_text_glossary_config
        operation = client.batch_translate_text(
            request={
                "parent": parent,
                "source_language_code": source_lang,
                "target_language_codes": target_language_codes,
                "input_configs": input_configs,
                "output_config": output_config,
                "glossaries": {target_lang: translate_text_glossary_config}
            }
        )
    else:
        operation = client.batch_translate_text(
            request={
                "parent": parent,
                "source_language_code": source_lang,
                "target_language_codes": target_language_codes,
                "input_configs": input_configs,
                "output_config": output_config,
            })

    # Initial delay
    total_wait_secs = 90
    print(f"Waiting for operation to complete... {total_wait_secs:.0f} secs")

    delay_secs = 10
    sleep(90)
    while not operation.done():
        # Exponential backoff
        delay_secs *= 1.1
        total_wait_secs += delay_secs
        print(f"Checking in: {delay_secs:.0f}s | total: {total_wait_secs:.0f}s")
        sleep(delay_secs)

    response = operation.result()
    print("Total Characters: {}".format(response.total_characters))
    print("Translated Characters: {}".format(response.translated_characters))