Google Speech-to-Textの直接GCSアップロード機能をPythonで動かしてみた


# 背景
- 先日、Google SpeechToTextの書き起こし結果の直接GCSアップロード機能がリリースされた

https://cloud.google.com/speech-to-text/docs/release-notes#March_23_2021

https://cloud.google.com/speech-to-text/docs/async-recognize#uploading_your_transcription_results_to_a_gcs_bucket

- これまで容量の大きな音声ファイルの書き起こしはlongRunning + polling して書き起こしを取得する必要があった
- この機能追加により、pollingが必要なくなり直接GCSに結果がuploadされるようになった
- python-libraryの2.2.0にこのリリース対応がされていたが、[httpリクエストではうまく動かなかった](https://qiita.com/munaita_/items/727e1ee5ffcfc4fc60ee#%E5%A4%B1%E6%95%97%E3%81%97%E3%81%9Fpython-library%E3%81%AE%E5%AE%9F%E8%A3%85%E3%81%A8%E3%82%A8%E3%83%A9%E3%83%BC)のでREST APIで実装した


# 実装

- 実装は [Uploading your transcription results to a Cloud Storage bucket](https://cloud.google.com/speech-to-text/docs/async-recognize#uploading_your_transcription_results_to_a_bucket) を参考に実装
- 認証は [Google Cloud APIのAccess TokenをPythonで取得する(gcloud auth application-default print-access-tokenのやつ)](https://qiita.com/munaita_/items/29ddf8dd27d957673a7c) を参考

```py
import google.auth
import google.auth.transport.requests
import requests
import json


if __name__ == '__main__':
    # Access Token取得
    credentials, your_project_id = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
    auth_req = google.auth.transport.requests.Request()
    credentials.refresh(auth_req)

    url = 'https://speech.googleapis.com/v1p1beta1/speech:longrunningrecognize'
    payload = {
      'config': {
         'language_code': 'ja-JP',
         'enable_word_time_offsets': 'True'
      },
      'output_config': {
         'gcs_uri':'gs://YOUR_BUCKET/PATH_TO_OUTPUT/direct.json'
      },
      'audio': {
        'uri': 'gs://YOUR_BUCKET/PATH_TO_AUDIO_FILE/audio.wav'
      }
    }

    headers = {'content-type': 'application/json', 'Authorization': f'Bearer {credentials.token}'}
    r = requests.post(url, data=json.dumps(payload), headers=headers)
```

これで問題なく動いた。
以下に一応動かなかった実装とエラーコードを貼っておく。

# 失敗したpython-libraryの実装とエラー


#### 実装

- google-cloud-speech のバージョンは `2.2.1`
- python 3.9

```py:main.py
from google.cloud import speech_v1p1beta1 as speech

client = speech.SpeechClient()

audio = speech.RecognitionAudio(uri='gs://YOUR_BUCKET/PATH_TO_AUDIO/audio.wav')
config = speech.RecognitionConfig(
        language_code="ja-JP",
        enable_word_time_offsets=True
        )
output = speech.TranscriptOutputConfig(
        gcs_uri="gs://YOUR_BUCKET/PATH_TO_OUTPUT/direct.json"
        )
operation = client.long_running_recognize(config=config, audio=audio, output_config=output)
print("Waiting for operation to complete...")
response = operation.result(timeout=1000)
for result in response.results:
    print(u"Transcript: {}".format(result.alternatives[0].transcript))
    print("Confidence: {}".format(result.alternatives[0].confidence))
```

#### エラーコード抜粋

```
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
	status = StatusCode.INTERNAL
	details = "Internal error encountered. Please try again."
	debug_error_string = "{"created":"@1616664073.940171000","description":"Error received from peer ipv6:[xxx:xxx:xxx:xxx]:443","file":"src/core/lib/surface/call.cc","file_line":1068,"grpc_message":"Internal error encountered. Please try again.","grpc_status":13}"
```