import os import json import librosa import numpy as np from typing import Optional def analyze_audio(file_path: str) -> dict: """ Phân tích âm thanh: BPM, beat tracking, ước lượng bars. """ # Load audio y, sr = librosa.load(file_path, sr=None) # Track beats tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr) # Handle tempo which might be scalar or numpy array in different librosa versions if isinstance(tempo, np.ndarray): if tempo.size > 0: bpm = float(tempo[0]) else: bpm = 120.0 else: bpm = float(tempo) # Convert frames to time (seconds) beat_times = librosa.frames_to_time(beat_frames, sr=sr).tolist() # Estimate bars (assume 4/4 time signature - grouping every 4 beats) bar_times = [beat_times[i] for i in range(0, len(beat_times), 4)] return { "bpm": round(bpm, 2), "beats": [round(t, 4) for t in beat_times], "bars": [round(t, 4) for t in bar_times], "duration": round(float(len(y)) / sr, 4) } def analyze_audio_advanced(file_path: str) -> dict: """ Phân tích âm thanh nâng cao: BPM, beats, bars, spectral features. Sử dụng librosa để trích xuất đặc trưng âm học chi tiết. """ y, sr = librosa.load(file_path, sr=None) duration = float(len(y)) / sr # Beat tracking tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr) if isinstance(tempo, np.ndarray): bpm = float(tempo[0]) if tempo.size > 0 else 120.0 else: bpm = float(tempo) beat_times = librosa.frames_to_time(beat_frames, sr=sr).tolist() bar_times = [beat_times[i] for i in range(0, len(beat_times), 4)] # Spectral centroid (brightness) spectral_centroids = librosa.feature.spectral_centroid(y=y, sr=sr)[0] avg_brightness = float(np.mean(spectral_centroids)) # RMS energy rms = librosa.feature.rms(y=y)[0] avg_energy = float(np.mean(rms)) # Zero crossing rate zcr = librosa.feature.zero_crossing_rate(y)[0] avg_zcr = float(np.mean(zcr)) return { "bpm": round(bpm, 2), "beats": [round(t, 4) for t in beat_times], "bars": [round(t, 4) for t in bar_times], "duration": round(duration, 4), "spectral_centroid_avg": round(avg_brightness, 2), "rms_energy_avg": round(avg_energy, 6), "zero_crossing_rate_avg": round(avg_zcr, 6), "sample_rate": sr } def analyze_structure_with_ai( file_path: str, api_base_url: Optional[str] = None, model: str = "deepseek-chat" ) -> dict: """ Phân tích cấu trúc khuôn nhạc bằng AI (OpenAI Compatible API). Gọi API DeepSeek/Ollama để phân đoạn bố cục (Intro, Verse, Chorus, Outro). Args: file_path: Đường dẫn tệp âm thanh api_base_url: Base URL của API (mặc định dùng env OPENAI_API_BASE) model: Model name (DeepSeek, Ollama, etc.) Note: API key is read exclusively from OPENAI_API_KEY env var. Returns: dict: Kết quả phân tích cấu trúc """ # Lấy thông tin phân tích cơ bản trước analysis = analyze_audio_advanced(file_path) # Cấu hình API - key chỉ đọc từ env var, không bao giờ truyền qua task queue base_url = api_base_url or os.getenv("OPENAI_API_BASE", "http://localhost:11434/v1") key = os.getenv("OPENAI_API_KEY", "ollama") # Chuẩn bị prompt phân tích prompt = f"""Analyze the following audio metadata and suggest the musical structure (sections). Audio Analysis: - BPM: {analysis['bpm']} - Duration: {analysis['duration']} seconds - Number of beats: {len(analysis['beats'])} - Number of bars: {len(analysis['bars'])} - Average spectral centroid: {analysis['spectral_centroid_avg']} Hz - Average RMS energy: {analysis['rms_energy_avg']} - Bar timestamps (seconds): {json.dumps(analysis['bars'][:20])} Based on this data, estimate the song structure by identifying sections like Intro, Verse, Chorus, Bridge, Outro. Respond ONLY with valid JSON in this exact format: {{ "sections": [ {{"name": "Intro", "start_time": 0.0, "end_time": 8.5}}, {{"name": "Verse", "start_time": 8.5, "end_time": 25.0}}, {{"name": "Chorus", "start_time": 25.0, "end_time": 40.0}} ] }}""" try: import httpx response = httpx.post( f"{base_url}/chat/completions", headers={ "Authorization": f"Bearer {key}", "Content-Type": "application/json" }, json={ "model": model, "messages": [ {"role": "system", "content": "You are a music analysis AI. Respond only with valid JSON."}, {"role": "user", "content": prompt} ], "temperature": 0.3 }, timeout=30.0 ) if response.status_code == 200: result = response.json() ai_content = result["choices"][0]["message"]["content"] # Parse JSON response try: # Xử lý trường hợp AI trả về text bọc trong ```json ... ``` if "```json" in ai_content: ai_content = ai_content.split("```json")[1].split("```")[0] elif "```" in ai_content: ai_content = ai_content.split("```")[1].split("```")[0] sections = json.loads(ai_content.strip()) except json.JSONDecodeError: sections = {"sections": [], "error": "AI response was not valid JSON"} return { **analysis, "ai_structure": sections, "ai_model": model, "ai_status": "success" } else: return { **analysis, "ai_structure": {"sections": []}, "ai_model": model, "ai_status": f"API error: {response.status_code}" } except ImportError: # httpx not available, fall back to basic heuristic return { **analysis, "ai_structure": _estimate_structure_heuristic(analysis), "ai_model": "heuristic", "ai_status": "httpx not installed, using heuristic" } except Exception as e: # API không khả dụng, dùng heuristic return { **analysis, "ai_structure": _estimate_structure_heuristic(analysis), "ai_model": "heuristic", "ai_status": f"AI unavailable ({str(e)}), using heuristic" } def _estimate_structure_heuristic(analysis: dict) -> dict: """ Ước lượng cấu trúc bài hát bằng heuristic khi AI không khả dụng. Dựa trên số bars và duration để phân đoạn. """ duration = analysis.get("duration", 0) bars = analysis.get("bars", []) if not bars or duration < 10: return {"sections": [{"name": "Full", "start_time": 0.0, "end_time": duration}]} sections = [] num_bars = len(bars) if num_bars >= 8: # Intro: ~first 2 bars intro_end = bars[min(2, num_bars - 1)] sections.append({"name": "Intro", "start_time": 0.0, "end_time": round(intro_end, 4)}) # Main body if num_bars >= 16: verse_end = bars[min(8, num_bars - 1)] sections.append({"name": "Verse", "start_time": round(intro_end, 4), "end_time": round(verse_end, 4)}) if num_bars >= 24: chorus_end = bars[min(16, num_bars - 1)] sections.append({"name": "Chorus", "start_time": round(verse_end, 4), "end_time": round(chorus_end, 4)}) # Outro: last bars to end outro_start = bars[min(num_bars - 2, num_bars - 1)] sections.append({"name": "Outro", "start_time": round(outro_start, 4), "end_time": round(duration, 4)}) else: sections.append({"name": "Outro", "start_time": round(verse_end, 4), "end_time": round(duration, 4)}) else: sections.append({"name": "Main", "start_time": round(intro_end, 4), "end_time": round(duration, 4)}) else: sections.append({"name": "Full", "start_time": 0.0, "end_time": round(duration, 4)}) return {"sections": sections}