배포

backend/app/files.py

@@ -0,0 +1,462 @@
+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)}")
+
+
+
+