### 질문
aws boto를 활용해서, 파이썬으로 다음과 같은 프로그램을 만들고 싶어.
1. 음성 메모 파일을 정해진 폴더에 넣으면 자동으로 s3의 audiotransjake 버킷으로 업로드 된다.
2. aws transcribe가 s3 audiotranjake 버킷에 업로드 된 새로운 음성 파일을 한국어 텍스트로 변환한다.
3. 변환된 텍스트는 다시 로컬 컴퓨터의 audio memo 폴더에 저장된다.
어떻게 하면 구현할 수 있을까?
---
### 답변
AWS Boto3와 Python을 사용하여 음성 메모 파일을 S3로 업로드하고, AWS Transcribe로 텍스트 변환을 처리한 후, 변환된 텍스트를 로컬 폴더에 저장하는 프로그램을 작성할 수 있습니다. 이를 위해 필요한 주요 단계는 다음과 같습니다:
1. **S3로 파일 업로드**
2. **AWS Transcribe 작업 생성**
3. **Transcribe 작업 결과 가져오기 및 로컬 저장**
아래는 이 프로세스를 구현하는 예제 코드입니다:
### 사전 준비
1. AWS CLI를 설치하고, `aws configure` 명령어로 자격 증명과 설정을 완료하세요.
2. 필요한 Python 패키지 설치: `boto3` 및 `watchdog`
```sh
pip install boto3 watchdog
```
```python
import boto3
import os
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# AWS S3 및 Transcribe 클라이언트 생성
s3_client = boto3.client('s3')
transcribe_client = boto3.client('transcribe')
bucket_name = 'audiotransjake' # S3 버킷 이름
local_folder = '/path/to/your/local/folder' # 로컬 폴더 경로
output_folder = '/path/to/output/folder' # 텍스트 파일 저장 폴더 경로
class UploadHandler(FileSystemEventHandler):
def on_created(self, event):
if event.is_directory:
return
filename = event.src_path
self.upload_to_s3(filename)
def upload_to_s3(self, filename):
try:
basename = os.path.basename(filename)
s3_client.upload_file(filename, bucket_name, basename)
print(f'File {basename} uploaded to S3')
self.start_transcription(basename)
except Exception as e:
print(f'Error uploading file: {e}')
def start_transcription(self, s3_filename):
job_name = f'transcription_{int(time.time())}'
job_uri = f's3://{bucket_name}/{s3_filename}'
transcribe_client.start_transcription_job(
TranscriptionJobName=job_name,
Media={'MediaFileUri': job_uri},
MediaFormat='wav', # 파일 형식에 따라 변경
LanguageCode='ko-KR',
OutputBucketName=bucket_name
)
print(f'Transcription job {job_name} started')
# Transcribe 작업 완료 여부 확인
while True:
status = transcribe_client.get_transcription_job(TranscriptionJobName=job_name)
if status['TranscriptionJob']['TranscriptionJobStatus'] in ['COMPLETED', 'FAILED']:
if status['TranscriptionJob']['TranscriptionJobStatus'] == 'COMPLETED':
self.download_transcription(job_name)
break
print('Waiting for transcription to complete...')
time.sleep(5)
def download_transcription(self, job_name):
transcription = transcribe_client.get_transcription_job(TranscriptionJobName=job_name)
transcription_uri = transcription['TranscriptionJob']['Transcript']['TranscriptFileUri']
transcript_text = s3_client.get_object(Bucket=bucket_name, Key=transcription_uri.split('/')[-1])['Body'].read().decode('utf-8')
# 텍스트 파일로 저장
output_file = os.path.join(output_folder, f'{job_name}.txt')
with open(output_file, 'w', encoding='utf-8') as f:
f.write(transcript_text)
print(f'Transcription saved to {output_file}')
# 로컬 폴더 감시 설정
event_handler = UploadHandler()
observer = Observer()
observer.schedule(event_handler, path=local_folder, recursive=False)
observer.start()
print(f'Starting to monitor {local_folder} for new files...')
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
```
### 주요 설명
1. **S3로 파일 업로드**: `UploadHandler` 클래스에서 새로운 파일이 로컬 폴더에 생성될 때 S3 버킷으로 업로드합니다.
2. **AWS Transcribe 작업 생성**: `start_transcription` 메소드에서 업로드된 파일을 바탕으로 Transcribe 작업을 시작합니다.
3. **Transcribe 결과 저장**: `download_transcription` 메소드에서 변환된 텍스트를 가져와 로컬 파일로 저장합니다.
### 추가 고려사항
- AWS 자격 증명이 올바르게 설정되었는지 확인하세요.
- `watchdog`을 사용하여 폴더를 모니터링하므로, 시스템 성능에 영향을 줄 수 있으니 주의하세요.
- S3에 업로드 시 파일 형식과 Transcribe 설정이 일치하는지 확인하세요.
이 코드를 사용하여 음성 파일이 추가될 때마다 자동으로 S3에 업로드하고 텍스트로 변환하여 저장할 수 있습니다.