463 lines
18 KiB
Python
463 lines
18 KiB
Python
import os
|
|
import shutil
|
|
import asyncio
|
|
import re
|
|
import json
|
|
import tempfile
|
|
import zipfile
|
|
from pathlib import Path
|
|
from typing import List, Dict, Any, Optional, AsyncGenerator, Tuple
|
|
from fastapi import HTTPException
|
|
from .config import settings
|
|
|
|
CATEGORY_EXTENSIONS = {
|
|
"video": [".mp4", ".mkv", ".avi", ".mov", ".wmv", ".flv", ".webm", ".ts"],
|
|
"audio": [".mp3", ".flac", ".wav", ".aac", ".ogg", ".m4a", ".wma"],
|
|
"image": [".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".svg"],
|
|
"archive": [".zip", ".rar", ".7z", ".tar", ".gz", ".bz2", ".iso"],
|
|
"document": [".pdf", ".txt", ".md", ".csv", ".json", ".xml", ".html", ".smi", ".srt", ".vtt"],
|
|
}
|
|
|
|
def get_category(ext: str) -> str:
|
|
ext_lower = ext.lower()
|
|
for cat, exts in CATEGORY_EXTENSIONS.items():
|
|
if ext_lower in exts:
|
|
return cat
|
|
return "other"
|
|
|
|
def resolve_secure_path(sub_path: str = "") -> Path:
|
|
"""Path traversal 방지 및 경로 확인"""
|
|
base_dir = Path(settings.DOWNLOADS_BASE_PATH).resolve()
|
|
if not base_dir.exists():
|
|
# 디렉토리가 없을 경우 생성 시도
|
|
base_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 상대 경로 처리
|
|
target_path = (base_dir / sub_path.lstrip("/")).resolve()
|
|
|
|
try:
|
|
# base_dir 내부 경로인지 확인
|
|
target_path.relative_to(base_dir)
|
|
except ValueError:
|
|
raise HTTPException(status_code=403, detail="접근이 제한된 잘못된 경로입니다.")
|
|
|
|
if not target_path.exists():
|
|
raise HTTPException(status_code=404, detail="요청한 파일 또는 폴더를 찾을 수 없습니다.")
|
|
|
|
return target_path
|
|
|
|
def list_directory(sub_path: str = "") -> Dict[str, Any]:
|
|
"""디렉토리 내 파일 및 폴더 목록 조회"""
|
|
target_path = resolve_secure_path(sub_path)
|
|
base_dir = Path(settings.DOWNLOADS_BASE_PATH).resolve()
|
|
|
|
if not target_path.is_dir():
|
|
raise HTTPException(status_code=400, detail="디렉토리가 아닙니다.")
|
|
|
|
items = []
|
|
total_size = 0
|
|
|
|
try:
|
|
entries = sorted(list(target_path.iterdir()), key=lambda p: (not p.is_dir(), p.name.lower()))
|
|
for entry in entries:
|
|
try:
|
|
stat = entry.stat()
|
|
is_dir = entry.is_dir()
|
|
rel_path = str(entry.relative_to(base_dir)).replace("\\", "/")
|
|
|
|
size = 0
|
|
if not is_dir:
|
|
size = stat.st_size
|
|
total_size += size
|
|
|
|
ext = entry.suffix
|
|
category = "folder" if is_dir else get_category(ext)
|
|
|
|
items.append({
|
|
"name": entry.name,
|
|
"path": rel_path,
|
|
"is_dir": is_dir,
|
|
"size": size,
|
|
"modified": stat.st_mtime,
|
|
"extension": ext,
|
|
"category": category,
|
|
})
|
|
except (PermissionError, FileNotFoundError):
|
|
continue
|
|
|
|
current_rel = str(target_path.relative_to(base_dir)).replace("\\", "/")
|
|
if current_rel == ".":
|
|
current_rel = ""
|
|
|
|
return {
|
|
"current_path": current_rel,
|
|
"items": items,
|
|
"total_items": len(items),
|
|
"total_size": total_size,
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"디렉토리 읽기 실패: {str(e)}")
|
|
|
|
def delete_path(sub_path: str) -> None:
|
|
"""파일 또는 폴더 삭제"""
|
|
target_path = resolve_secure_path(sub_path)
|
|
base_dir = Path(settings.DOWNLOADS_BASE_PATH).resolve()
|
|
|
|
# 루트 디렉토리 자체는 삭제 불가
|
|
if target_path == base_dir:
|
|
raise HTTPException(status_code=400, detail="최상위 다운로드 디렉토리는 삭제할 수 없습니다.")
|
|
|
|
try:
|
|
if target_path.is_dir():
|
|
shutil.rmtree(target_path)
|
|
else:
|
|
target_path.unlink()
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"삭제 실패: {str(e)}")
|
|
|
|
async def check_archive_encrypted(sub_path: str) -> Dict[str, Any]:
|
|
"""압축 파일 내부 암호화 여부 초고속 검사"""
|
|
archive_path = resolve_secure_path(sub_path)
|
|
if not archive_path.is_file():
|
|
raise HTTPException(status_code=400, detail="압축 파일이 아니거나 존재하지 않습니다.")
|
|
|
|
# 1. zip 파일인 경우: 파이썬 표준 라이브러리로 즉각 확인 (대용량 파일도 수 밀리초만에 완료)
|
|
if archive_path.suffix.lower() == ".zip":
|
|
try:
|
|
import zipfile
|
|
with zipfile.ZipFile(str(archive_path)) as zf:
|
|
is_encrypted = any(z.flag_bits & 0x1 for z in zf.infolist())
|
|
return {
|
|
"is_encrypted": is_encrypted,
|
|
"filename": archive_path.name,
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
# 2. 기타 확장자(.7z, .rar 등)인 경우: 7z l 명령으로 확인 (타임아웃 적용)
|
|
cmd = ["7z", "l", "-slt", "-p-", str(archive_path)]
|
|
try:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
stdin=asyncio.subprocess.DEVNULL
|
|
)
|
|
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=3.0)
|
|
out_text = stdout.decode('utf-8', errors='ignore') + stderr.decode('utf-8', errors='ignore')
|
|
|
|
is_encrypted = (
|
|
"Encrypted = +" in out_text or
|
|
any(kw in out_text for kw in ["Wrong password", "Enter password", "Can not open encrypted"])
|
|
)
|
|
return {
|
|
"is_encrypted": is_encrypted,
|
|
"filename": archive_path.name,
|
|
}
|
|
except Exception:
|
|
return {
|
|
"is_encrypted": False,
|
|
"filename": archive_path.name,
|
|
}
|
|
|
|
async def extract_archive(
|
|
sub_path: str,
|
|
password: Optional[str] = None,
|
|
create_subfolder: bool = True,
|
|
custom_dest: Optional[str] = None,
|
|
delete_archive: bool = False
|
|
) -> Dict[str, Any]:
|
|
"""압축 파일 해제 (7z 활용, 암호 지원)"""
|
|
archive_path = resolve_secure_path(sub_path)
|
|
base_dir = Path(settings.DOWNLOADS_BASE_PATH).resolve()
|
|
|
|
if not archive_path.is_file():
|
|
raise HTTPException(status_code=400, detail="압축 파일이 아니거나 존재하지 않습니다.")
|
|
|
|
parent_dir = archive_path.parent
|
|
if create_subfolder:
|
|
folder_name = custom_dest.strip() if (custom_dest and custom_dest.strip()) else archive_path.stem
|
|
folder_name = folder_name.replace("/", "_").replace("\\", "_")
|
|
dest_dir = parent_dir / folder_name
|
|
else:
|
|
dest_dir = parent_dir
|
|
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 7z 명령어 구성
|
|
cmd = ["7z", "x", "-y"]
|
|
if password is not None and str(password).strip():
|
|
cmd.append(f"-p{password.strip()}")
|
|
else:
|
|
cmd.append("-p-") # 암호 프롬프트 대기 없이 즉시 에러 반환
|
|
|
|
cmd.append(f"-o{str(dest_dir)}")
|
|
cmd.append(str(archive_path))
|
|
|
|
try:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
stdin=asyncio.subprocess.DEVNULL
|
|
)
|
|
stdout, stderr = await proc.communicate()
|
|
out_text = stdout.decode('utf-8', errors='ignore')
|
|
err_text = stderr.decode('utf-8', errors='ignore')
|
|
combined_output = out_text + "\n" + err_text
|
|
|
|
# 암호 오류 패턴 검사
|
|
password_err_keywords = [
|
|
"Wrong password",
|
|
"Enter password",
|
|
"Data Error in encrypted file",
|
|
"Cannot open encrypted",
|
|
"Headers Error",
|
|
"Sub items Errors"
|
|
]
|
|
has_password_error = any(kw in combined_output for kw in password_err_keywords)
|
|
|
|
if has_password_error:
|
|
if not password or not str(password).strip():
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="PASSWORD_REQUIRED: 이 파일은 암호로 보호되어 있습니다. 압축 해제 암호를 입력해주세요."
|
|
)
|
|
else:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="WRONG_PASSWORD: 입력한 암호가 올바르지 않습니다. 암호를 다시 확인해주세요."
|
|
)
|
|
|
|
if proc.returncode != 0:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"압축 해제 중 오류가 발생했습니다: {err_text[:300] or out_text[:300]}"
|
|
)
|
|
|
|
# 원본 압축 파일 삭제 옵션
|
|
if delete_archive:
|
|
try:
|
|
archive_path.unlink()
|
|
except Exception:
|
|
pass
|
|
|
|
rel_dest = str(dest_dir.relative_to(base_dir)).replace("\\", "/")
|
|
if rel_dest == ".":
|
|
rel_dest = ""
|
|
|
|
return {
|
|
"success": True,
|
|
"message": f"'{archive_path.name}' 압축이 성공적으로 해제되었습니다.",
|
|
"dest_path": rel_dest
|
|
}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"압축 해제 실행 실패: {str(e)}")
|
|
|
|
async def extract_archive_stream(
|
|
sub_path: str,
|
|
password: Optional[str] = None,
|
|
create_subfolder: bool = True,
|
|
custom_dest: Optional[str] = None,
|
|
delete_archive: bool = False
|
|
) -> AsyncGenerator[str, None]:
|
|
"""실시간 진행률을 SSE 형식으로 스트리밍하는 압축 해제 함수"""
|
|
archive_path = resolve_secure_path(sub_path)
|
|
base_dir = Path(settings.DOWNLOADS_BASE_PATH).resolve()
|
|
|
|
if not archive_path.is_file():
|
|
yield f"data: {json.dumps({'status': 'error', 'error_type': 'NOT_FOUND', 'message': '압축 파일이 아니거나 존재하지 않습니다.'})}\n\n"
|
|
return
|
|
|
|
parent_dir = archive_path.parent
|
|
if create_subfolder:
|
|
folder_name = custom_dest.strip() if (custom_dest and custom_dest.strip()) else archive_path.stem
|
|
folder_name = folder_name.replace("/", "_").replace("\\", "_")
|
|
dest_dir = parent_dir / folder_name
|
|
else:
|
|
dest_dir = parent_dir
|
|
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 7z 명령어: -bsp1 (진행률 출력), -bb1 (파일명 출력)
|
|
cmd = ["7z", "x", "-y", "-bsp1", "-bb1"]
|
|
if password is not None and str(password).strip():
|
|
cmd.append(f"-p{password.strip()}")
|
|
else:
|
|
cmd.append("-p-")
|
|
|
|
cmd.append(f"-o{str(dest_dir)}")
|
|
cmd.append(str(archive_path))
|
|
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
stdin=asyncio.subprocess.DEVNULL
|
|
)
|
|
|
|
current_percent = 0
|
|
current_file = ""
|
|
files_done = 0
|
|
full_output = []
|
|
|
|
# 초기 상태 전송
|
|
yield f"data: {json.dumps({'status': 'progress', 'percent': 0, 'current_file': '압축 해제 준비 중...', 'files_done': 0})}\n\n"
|
|
|
|
try:
|
|
buffer = b""
|
|
while True:
|
|
chunk = await proc.stdout.read(256)
|
|
if not chunk:
|
|
break
|
|
buffer += chunk
|
|
# \r, \n, \x08 단위로 분할하여 파싱
|
|
if b"\n" in buffer or b"\r" in buffer or b"\x08" in buffer:
|
|
# 텍스트 디코딩
|
|
text = buffer.decode("utf-8", errors="ignore")
|
|
buffer = b""
|
|
full_output.append(text)
|
|
|
|
clean = re.sub(r"[\x08\r]+", " ", text)
|
|
|
|
percent_m = re.search(r"(\d+)\s*%", clean)
|
|
file_m = re.search(r"-\s+(.+)$", clean)
|
|
done_m = re.search(r"%\s+(\d+)", clean)
|
|
|
|
updated = False
|
|
if percent_m:
|
|
p = int(percent_m.group(1))
|
|
if p >= current_percent:
|
|
current_percent = p
|
|
updated = True
|
|
|
|
if file_m:
|
|
f = file_m.group(1).strip()
|
|
if f and f != current_file:
|
|
current_file = f
|
|
updated = True
|
|
|
|
if done_m:
|
|
files_done = int(done_m.group(1))
|
|
updated = True
|
|
|
|
if updated:
|
|
yield f"data: {json.dumps({'status': 'progress', 'percent': current_percent, 'current_file': current_file, 'files_done': files_done})}\n\n"
|
|
|
|
stderr_bytes = await proc.stderr.read()
|
|
err_text = stderr_bytes.decode("utf-8", errors="ignore")
|
|
combined_output = "".join(full_output) + "\n" + err_text
|
|
|
|
# 암호 오류 패턴 검사
|
|
password_err_keywords = [
|
|
"Wrong password",
|
|
"Enter password",
|
|
"Data Error in encrypted file",
|
|
"Cannot open encrypted",
|
|
"Headers Error",
|
|
"Sub items Errors"
|
|
]
|
|
has_password_error = any(kw in combined_output for kw in password_err_keywords)
|
|
|
|
if has_password_error:
|
|
if not password or not str(password).strip():
|
|
yield f"data: {json.dumps({'status': 'error', 'error_type': 'PASSWORD_REQUIRED', 'message': '이 파일은 암호로 보호되어 있습니다. 압축 해제 암호를 입력해주세요.'})}\n\n"
|
|
else:
|
|
yield f"data: {json.dumps({'status': 'error', 'error_type': 'WRONG_PASSWORD', 'message': '입력한 암호가 올바르지 않습니다. 암호를 다시 확인해주세요.'})}\n\n"
|
|
return
|
|
|
|
if proc.returncode != 0:
|
|
yield f"data: {json.dumps({'status': 'error', 'error_type': 'EXTRACT_FAILED', 'message': f'압축 해제 오류: {err_text[:200]}'})}\n\n"
|
|
return
|
|
|
|
# 원본 압축 파일 삭제 옵션
|
|
if delete_archive:
|
|
try:
|
|
archive_path.unlink()
|
|
except Exception:
|
|
pass
|
|
|
|
rel_dest = str(dest_dir.relative_to(base_dir)).replace("\\", "/")
|
|
if rel_dest == ".":
|
|
rel_dest = ""
|
|
|
|
yield f"data: {json.dumps({'status': 'completed', 'percent': 100, 'current_file': '압축 해제 완료', 'files_done': files_done, 'message': f'{archive_path.name} 압축이 성공적으로 해제되었습니다.', 'dest_path': rel_dest})}\n\n"
|
|
|
|
except Exception as e:
|
|
yield f"data: {json.dumps({'status': 'error', 'error_type': 'EXCEPTION', 'message': str(e)})}\n\n"
|
|
|
|
def create_folder_zip(sub_path: str) -> Tuple[str, str]:
|
|
"""폴더 전체를 zip으로 묶어 임시 파일 생성"""
|
|
target_path = resolve_secure_path(sub_path)
|
|
if not target_path.is_dir():
|
|
raise HTTPException(status_code=400, detail="대상 경로가 폴더가 아닙니다.")
|
|
|
|
folder_name = target_path.name or "downloads"
|
|
zip_filename = f"{folder_name}.zip"
|
|
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp:
|
|
tmp_path = tmp.name
|
|
|
|
try:
|
|
# ZIP_STORED: 이미 압축된 미디어 파일이 많으므로 CPU 낭비 없이 최고 속도로 패키징
|
|
with zipfile.ZipFile(tmp_path, "w", compression=zipfile.ZIP_STORED, allowZip64=True) as zf:
|
|
for root, dirs, files in os.walk(target_path):
|
|
for file in files:
|
|
file_path = Path(root) / file
|
|
# 상대 경로를 아카이브 내부 경로로 설정
|
|
arcname = file_path.relative_to(target_path)
|
|
try:
|
|
zf.write(file_path, arcname=str(arcname))
|
|
except Exception:
|
|
continue
|
|
return tmp_path, zip_filename
|
|
except Exception as e:
|
|
if os.path.exists(tmp_path):
|
|
os.unlink(tmp_path)
|
|
raise HTTPException(status_code=500, detail=f"ZIP 생성 실패: {str(e)}")
|
|
|
|
def create_batch_zip(sub_paths: List[str], custom_name: Optional[str] = "bundle.zip") -> Tuple[str, str]:
|
|
"""선택된 여러 파일/폴더를 하나의 zip으로 묶어 임시 파일 생성"""
|
|
if not sub_paths:
|
|
raise HTTPException(status_code=400, detail="다운로드할 파일이 선택되지 않았습니다.")
|
|
|
|
base_dir = Path(settings.DOWNLOADS_BASE_PATH).resolve()
|
|
zip_filename = custom_name.strip() if custom_name and custom_name.strip() else "bundle.zip"
|
|
if not zip_filename.lower().endswith(".zip"):
|
|
zip_filename += ".zip"
|
|
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp:
|
|
tmp_path = tmp.name
|
|
|
|
try:
|
|
with zipfile.ZipFile(tmp_path, "w", compression=zipfile.ZIP_STORED, allowZip64=True) as zf:
|
|
for sp in sub_paths:
|
|
target = resolve_secure_path(sp)
|
|
if target.is_dir():
|
|
for root, dirs, files in os.walk(target):
|
|
for file in files:
|
|
file_path = Path(root) / file
|
|
# 선택된 폴더명을 최상위로 유지
|
|
rel_to_target = file_path.relative_to(target)
|
|
arcname = Path(target.name) / rel_to_target
|
|
try:
|
|
zf.write(file_path, arcname=str(arcname))
|
|
except Exception:
|
|
continue
|
|
elif target.is_file():
|
|
try:
|
|
zf.write(target, arcname=target.name)
|
|
except Exception:
|
|
continue
|
|
return tmp_path, zip_filename
|
|
except Exception as e:
|
|
if os.path.exists(tmp_path):
|
|
os.unlink(tmp_path)
|
|
raise HTTPException(status_code=500, detail=f"선택 항목 ZIP 생성 실패: {str(e)}")
|
|
|
|
|
|
|
|
|