各種AIを使って動画ファイルから議事メモを作成する２

import asyncio
from audio_extractor import AudioExtractor
from audio_splitter import AudioSplitter
from summarizer_factory import SummarizerFactory
from speech_to_text import Transcriber


class MeetingMinutesCreationService():

  # 対象動画ファイル指定
  target_file = "./data/画面収録 2024-02-28 9.05.31.mov"

  async def execute(self):
    # 音声ファイル抽出
    audio_extractor = AudioExtractor(self.target_file)
    audio_file_name = audio_extractor.extract_audio()

    # 音声ファイルの分割
    splitter = AudioSplitter()
    separated_files: list[str] = splitter.split_wav(audio_file_name)

    # 音声→文字変換
    transcriber = Transcriber()
    transcribed_files = await transcriber.perform_transcription(separated_files)

    # 文字列の要約
    summarizer = SummarizerFactory.get_instance().create_summarizer_instance()
    summarizer.summarize_text(transcribed_files)

ibm-watson
moviepy
python-dotenv
scipy

pip install -r requirements.txt

import asyncio
import datetime
import json
from env_settings import EnvSettings
from ibm_watson import SpeechToTextV1
from ibm_cloud_sdk_core.authenticators import IAMAuthenticator

class Transcriber():
  # 各設定の初期化
  def __init__(self) -> None:
    env_settings = EnvSettings()
    self.api_key = env_settings.get_value_of('WD_API_KEY') # IBM watson discovery speech to text APIKey
    self.url = env_settings.get_value_of('WD_URL') # IBM Cloud Tokyo Region
    self.wd_concurrent_cons = int( env_settings.get_value_of('WD_CONCURRENT_CONS') or "5") # WD concurrent connections Count
    self.speech_to_text = SpeechToTextV1(authenticator=IAMAuthenticator(self.api_key))
    self.speech_to_text.set_service_url(self.url)

  async def perform_transcription(self, target_files):
    semaphore = asyncio.Semaphore(self.wd_concurrent_cons)
    tasks = []
    for target_file in target_files:
      task = asyncio.create_task(self.__transcript_voice_file(target_file, semaphore))
      tasks.append(task)
    output_files = await asyncio.gather(*tasks)
    return output_files
  
  async def __transcript_voice_file(self, target_file, semaphore) -> str:
    async with semaphore:
      print(' start transcription at ' , datetime.datetime.now(), ', target = ', target_file) # 開始のゴング
      loop = asyncio.get_running_loop()
      speech_recognition_results = await loop.run_in_executor(None, self.__recognize_sync, target_file) # WDへのリクエストを非同期化
      # result全量出力
      Transcript=json.dumps(speech_recognition_results, indent=2, ensure_ascii=False)
      with open(target_file + '.Result.txt', 'w') as file_obj:
          file_obj.write(Transcript)
      # textのみ出力
      with open(target_file + '.ResultTextOnly.txt', 'w', newline='\n') as file_obj:
        for line in speech_recognition_results['results']:
          file_obj.write(line['alternatives'][0]['transcript'].replace(' ', '') + '\n')

      print(' end transcription at ' , datetime.datetime.now(),', target = ', target_file) # 終了のゴング
      return target_file + '.ResultTextOnly.txt'
    
  def __recognize_sync(self, target_file):
    with open(target_file, 'rb') as audio_file:
        return self.speech_to_text.recognize(
          audio=audio_file,
          model='ja-JP_Telephony',
          content_type='audio/wav',
        ).get_result()