240 lines
8.4 KiB
Python
240 lines
8.4 KiB
Python
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}
|