SonicForgeStudio/app/api/v1/media.py

import os
import platform
from typing import List, Optional

from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import FileResponse

router = APIRouter()

MEDIA_EXTS = {
    ".wav", ".mp3", ".ogg", ".flac", ".aiff", ".aif", ".m4a", ".aac", ".opus",
    ".mid", ".midi"
}

AUDIO_EXTS = {".wav", ".mp3", ".ogg", ".flac", ".aiff", ".aif", ".m4a", ".aac", ".opus"}
MIDI_EXTS = {".mid", ".midi"}


def _safe_path(path: str) -> str:
    if not path:
        raise HTTPException(status_code=400, detail="Thiếu path")
    if "\x00" in path:
        raise HTTPException(status_code=400, detail="Path không hợp lệ")
    return os.path.normpath(path)


REAL_FS_TYPES = {
    "ext2", "ext3", "ext4", "xfs", "btrfs", "jfs", "reiserfs",
    "ntfs", "ntfs3", "vfat", "exfat", "fat", "hfs", "hfsplus", "apfs",
    "zfs", "f2fs", "iso9660", "udf", "nfs", "nfs4", "cifs", "smb3", "fuseblk",
}

PSEUDO_FS_TYPES = {
    "proc", "sysfs", "devpts", "tmpfs", "devtmpfs", "overlay", "squashfs",
    "cgroup", "cgroup2", "pstore", "securityfs", "debugfs", "tracefs",
    "configfs", "fusectl", "hugetlbfs", "mqueue", "binfmt_misc", "nsfs",
    "autofs", "ramfs", "efivarfs", "rpc_pipefs", "fuse", "fusefs",
}


@router.get("/computer")
async def list_computer_roots():
    """Liệt kê các ổ đĩa / mount point thật của máy (My Computer)."""
    system = platform.system()
    roots = []
    if system == "Windows":
        import string
        for drive in string.ascii_uppercase:
            root = drive + ":\\"
            try:
                if os.path.exists(root):
                    roots.append({"path": root, "name": drive + ":", "is_dir": True})
            except OSError:
                continue
    else:
        # Unix/Linux/macOS: chỉ liệt kê filesystem thật, bỏ pseudo/docker/systemd mounts
        seen = set()
        try:
            with open("/proc/mounts", "r") as f:
                for line in f:
                    parts = line.split()
                    if len(parts) < 3:
                        continue
                    device, mount, fstype = parts[0], parts[1], parts[2]
                    if fstype in PSEUDO_FS_TYPES:
                        continue
                    if fstype not in REAL_FS_TYPES:
                        # giữ mount point root "/" nếu không thuộc pseudo
                        if mount != "/":
                            continue
                    if mount in seen:
                        continue
                    seen.add(mount)
                    # lọc mount point rác kiểu /run/credentials/...
                    if mount.startswith("/run/") or mount.startswith("/var/lib/docker"):
                        continue
                    try:
                        if os.path.isdir(mount):
                            label = mount if mount != "/" else "Root (/)"
                            roots.append({"path": mount, "name": label, "is_dir": True})
                    except OSError:
                        pass
        except OSError:
            pass
        # macOS fallback: liệt kê /Volumes
        if system == "Darwin":
            try:
                for name in sorted(os.listdir("/Volumes")):
                    full = os.path.join("/Volumes", name)
                    if os.path.isdir(full):
                        roots.append({"path": full, "name": name, "is_dir": True})
            except OSError:
                pass
        if not roots:
            roots = [{"path": "/", "name": "Root (/)", "is_dir": True}]
    return {"system": system, "roots": roots}


@router.get("/browse")
async def browse_directory(path: str = Query(...)):
    """Liệt kê nội dung một thư mục trên máy: thư mục con + file audio/MIDI."""
    resolved = _safe_path(path)
    if not os.path.isdir(resolved):
        raise HTTPException(status_code=404, detail="Không tìm thấy thư mục")

    dirs, files = [], []
    try:
        entries = os.listdir(resolved)
    except OSError as e:
        raise HTTPException(status_code=403, detail=f"Không thể đọc thư mục: {e}")

    for name in entries:
        if name.startswith("."):
            continue
        full = os.path.join(resolved, name)
        try:
            if os.path.isdir(full):
                dirs.append({"name": name, "path": full, "is_dir": True})
            else:
                ext = os.path.splitext(name)[1].lower()
                if ext in MEDIA_EXTS:
                    size = os.path.getsize(full)
                    files.append({
                        "name": name,
                        "path": full,
                        "is_dir": False,
                        "size_mb": round(size / (1024 * 1024), 2),
                        "ext": ext,
                        "kind": "midi" if ext in MIDI_EXTS else "audio"
                    })
        except OSError:
            continue

    dirs.sort(key=lambda d: d["name"].lower())
    files.sort(key=lambda f: f["name"].lower())
    parent = os.path.dirname(resolved)
    return {
        "path": resolved,
        "parent": parent if parent != resolved else None,
        "dirs": dirs,
        "files": files
    }


@router.get("/file")
async def serve_local_file(path: str = Query(...)):
    """Phục vụ file audio/MIDI cục bộ để preview."""
    resolved = _safe_path(path)
    if not os.path.isfile(resolved):
        raise HTTPException(status_code=404, detail="Không tìm thấy file")
    ext = os.path.splitext(resolved)[1].lower()
    if ext not in MEDIA_EXTS:
        raise HTTPException(status_code=403, detail="Loại file không được hỗ trợ preview")
    media_type = "audio/wav" if ext in AUDIO_EXTS else "audio/midi"
    return FileResponse(resolved, media_type=media_type, filename=os.path.basename(resolved))