200720 음성 파일 Text 변환 (feat Google STT)

	# -*- coding: utf-8 -*-
	import os
	import shutil
	import glob
	from subprocess import call
	import speech_recognition as sr
	from tqdm import tqdm
	from multiprocessing.dummy import Pool
	import time
	pool = Pool(8) # Number of concurrent threads
	r = sr.Recognizer() # speech_recognition recognizer r에 저장
	with open("api-key.json") as f: # 구글 클라우드 인증키 불러오기
	GOOGLE_CLOUD_SPEECH_CREDENTIALS = f.read() # 해당 변수에 인증키 저장
	def copy(dir1, dir2, rm=True):
	if not os.path.exists(dir2):
	shutil.copytree(dir1, dir2)
	if rm:
	shutil.rmtree(dir2)
	shutil.copytree(dir1, dir2)
	#noisy_TIMIT 폴더를 하위경로와 함께 통째로 복사하기 위한 함수
	direction = input("input the folder noisefile contained :")
	curr_path = os.path.join(os.path.abspath(os.getcwd()), direction)
	txt_path = os.path.join(os.path.abspath(os.getcwd()), 'tmp_filelist.txt')
	copy_path = os.path.join(os.path.abspath(os.getcwd()), 'rawfile') # rawfile 폴더 생성용
	copy(curr_path, copy_path, rm=True)
	for root, dirs, files in os.walk(copy_path):
	if files:
	for file in files:
	if os.path.splitext(file)[1] == '.snd' :
	newname = os.path.splitext(file)[0] + '.raw'
	os.rename(os.path.join(root,file), os.path.join(root, newname))
	with open(txt_path,'wt') as f:
	for root, dirs, files in os.walk(copy_path):
	if files:
	for file in files:
	if file.find("filelist") < 0:
	filename = os.path.join(root, file)
	f.write(filename + '\n')
	call(["sort","tmp_filelist.txt","-o", os.path.join(os.path.abspath(os.getcwd()), 'filelist.txt')])
	call(["rm","-rf",txt_path])
	try:
	with open('filelist.txt', 'r') as file: #파일리스트를 연다
	lines = file.readlines() #모든 줄을 lines에 리스트의 원소로 전부 저장한다
	lines.sort() #리스트를 사전순 정렬한다
	for line in lines: #txt파일의 모든 줄에 대하여 (경로)
	temp = os.path.splitext(line)
	temp2 = [temp[0], '.raw'] # .raw\n 제거 후 .raw 붙이기
	origfile = ''.join(temp2)
	temp2 = [temp[0], '.wav'] # .raw\n 제거 후 .wav 붙이기 # 해당 줄의 경로 split
	newfile = ''.join(temp2)
	call(["sox","-r","8000", "-e", "signed-integer", "-c", "1", "-b", "16", origfile, newfile])
	for root, dirs, files in os.walk(copy_path): #rawfile 내의 모든 파일 검색
	if files:
	for file in files: #모든 파일에 대해서
	if os.path.splitext(file)[1] in ['.raw', '.RAW']: #확장자가 raw이면
	os.remove(os.path.join(root,file)) #삭제
	with open(txt_path,'wt') as f:
	for root, dirs, files in os.walk(copy_path):
	if files:
	for file in files:
	if file.find("filelist") < 0:
	filename = os.path.join(root, file)
	f.write(filename + '\n')
	call(["sort","tmp_filelist.txt","-o", os.path.join(os.path.abspath(os.getcwd()), 'filelist.txt')])
	call(["rm","-rf",txt_path])
	except FileNotFoundError:
	print("The folder name is invalid")
	#transcribe start
	def transcribe(file): # Transcribe 함수 선언
	print(os.path.abspath(file) + " started" + '\n')
	with sr.AudioFile(os.path.abspath(file)) as source: # name 경로에 해당하는 오디오 파일을 source로 open하여
	audio = r.record(source) # audio는 source가 담고 있는 파일을 record
	try:
	# text = r.recognize_google_cloud(audio, credentials_json=GOOGLE_CLOUD_SPEECH_CREDENTIALS)
	text = r.recognize_google_cloud(audio, credentials_json=GOOGLE_CLOUD_SPEECH_CREDENTIALS, language="ko-KR")
	print(os.path.abspath(file) + " done")
	dirtext = [os.path.abspath(file), " : ", text, "\n"]
	''.join(dirtext)
	return dirtext
	except sr.UnknownValueError: # 값을 읽어올 수 없을 때
	print("Google Speech Recognition could not understand file")
	dirtext = [os.path.abspath(file), " : -\n"]
	''.join(dirtext)
	return dirtext
	except sr.RequestError as e: # Google Speech에 접속할 수 없을때
	print("Could not request results from Google Speech Recognition service; {0}".format(e))
	dirtext = [os.path.abspath(file), " : -\n"]
	''.join(dirtext)
	return dirtext
	except sr.HTTPError as e: # Http 에러
	print("Couldn't connect to the websites perhaps , Hyper text transfer protocol error", e)
	dirtext = [os.path.abspath(file), " : -\n"]
	''.join(dirtext)
	return dirtext
	start = time.time() # 시작 시간 저장
	with open('transcribe.txt', 'w') as f:
	with open('filelist.txt', 'r') as file:
	modifl = []
	lines = file.readlines()
	lines.sort()
	for line in lines:
	temp = os.path.splitext(line)
	temp = [temp[0], '.wav']
	line = ''.join(temp)
	modifl.append(line) #맨 끝의 \n을 없애기 위한 작업
	modifl.sort()
	all_text = pool.map(transcribe, modifl)
	for l in all_text:
	f.write(''.join(l))
	print("time :", time.time() - start)
	call(["rm","-rf",os.path.join(os.path.abspath(os.getcwd()), 'filelist.txt')])
	call(["rm","-rf",copy_path])
	pool.close() # 리소스 방지를 위한 close 호출
	pool.join() # 작업 완료 대기를 위한 join 호출

view raw transcribe.py hosted with ❤ by GitHub

 

~line 50까지 파일명 변경 및 파일리스트 출력작업

1 raw파일 (or snd파일) 들어있는 폴더를 통째로 새로 복사 (원본음성 건드리면 안되기때문에)

2 해당 폴더내의 모든 snd파일을 찾아서 raw로 확장자를 변경

이작업 하는 이유는 snd는 sox에서 못읽어들이는데 snd랑 raw랑 확장자만 다르지 파일자체는 똑같음

3 해당 폴더내의 모든 파일을 경로랑 같이 txt에 저장 (filelist.txt)

이때 file.find("filelist") < 0 설정해주는 이유는 filelist.txt 포함안되게 하려고

 

4 filelist.txt 오름차순 정렬 (transcribe할 때 순서 꼬이면 문장 비교가 안 돼서 미리 정렬해줌)

 

~line 82까지 raw에서 wav로 변경작업

1 filelist.txt 내의 모든 라인 (파일경로) 마다 raw파일이름을 origfile (오리지널), 확장자를 wav로 바꾼것을 newfile 변수로 설정

2 sox 라이브러리 써서 origfile에 해당하는 파일을 raw에서 wav로 변환해서 newfile 파일명으로 저장

3 이 시점에서 폴더내에 raw 파일이랑 wav 파일이 공존하기 때문에 raw만 싹 지움

어차피 아까 폴더 통째로 복사해뒀기 때문에 파일 날려먹어도 원본은 따로 존재

4 해당 wav파일들로 다시한번 filelist.txt 만듦 + 정렬

 

~끝까지 transcribe

filelist에 지정된 모든 파일들에 대해서 구글 클라우드 서비스에 파일을 날리고 인식된 텍스트를 받아오기 때문에 파일리스트 생성작업을 따로 해줌

transcribe 하는 함수는 transcribe(file) 로 따로 선언해줬는데 그이유는 이게 하나하나 날리고 받아오는게 진짜 오래걸려서 나중에 멀티프로세스 (8개한번에돌림) 설정해주려고 일부러 파일경로만 집어넣으면 변환하는 함수로 선언해준거임

함수 선언하고 뒤에 아까 저장해뒀던 filelist.txt 열어서 해당 줄마다 저장되어있는 파일 경로를 리스트에 다시 저장해둔 다음에 그 경로 담은 리스트 (modifl)랑 transcribe 함수를 pool.map에 input으로 입력하면 modifl 리스트에 존재하는 모든 값들을 transcribe 함수의 입력변수로 사용하겠다는뜻

여튼 여기까지 설정해주면 알아서 8개씩 병렬처리 해줌 (pool = Pool(8) 에서 8개 한번에 돌리겟다는 의미)

8개씩 돌려도 진짜 오래걸리는건 정말오래걸려서 멀티프로세싱이 필수

 

나중에 응용할일이 있지않을까 싶어서 저장해둠

파일명 일괄변경이나 특정 확장자 가진 파일 리스트생성같은건 아직도 종종써먹음