import glob
import subprocess
import os
import shutil
from google.colab import drive
from tqdm import tqdm
# ✅ Google Drive가 언마운트되면 자동으로 다시 마운트하는 함수
def remount_drive():
print("🔄 Google Drive 연결이 끊어졌습니다. 다시 마운트 중...")
drive.flush_and_unmount()
drive.mount('/content/drive')
# ✅ 압축 파일이 위치한 디렉토리 설정 (Google Drive)
drive_tar_dir = "/content/drive/MyDrive/DMS/data/AI-HUB/Training/원천데이터/TS4/tar/"
tar_pattern = drive_tar_dir + "*.tar"
# ✅ Colab 로컬 저장소에서 압축 해제할 경로
local_extract_dir = "/content/temp_extracted/"
os.makedirs(local_extract_dir, exist_ok=True)
# ✅ Colab에서 압축 해제할 `.tar` 파일 저장 경로
local_tar_dir = "/content/temp_tar/"
os.makedirs(local_tar_dir, exist_ok=True)
# ✅ 압축 해제 후 최종 저장할 Google Drive 경로
drive_destination_dir = "/content/drive/MyDrive/DMS/data/AI-HUB/Training/원천데이터/TS4"
# ✅ tar 파일 목록 가져오기
tar_files = glob.glob(tar_pattern)
total_files = len(tar_files)
print(f"📂 총 {total_files}개의 tar 파일을 찾았습니다.")
# ✅ 각 tar 파일을 개별적으로 추출하며 진행률 표시
for i, archive in enumerate(tqdm(tar_files, desc="Extracting TAR files", unit="file")):
print(f"\n[{i+1}/{total_files}] Google Drive에서 로컬로 파일 복사 중: {archive}")
# ✅ Google Drive에서 Colab 로컬로 tar 파일 복사
local_tar_file = os.path.join(local_tar_dir, os.path.basename(archive))
try:
shutil.copy2(archive, local_tar_file) # 파일 복사
except Exception as e:
print(f"❌ 파일 복사 실패: {archive} (에러: {str(e)})")
remount_drive() # Google Drive 재마운트
continue
# ✅ tar 내부 파일 개수 확인
list_command = ["tar", "-tf", local_tar_file]
result = subprocess.run(list_command, capture_output=True, text=True)
if result.returncode != 0:
print(f" ❌ 오류 발생: {local_tar_file}")
print(f" ⚠️ 오류 내용: {result.stderr}")
remount_drive() # Google Drive 재마운트
continue # 다음 tar 파일로 넘어감
file_list = result.stdout.split("\n")
total_items = len(file_list)
# ✅ Colab 로컬에서 압축 해제
with tqdm(total=total_items, desc=f"Extracting {os.path.basename(local_tar_file)}", unit="file") as pbar:
extract_command = ["tar", "--exclude=*.jpg", "-xf", local_tar_file, "-C", local_extract_dir]
process = subprocess.Popen(extract_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
for _ in process.stdout:
pbar.update(1)
process.wait()
if process.returncode != 0:
print(f" ❌ 오류 발생: {local_tar_file}")
print(f" ⚠️ 오류 내용: {process.stderr.read()}")
remount_drive() # Google Drive 재마운트
continue
print(f" ✅ {local_tar_file} 압축 해제 완료.")
# ✅ Google Drive로 이동
print(f" 🚀 Google Drive로 이동 중: {local_extract_dir} → {drive_destination_dir}")
for item in os.listdir(local_extract_dir):
src_path = os.path.join(local_extract_dir, item)
dest_path = os.path.join(drive_destination_dir, item)
shutil.move(src_path, dest_path)
# ✅ 임시 파일 정리
print(f" 🧹 Colab 로컬 임시 파일 정리 중...")
os.remove(local_tar_file) # 압축 해제 후 tar 파일 삭제
shutil.rmtree(local_extract_dir) # 압축 해제된 폴더 삭제 후 재생성
os.makedirs(local_extract_dir, exist_ok=True)
print("\n✅ 모든 tar 파일의 압축 해제가 완료되었습니다.")