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) @router.get("/computer") async def list_computer_roots(): """Liệt kê các ổ đĩa / mount point của máy (My Computer).""" system = platform.system() roots = [] if system == "Windows": import string from ctypes import windll for drive in string.ascii_uppercase: root = drive + ":\\" if os.path.exists(root): roots.append({"path": root, "name": drive + ":", "is_dir": True}) else: # Unix/Linux/macOS: liệt kê mount points từ /proc/mounts seen = set() try: with open("/proc/mounts", "r") as f: for line in f: parts = line.split() if len(parts) < 2: continue mount = parts[1] if mount in seen: continue seen.add(mount) if mount.startswith("/dev") or mount.startswith("/sys") or mount.startswith("/proc"): continue try: if os.path.isdir(mount): roots.append({"path": mount, "name": mount, "is_dir": True}) except OSError: pass except OSError: pass if not roots: roots = [{"path": "/", "name": "/", "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))