import os import aiofiles import httpx from typing import List from fastapi import APIRouter, UploadFile, File, HTTPException from pydantic import BaseModel from app.models.schemas import BGMInfo, BGMUploadResponse, TranscriptSegment from app.services.video_processor import get_audio_duration from app.services.bgm_provider import ( get_free_bgm_sources, search_freesound, download_freesound, search_and_download_bgm, BGMSearchResult, ) from app.services.bgm_recommender import ( recommend_bgm_for_script, get_preset_recommendation, BGMRecommendation, BGM_PRESETS, ) from app.services.default_bgm import ( initialize_default_bgm, get_default_bgm_list, check_default_bgm_status, DEFAULT_BGM_TRACKS, ) from app.config import settings router = APIRouter() class BGMDownloadRequest(BaseModel): """Request to download BGM from URL.""" url: str name: str class BGMRecommendRequest(BaseModel): """Request for BGM recommendation based on script.""" segments: List[dict] # TranscriptSegment as dict use_translated: bool = True class FreesoundSearchRequest(BaseModel): """Request to search Freesound.""" query: str min_duration: int = 10 max_duration: int = 180 page: int = 1 page_size: int = 15 commercial_only: bool = True # 상업적 사용 가능한 라이선스만 (CC0, CC-BY) class FreesoundDownloadRequest(BaseModel): """Request to download from Freesound.""" sound_id: str name: str # Custom name for the downloaded file class AutoBGMRequest(BaseModel): """Request for automatic BGM search and download.""" keywords: List[str] # Search keywords (from BGM recommendation) max_duration: int = 120 commercial_only: bool = True # 상업적 사용 가능한 라이선스만 @router.get("/", response_model=list[BGMInfo]) async def list_bgm(): """List all available BGM files.""" bgm_list = [] if not os.path.exists(settings.BGM_DIR): return bgm_list for filename in os.listdir(settings.BGM_DIR): if filename.endswith((".mp3", ".wav", ".m4a", ".ogg")): filepath = os.path.join(settings.BGM_DIR, filename) bgm_id = os.path.splitext(filename)[0] duration = await get_audio_duration(filepath) bgm_list.append(BGMInfo( id=bgm_id, name=bgm_id.replace("_", " ").replace("-", " ").title(), duration=duration or 0, path=f"/static/bgm/{filename}" )) return bgm_list @router.post("/upload", response_model=BGMUploadResponse) async def upload_bgm( file: UploadFile = File(...), name: str | None = None ): """Upload a new BGM file.""" if not file.filename: raise HTTPException(status_code=400, detail="No filename provided") # Validate file type allowed_extensions = (".mp3", ".wav", ".m4a", ".ogg") if not file.filename.lower().endswith(allowed_extensions): raise HTTPException( status_code=400, detail=f"Invalid file type. Allowed: {allowed_extensions}" ) # Generate ID bgm_id = name or os.path.splitext(file.filename)[0] bgm_id = bgm_id.lower().replace(" ", "_") # Get extension ext = os.path.splitext(file.filename)[1].lower() filepath = os.path.join(settings.BGM_DIR, f"{bgm_id}{ext}") # Save file os.makedirs(settings.BGM_DIR, exist_ok=True) async with aiofiles.open(filepath, 'wb') as out_file: content = await file.read() await out_file.write(content) return BGMUploadResponse( id=bgm_id, name=name or file.filename, message="BGM uploaded successfully" ) @router.delete("/{bgm_id}") async def delete_bgm(bgm_id: str): """Delete a BGM file.""" for ext in (".mp3", ".wav", ".m4a", ".ogg"): filepath = os.path.join(settings.BGM_DIR, f"{bgm_id}{ext}") if os.path.exists(filepath): os.remove(filepath) return {"message": f"BGM '{bgm_id}' deleted"} raise HTTPException(status_code=404, detail="BGM not found") @router.get("/sources/free", response_model=dict) async def get_free_sources(): """Get list of recommended free BGM sources for commercial use.""" sources = get_free_bgm_sources() return { "sources": sources, "notice": "이 소스들은 상업적 사용이 가능한 무료 음악을 제공합니다. 각 사이트의 라이선스를 확인하세요.", "recommended": [ { "name": "Pixabay Music", "url": "https://pixabay.com/music/search/", "why": "CC0 라이선스, 저작권 표기 불필요, 쇼츠용 짧은 트랙 많음", "search_tips": ["upbeat", "energetic", "chill", "cinematic", "funny"], }, { "name": "Mixkit", "url": "https://mixkit.co/free-stock-music/", "why": "고품질, 카테고리별 정리, 상업적 무료 사용", "search_tips": ["short", "intro", "background"], }, ], } @router.post("/download-url", response_model=BGMUploadResponse) async def download_bgm_from_url(request: BGMDownloadRequest): """ Download BGM from external URL (Pixabay, Mixkit, etc.) Use this to download free BGM files directly from their source URLs. """ url = request.url name = request.name.lower().replace(" ", "_") # Validate URL - allow trusted free music sources allowed_domains = [ "pixabay.com", "cdn.pixabay.com", "mixkit.co", "assets.mixkit.co", "uppbeat.io", "freemusicarchive.org", ] from urllib.parse import urlparse parsed = urlparse(url) domain = parsed.netloc.lower() if not any(allowed in domain for allowed in allowed_domains): raise HTTPException( status_code=400, detail=f"URL must be from allowed sources: {', '.join(allowed_domains)}" ) try: async with httpx.AsyncClient(follow_redirects=True) as client: response = await client.get(url, timeout=60) if response.status_code != 200: raise HTTPException( status_code=400, detail=f"Failed to download: HTTP {response.status_code}" ) # Determine file extension content_type = response.headers.get("content-type", "") if "mpeg" in content_type or url.endswith(".mp3"): ext = ".mp3" elif "wav" in content_type or url.endswith(".wav"): ext = ".wav" elif "ogg" in content_type or url.endswith(".ogg"): ext = ".ogg" else: ext = ".mp3" # Save file os.makedirs(settings.BGM_DIR, exist_ok=True) filepath = os.path.join(settings.BGM_DIR, f"{name}{ext}") async with aiofiles.open(filepath, 'wb') as f: await f.write(response.content) return BGMUploadResponse( id=name, name=request.name, message=f"BGM downloaded from {domain}" ) except httpx.TimeoutException: raise HTTPException(status_code=408, detail="Download timed out") except Exception as e: raise HTTPException(status_code=500, detail=f"Download failed: {str(e)}") @router.post("/recommend") async def recommend_bgm(request: BGMRecommendRequest): """ AI-powered BGM recommendation based on script content. Analyzes the script mood and suggests appropriate background music. Returns matched BGM from library if available, otherwise provides search keywords. """ # Convert dict to TranscriptSegment segments = [TranscriptSegment(**seg) for seg in request.segments] # Get available BGM list available_bgm = [] if os.path.exists(settings.BGM_DIR): for filename in os.listdir(settings.BGM_DIR): if filename.endswith((".mp3", ".wav", ".m4a", ".ogg")): bgm_id = os.path.splitext(filename)[0] available_bgm.append({ "id": bgm_id, "name": bgm_id.replace("_", " ").replace("-", " "), }) # Get recommendation success, message, recommendation = await recommend_bgm_for_script( segments, available_bgm, use_translated=request.use_translated, ) if not success: raise HTTPException(status_code=500, detail=message) # Build Pixabay search URL search_keywords = "+".join(recommendation.search_keywords[:3]) pixabay_url = f"https://pixabay.com/music/search/{search_keywords}/" return { "recommendation": { "mood": recommendation.mood, "energy": recommendation.energy, "reasoning": recommendation.reasoning, "suggested_genres": recommendation.suggested_genres, "search_keywords": recommendation.search_keywords, }, "matched_bgm": recommendation.matched_bgm_id, "search_urls": { "pixabay": pixabay_url, "mixkit": f"https://mixkit.co/free-stock-music/?q={search_keywords}", }, "message": message, } @router.get("/recommend/presets") async def get_bgm_presets(): """ Get predefined BGM presets for common content types. Use these presets for quick BGM selection without AI analysis. """ presets = {} for content_type, preset_info in BGM_PRESETS.items(): presets[content_type] = { "mood": preset_info["mood"], "keywords": preset_info["keywords"], "description": f"Best for {content_type} content", } return { "presets": presets, "usage": "Use content_type parameter with /recommend/preset/{content_type}", } @router.get("/recommend/preset/{content_type}") async def get_preset_bgm(content_type: str): """ Get BGM recommendation for a specific content type. Available types: cooking, fitness, tutorial, comedy, travel, asmr, news, gaming """ recommendation = get_preset_recommendation(content_type) if not recommendation: available_types = list(BGM_PRESETS.keys()) raise HTTPException( status_code=404, detail=f"Unknown content type. Available: {', '.join(available_types)}" ) # Check for matching BGM in library if os.path.exists(settings.BGM_DIR): for filename in os.listdir(settings.BGM_DIR): if filename.endswith((".mp3", ".wav", ".m4a", ".ogg")): bgm_id = os.path.splitext(filename)[0] bgm_name = bgm_id.lower() # Check if any keyword matches for keyword in recommendation.search_keywords: if keyword in bgm_name: recommendation.matched_bgm_id = bgm_id break if recommendation.matched_bgm_id: break search_keywords = "+".join(recommendation.search_keywords[:3]) return { "content_type": content_type, "recommendation": { "mood": recommendation.mood, "energy": recommendation.energy, "suggested_genres": recommendation.suggested_genres, "search_keywords": recommendation.search_keywords, }, "matched_bgm": recommendation.matched_bgm_id, "search_urls": { "pixabay": f"https://pixabay.com/music/search/{search_keywords}/", "mixkit": f"https://mixkit.co/free-stock-music/?q={search_keywords}", }, } @router.post("/freesound/search") async def search_freesound_api(request: FreesoundSearchRequest): """ Search for music on Freesound. Freesound API provides 500,000+ CC licensed sounds. Get your API key at: https://freesound.org/apiv2/apply Set commercial_only=true (default) to only return CC0 licensed sounds that can be used commercially without attribution. """ success, message, results = await search_freesound( query=request.query, min_duration=request.min_duration, max_duration=request.max_duration, page=request.page, page_size=request.page_size, commercial_only=request.commercial_only, ) if not success: raise HTTPException(status_code=400, detail=message) def is_commercial_ok(license_str: str) -> bool: return "CC0" in license_str or license_str == "CC BY (Attribution)" return { "message": message, "commercial_only": request.commercial_only, "results": [ { "id": r.id, "title": r.title, "duration": r.duration, "tags": r.tags, "license": r.license, "commercial_use_ok": is_commercial_ok(r.license), "preview_url": r.preview_url, "source": r.source, } for r in results ], "search_url": f"https://freesound.org/search/?q={request.query}", } @router.post("/freesound/download", response_model=BGMUploadResponse) async def download_freesound_api(request: FreesoundDownloadRequest): """ Download a sound from Freesound by ID. Downloads the high-quality preview (128kbps MP3). """ name = request.name.lower().replace(" ", "_") name = "".join(c for c in name if c.isalnum() or c == "_") success, message, file_path = await download_freesound( sound_id=request.sound_id, output_dir=settings.BGM_DIR, filename=name, ) if not success: raise HTTPException(status_code=400, detail=message) return BGMUploadResponse( id=name, name=request.name, message=message, ) @router.post("/auto-download") async def auto_download_bgm(request: AutoBGMRequest): """ Automatically search and download BGM based on keywords. Use this with keywords from /recommend endpoint to auto-download matching BGM. Requires FREESOUND_API_KEY to be configured. Set commercial_only=true (default) to only download CC0 licensed sounds that can be used commercially without attribution. """ success, message, file_path, matched_result = await search_and_download_bgm( keywords=request.keywords, output_dir=settings.BGM_DIR, max_duration=request.max_duration, commercial_only=request.commercial_only, ) if not success: return { "success": False, "message": message, "downloaded": None, "suggestion": "Configure FREESOUND_API_KEY or manually download from Pixabay/Mixkit", } # Get duration of downloaded file duration = 0 if file_path: duration = await get_audio_duration(file_path) or 0 # Check if license is commercially usable license_name = matched_result.license if matched_result else "" commercial_ok = "CC0" in license_name or license_name == "CC BY (Attribution)" return { "success": True, "message": message, "downloaded": { "id": os.path.splitext(os.path.basename(file_path))[0] if file_path else None, "name": matched_result.title if matched_result else None, "duration": duration, "license": license_name, "commercial_use_ok": commercial_ok, "source": "freesound", "path": f"/static/bgm/{os.path.basename(file_path)}" if file_path else None, }, "original": { "freesound_id": matched_result.id if matched_result else None, "tags": matched_result.tags if matched_result else [], }, } @router.get("/defaults/status") async def get_default_bgm_status(): """ Check status of default BGM tracks. Returns which default tracks are installed and which are missing. """ status = check_default_bgm_status(settings.BGM_DIR) # Add track details tracks = [] for track in DEFAULT_BGM_TRACKS: installed = track.id in status["installed_ids"] tracks.append({ "id": track.id, "name": track.name, "category": track.category, "description": track.description, "installed": installed, }) return { "total": status["total"], "installed": status["installed"], "missing": status["missing"], "tracks": tracks, } @router.post("/defaults/initialize") async def initialize_default_bgms(force: bool = False): """ Download default BGM tracks. Downloads pre-selected royalty-free BGM tracks (Pixabay License). Use force=true to re-download all tracks. These tracks are free for commercial use without attribution. """ downloaded, skipped, errors = await initialize_default_bgm( settings.BGM_DIR, force=force, ) return { "success": len(errors) == 0, "downloaded": downloaded, "skipped": skipped, "errors": errors, "message": f"Downloaded {downloaded} tracks, skipped {skipped} existing" if downloaded > 0 else "All default tracks already installed" if skipped > 0 else "Failed to download tracks", } @router.get("/defaults/list") async def list_default_bgms(): """ Get list of available default BGM tracks with metadata. Returns information about all pre-configured default tracks. """ tracks = await get_default_bgm_list() status = check_default_bgm_status(settings.BGM_DIR) for track in tracks: track["installed"] = track["id"] in status["installed_ids"] return { "tracks": tracks, "total": len(tracks), "installed": status["installed"], "license": "Pixabay License (Free for commercial use, no attribution required)", } @router.get("/{bgm_id}") async def get_bgm(bgm_id: str): """Get BGM info by ID.""" for ext in (".mp3", ".wav", ".m4a", ".ogg"): filepath = os.path.join(settings.BGM_DIR, f"{bgm_id}{ext}") if os.path.exists(filepath): duration = await get_audio_duration(filepath) return BGMInfo( id=bgm_id, name=bgm_id.replace("_", " ").replace("-", " ").title(), duration=duration or 0, path=f"/static/bgm/{bgm_id}{ext}" ) raise HTTPException(status_code=404, detail="BGM not found")