123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194 |
- import re
- import glob
- import time
- import tqdm
- from datetime import datetime
- from shutil import copyfile
- import os
- a1 = "머리를 (()) 그담에 뭐 하트도 날리고 뽀뽀도 해주고 윙크도 해주고 토닥토닥 안아주고 이 두 가지를 적절하게 잘 사용하면 굉장히 인제 조금 칭찬 방법이 되는 거죠."
- a2 = "니가 이반에서 제일 머리가 좋아 그런 거는 쪼끔 안 좋습니다 왜냐면 인제 자기보다 예쁜 아이가 있는 다른 반에 가지 않을려고 {laughing} 다른 친구보다 다른 형제자매들 보다 훨씬 낫다는 식의 비교 칭찬은 가급적 하지 않는 것이 좋습니다."
- a3 = "그니까 아까 왜 너 얼마나 잘할 거 같애 칠십 퍼센트요 팔십 퍼센트요 그니까 (아이 캔 두잇)(i can do it 나 할 수 있을 거 같애요 라고 하는 건데요."
- a4 = f" "
- a = [a1, a2, a3, a4]
- pattern_1 = r'\(\(\)\)' ## (())
- pattern_2 = r'\{(\w*)\}' ## {@@@}
- pattern_3 = r'\((\w*)\)' ## (@@@) 찾기
- pattern_4 = r'[\(\)\{\}]' ## (, ), {, } 찾기
- pattern_4_1 = r'&([\w|-]*)&' ## &(id-id123)& 찾기
- pattern_5 = r"[^\uAC00-\uD7A30-9a-zA-Z\s]" ## 특문 repl=' '
- pattern_6 = r" {2,}" ## 빈 문자열 2개 이상
- def postprocess(origin):
- _a = re.sub(pattern=pattern_1, repl='', string = origin)
- _a = re.sub(pattern=pattern_2, repl='', string = _a)
- _a = re.sub(pattern=pattern_3, repl='', string = _a)
- tmp = re.search(pattern=pattern_4, string = _a)
- if tmp:
- return None
-
- tmp = re.search(pattern=pattern_4_1, string = _a)
- if tmp:
- return None
- _a = re.sub(pattern=pattern_5, repl = " ", string = _a)
- _a = re.sub(pattern=pattern_6, repl = " ", string = _a)
- _a = _a.strip()
- if len(_a) == 0:
- return None
- else:
- return _a
- # test = '{laughing} 기억하고 있어야지 후생에도 혹시 {laughing} (()) (()) (광)'
- # test = '저희 음악 한 곡 듣고 와서 조금 더 마지막 사연 이야기 나눠볼게요.여러분의 의견도 많이 보내주시길 바래요.&company-name225& &tel-num1& 오십 원 혹은 백 원의 유료문자'
- # test = '저희 음악 한 곡 듣고 와서 조금 더 마지막 사연 이야기 나눠볼게요.여러분의 의견도 많이 보내주시길 바래요.'
- # print(list(map(postprocess, a)))
- # print(re.sub(pattern=pattern_1, repl='', string = test))
- # exit()
- root_dir = "/root/sata/broadcast_split/data/"
- total_txt = "/root/mnt/data_aug/audio_augmentation/script/total.txt"
- # with open(total_txt, 'w+') as tf:
- # for idx, content in enumerate(a):
- # content = content.strip()
-
- # post = postprocess(content)
- # if post is not None:
- # print(content + "\n --> " + post)
- # tf.write(post + "\n")
- # exit()
- ## 1 = broadcast_split
- ## 2 = broadcast/split
- ## 1에 있는 ogg를 기준으로, txt2를 가져오고, 이를 후처리 해서 txt1에 적용함.
- def broadcast_move_split2_split1(time_str):
- total_txt_file = '/root/mnt/data_aug/audio_augmentation/script/txt/broadcast_total_2.txt'
- ## for test
- # total_txt_file = '/root/mnt/data_aug/audio_augmentation/script/txt/broadcast_total_test.txt'
-
- ogg_path = '/root/sata/etc/broadcast_split/data'
- ## for text
- # ogg_path = '/root/mnt/data_aug/audio_augmentation/sample/broadcast_split'
- post_none_list = []
- not_exist_list = []
- ogg_list = []
- print('glob start !!')
- if os.path.isfile(total_txt_file):
- with open(total_txt_file, 'r') as f:
- for ogg in f:
- tt = ogg.strip()
- ogg_list.append(tt)
- else:
- ogg_list = glob.glob(os.path.join(ogg_path, '**/*.ogg'), recursive=True)
- with open(total_txt_file, 'w+') as f:
- for ogg in ogg_list:
- f.write(ogg + '\n')
-
- print('glob end !!', len(ogg_list))
- # return
- for source_ogg in tqdm.tqdm(ogg_list):
- source_ogg = source_ogg.strip()
-
- split1_txt_file = source_ogg.strip().replace("ogg", "txt")
- split2_txt_file = source_ogg.strip().replace("ogg", "txt").replace("broadcast_split", "broadcast/split")
- if not os.path.isfile(split2_txt_file):
- not_exist_list.append([source_ogg, split2_txt_file])
- else:
- with open(split2_txt_file, 'r') as f:
- content = f.read().strip()
- post = postprocess(content)
- with open(split1_txt_file, 'w') as f:
- if post is None:
- post_none_list.append([split2_txt_file, content])
- if os.path.isfile(split1_txt_file):
- os.remove(split1_txt_file)
- tmp_ogg = split1_txt_file.replace('txt', 'ogg')
- if os.path.isfile(tmp_ogg):
- os.remove(tmp_ogg)
- else:
- f.write(post + '\n')
- if post_none_list:
- print('postprocess none = ', len(post_none_list))
- txt = f'log/post_none_{time_str}.txt'
- with open(txt, 'w+') as f:
- for item in post_none_list:
- f.write("\t".join(item) + "\n")
-
- if not_exist_list:
- print('not exist = ', len(not_exist_list))
- txt = f'log/not_exist_{time_str}.txt'
- for item in post_none_list:
- with open(txt, 'w+') as f:
- f.write("\t".join(item) + "\n")
-
-
-
-
- if __name__ == '__main__':
- now = datetime.now()
- time_str = now.astimezone().strftime('%Y_%m_%d_%H_%M_%S')
- # print(time_str)
- # # exit()
- broadcast_move_split2_split1(time_str)
- # failure = []
- # failure_txt = f"broadcast_failure_{time_str}.txt"
- # with open(total_txt, 'w+') as tf:
- # for file in tqdm.tqdm(glob.glob(f'{root_dir}**/*.txt', recursive=True)):
- # content = ''
- # with open(file, 'r') as f:
- # content = f.read().strip()
-
- # post = postprocess(content)
- # # print(str(idx) + " : " + content + "\n --> " + post)
- # with open(file, 'w') as f:
-
- # if post is not None:
- # # tf.write(file + "\n")
- # # f.write(post)
- # pass
- # else:
- # failure.append([file, content, post])
-
- # # print(failure)
- # print(len(failure))
- # with open(failure_txt, 'w+') as ft:
- # for fail in tqdm.tqdm(failure):
- # if fail[2] is None: fail[2] = 'None'
- # ft.write("\t".join(fail) + "\n")
-
- # else:
- # print(file)
- # cnt += 1
|