Google Cloud Speech-to-Text APIを用いた文字起こし（その２　というか後編）

Google Cloud Speech-to-Text API とGoogle Cloud Strageの組み合わせ

久しぶりの記事の更新になるなぁ、、、

Google Cloud Strageに音声をアップロードしてみたけど、正直イマイチ。　英語だったらもっと良いのかもしれないけど、日本語への変換に関しては正直な所、Notta等のほかのサービスに劣る気がします。　

なので、そこまでがっつり記事を書くつもりはないです。

せっかくなので、動画から一気通貫でテキストを抽出するコードを下に載せていくよ。

データをCloudストレージに置いたままだと料金がかかり続けるので、アップロード後　すぐ消しています。

import subprocess
import os
from google.cloud import storage, speech
import io

# Google Cloud認証ファイルの設定
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = r".\cellular-axon-432906-*******.json"  # ここで認証情報を設定　あなたの設定ファイルのファイル名を入れてください

# Step 1: MP4ファイルからWAVファイルを抽出する関数
def extract_audio_from_mp4(mp4_file_path, wav_file_path):
    """Extracts audio from an MP4 file and saves it as a WAV file."""
    try:
        # subprocessを使用してffmpegコマンドを実行
        subprocess.run(
            ["ffmpeg", "-i", mp4_file_path, "-q:a", "0", "-map", "a", wav_file_path],
            check=True
        )
        print(f"Audio extracted and saved to {wav_file_path}")
    except subprocess.CalledProcessError as e:
        print(f"An error occurred: {e}")

# Step 2: 抽出されたWAVファイルをモノラルに変換する関数
def convert_to_mono(input_file, output_file):
    """Converts a stereo audio file to mono using ffmpeg."""
    try:
        # subprocessを使用してモノラルに変換
        subprocess.run(
            ["ffmpeg", "-i", input_file, "-ac", "1", output_file],
            check=True
        )
        print(f"Converted {input_file} to mono and saved as {output_file}")
    except subprocess.CalledProcessError as e:
        print(f"An error occurred: {e}")

# Step 3: Google Cloud Storageにファイルをアップロードする関数
def upload_to_gcs(bucket_name, source_file_name, destination_blob_name):
    """Uploads a file to Google Cloud Storage."""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)
    blob.upload_from_filename(source_file_name)
    
    gcs_uri = f"gs://{bucket_name}/{destination_blob_name}"
    print(f"File {source_file_name} uploaded to {gcs_uri}.")
    return gcs_uri

# Step 4: Google Cloud Speech-to-Textを使用してGCS上の音声を文字起こしする関数
def transcribe_gcs(gcs_uri):
    """Transcribes the audio file stored in Google Cloud Storage using Google Cloud Speech-to-Text."""
    client = speech.SpeechClient()
    audio = speech.RecognitionAudio(uri=gcs_uri)

    config = speech.RecognitionConfig(
        encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
        sample_rate_hertz=44100,
        language_code="ja-JP"  # 日本語での文字起こしを指定
    )

    operation = client.long_running_recognize(config=config, audio=audio)
    print("Waiting for operation to complete...")
    response = operation.result(timeout=90000)  # 長時間処理がかかることを想定して90秒のタイムアウトを設定

    transcription = ""
    for result in response.results:
        transcription += result.alternatives[0].transcript + "\n"
    
    return transcription

# Step 5: ローカルに文字起こし結果を保存する関数
def save_transcription_to_file(transcription, file_name):
    """Saves the transcription to a local text file."""
    with open(file_name, "w", encoding="utf-8") as file:
        file.write(transcription)
    print(f"Transcription saved to {file_name}.")

# Step 6: GCSからアップロードしたファイルを削除する関数
def delete_from_gcs(bucket_name, blob_name):
    """Deletes a file from Google Cloud Storage."""
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(blob_name)
    
    blob.delete()
    print(f"File gs://{bucket_name}/{blob_name} deleted from GCS.")

# 一連の音声ファイル処理を行う関数
def process_audio_file(wav_file_path, bucket_name, output_text_file):
    """Uploads a WAV file to GCS, transcribes it, saves the transcription, and deletes the GCS file."""
    blob_name = os.path.basename(wav_file_path)
    
    # Step 1: GCSにWAVファイルをアップロード
    gcs_uri = upload_to_gcs(bucket_name, wav_file_path, blob_name)
    
    # Step 2: GCSにアップロードされた音声を文字起こし
    transcription = transcribe_gcs(gcs_uri)
    
    # Step 3: 文字起こし結果をローカルファイルに保存
    save_transcription_to_file(transcription, output_text_file)
    
    # Step 4: GCSからアップロードした音声ファイルを削除
    delete_from_gcs(bucket_name, blob_name)

# 実行例
mp4_file_path = "your_video.mp4"  # 元のMP4ファイル
wav_file_path = "output_audio.wav"  # 抽出後のWAVファイル
mono_wav_file_path = "output_audio_mono.wav"  # モノラル変換後のWAVファイル
bucket_name = "bucket1_mojiokoshi1"  # GCSのバケット名
output_text_file = "transcription.txt"  # ローカルに保存する文字起こし結果のファイル名

# 音声抽出、モノラル変換、GCSアップロード、文字起こし、削除の一連の処理を実行
extract_audio_from_mp4(mp4_file_path, wav_file_path)  # MP4からWAVに抽出
convert_to_mono(wav_file_path, mono_wav_file_path)  # WAVをモノラルに変換
process_audio_file(mono_wav_file_path, bucket_name, output_text_file)  # 文字起こしプロセスを実行

説明:

extract_audio_from_mp4
この関数では、MP4ファイルから音声をWAV形式で抽出します。使っているのはffmpegというツールで、動画から音声トラックだけを取り出す感じです。動画ファイルに関しては、例えばZoomの録画データなどが想定されます。

convert_to_mono
抽出したWAVファイルを、今度はモノラルに変換します。これもffmpegを使っていて、音声のチャンネルを1（モノラル）に変更します。モノラルにする理由は、音声データを軽くしたり、文字起こしの精度を上げたりするためです。

process_audio_file
この関数では、WAVファイルをGoogle Cloud Storage（GCS）にアップロードして、GoogleのSpeech-to-Textを使って音声を文字に変換します。できたテキストはローカルファイルに保存して、その後、GCSからファイルを削除するという一連の流れを自動化しています。GoogleのAPIを使うことで、精度の高い文字起こしができるのがポイントです。

Google Cloud Speech-to-Text API とGoogle Cloud Strageの組み合わせ

コメント