123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170 |
- from tqdm import tqdm
- from postprocess import postprocess
- import glob
- import os
- import re
- import re
- from datetime import datetime
- now = datetime.now()
- time_str = now.strftime('%Y_%m_%d_%H_%M_%S')
- ## none 이면 저장 안됨.
- def broadcast_none_check():
- total_txt = "/root/mnt/data_aug/audio_augmentation/script/total.txt"
- failure_txt = f"broadcast_none_{time_str}.txt"
- ogg_path = '/root/sata/broadcast/split/data'
- total_ogg_txt = 'broadcast_total_ogg.txt'
- total_txt_set = set()
- ogg_set = set()
- failure = []
- cnt = 0
- def split1_split2_compare():
- failure_txt = f"broadcast_split1_split2_compare_{time_str}.txt"
- ogg_files = glob.glob(os.path.join('/root/sata/broadcast_split/data', '**/*.ogg'), recursive=True)
- for ogg in tqdm(ogg_files):
- txt_file = ogg.strip().replace("ogg", "txt").replace("broadcast_split", "broadcast/split")
- if not os.path.isfile(txt_file):
- failure.append([ogg, txt_file])
- with open(failure_txt, 'w+') as ft:
- for fail in failure:
- ft.write("\t".join(fail) + "\n")
- print("len failire = ", len(failure))
-
- def split2_split1_compare():
- failure_txt = f"broadcast_split2_split1_compare_{time_str}.txt"
- txt_files = glob.glob(os.path.join('/root/sata/broadcast/split/data', '**/*.txt'), recursive=True)
- for txt in tqdm(txt_files):
- ogg_file = txt.strip().replace("txt", "ogg").replace("broadcast/split", "broadcast_split" )
- if not os.path.isfile(ogg_file):
- failure.append([txt, ogg_file])
- with open(failure_txt, 'w+') as ft:
- for fail in failure:
- ft.write("\t".join(fail) + "\n")
- print("len failire = ", len(failure))
-
- split2_split1_compare()
- return
- def broadcast_special_check():
- failure = []
- failure_txt = f"broadcast_special_check{time_str}.txt"
- pattern_1 = r'[%$#]'
- txt_files = glob.glob(os.path.join('/root/sata/broadcast/split/data', '**/*.txt'), recursive=True)
- for txt in tqdm(txt_files):
- with open(txt, 'r') as t:
- tt = t.read().strip()
- tt = tt.strip()
- tmp = re.search(pattern=pattern_1, string = tt)
- if tmp:
- failure.append([txt, tt])
- with open(failure_txt, 'w+') as ft:
- for fail in failure:
- ft.write("\t".join(fail) + "\n")
- print("len failire = ", len(failure))
- def broadcast_post_check():
- failure_list = []
- for source_txt in tqdm.tqdm(glob.glob(os.path.join('/root/sata/broadcast/split/data', '**/*.txt'), recursive=True)):
- with open(source_txt, 'r') as f:
- content = f.read().strip()
- post = postprocess(content)
- if post is None:
- failure_list.append([source_txt, content])
- failure_txt = f"broadcast_post.txt"
- with open(failure_txt, 'w+') as ft:
- for fail in failure_list:
- ft.write("\t".join(fail) + "\n")
- print("len = ", len(failure_list))
- def broadcast_post_check():
- total_txt = "/root/mnt/data_aug/audio_augmentation/script/total.txt"
- failure_txt = f"broadcast_none_{time_str}.txt"
- ogg_path = '/root/sata/broadcast_split/data'
- total_ogg_txt = 'broadcast_total_ogg.txt'
- total_txt_set = set()
- ogg_set = set()
- failure = []
- cnt = 0
- try:
- with open(total_ogg_txt, 'r') as tf:
- for ogg in tqdm(tf):
- ogg = ogg.strip()
- txt_file = ogg.replace('ogg', 'txt')
- if not os.path.isfile(txt_file):
- failure.append(ogg)
- else:
- cnt += 1
- except:
- tf.close()
- finally:
- with open(failure_txt, 'w+') as ft:
- for fail in failure:
- ft.write(fail + "\n")
- print('cnt', cnt, len(failure))
- return
- def freetalk_special_check():
- ogg_path = '/root/nas/data/freetalk/ogg'
- total_txt = 'freetalk_total.txt'
- failure_txt = f"freetalk_failure_{time_str}.txt"
- failure = []
- pattern = r"[^\uAC00-\uD7A30-9a-zA-Z\s]" ## 특문
- with open(total_txt, 'w+') as tf:
- for txt_file in tqdm(glob.glob(os.path.join(ogg_path, '**/*.txt'), recursive=True), leave=True):
- tf.write(f"{txt_file}\n")
- with open(txt_file, 'r') as f:
- content = f.read().strip()
-
-
- exist = re.search(pattern=pattern, string=content)
-
- if exist:
- failure.append((file, content))
- if idx == 100:
- break
- if len(failure) >= 1:
- with open(failure_txt, "w+") as f:
- for _failure in failure:
- f.write(f"{' '.join(_failure)}\n")
- return
-
- # print('test')
- # freetalk_special_check()
-
- if __name__ == '__main__':
- # broadcast_none_check()
- broadcast_special_check()
- # freetalk_special_check()
|