306 lines
9.9 KiB
Python
306 lines
9.9 KiB
Python
import os
|
|
import uuid
|
|
import time
|
|
import glob
|
|
import logging
|
|
from celery import Celery
|
|
from app.config import settings
|
|
from app.core.analyzer import analyze_audio, analyze_structure_with_ai
|
|
from app.core.audio_editor import (
|
|
edit_audio_file, cut_and_loop_segment, mix_multitrack_session, export_audio
|
|
)
|
|
from app.core.dsp_utils import find_nearest_zero_crossing_file
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
celery_app = Celery(
|
|
"audio_tasks",
|
|
broker=settings.CELERY_BROKER_URL,
|
|
backend=settings.CELERY_RESULT_BACKEND
|
|
)
|
|
|
|
celery_app.conf.update(
|
|
task_serializer="json",
|
|
accept_content=["json"],
|
|
result_serializer="json",
|
|
timezone="UTC",
|
|
enable_utc=True,
|
|
)
|
|
|
|
# ── Lịch trình tự động dọn dẹp file hết hạn (Week 5) ──
|
|
celery_app.conf.beat_schedule = {
|
|
"cleanup-expired-files-every-hour": {
|
|
"task": "app.tasks.worker.cleanup_expired_files_task",
|
|
"schedule": 3600.0, # Chạy mỗi giờ
|
|
},
|
|
}
|
|
|
|
|
|
@celery_app.task
|
|
def analyze_audio_task(file_id: str):
|
|
file_path = os.path.join(settings.UPLOADS_DIR, file_id)
|
|
if not os.path.exists(file_path):
|
|
raise FileNotFoundError(f"Upload file not found: {file_id}")
|
|
return analyze_audio(file_path)
|
|
|
|
|
|
@celery_app.task
|
|
def analyze_ai_task(file_id: str, api_base_url: str = None,
|
|
model: str = "deepseek-chat"):
|
|
"""
|
|
Task phân tích cấu trúc khuôn nhạc bằng AI (Week 4).
|
|
API key is read from OPENAI_API_KEY env var only (never serialized into task queue).
|
|
"""
|
|
upload_path = os.path.join(settings.UPLOADS_DIR, file_id)
|
|
processed_path = os.path.join(settings.PROCESSED_DIR, file_id)
|
|
|
|
if os.path.exists(processed_path):
|
|
file_path = processed_path
|
|
elif os.path.exists(upload_path):
|
|
file_path = upload_path
|
|
else:
|
|
raise FileNotFoundError(f"File not found: {file_id}")
|
|
|
|
return analyze_structure_with_ai(
|
|
file_path=file_path,
|
|
api_base_url=api_base_url,
|
|
model=model
|
|
)
|
|
|
|
|
|
@celery_app.task
|
|
def edit_audio_task(config: dict):
|
|
file_id = config.get("file_id")
|
|
|
|
input_path = os.path.join(settings.UPLOADS_DIR, file_id)
|
|
if not os.path.exists(input_path):
|
|
input_path = os.path.join(settings.PROCESSED_DIR, file_id)
|
|
if not os.path.exists(input_path):
|
|
raise FileNotFoundError(f"Source file not found: {file_id}")
|
|
|
|
output_path = os.path.join(settings.PROCESSED_DIR, file_id)
|
|
|
|
res = edit_audio_file(config, input_path, output_path)
|
|
return {
|
|
"file_id": file_id,
|
|
"success": True,
|
|
"details": res
|
|
}
|
|
|
|
|
|
@celery_app.task
|
|
def export_audio_task(file_id: str, format: str = "wav",
|
|
sample_rate: int = 44100, bit_depth: int = 16):
|
|
"""
|
|
Task xuất tệp âm thanh sang nhiều định dạng (WAV/MP3/OGG).
|
|
"""
|
|
upload_path = os.path.join(settings.UPLOADS_DIR, file_id)
|
|
processed_path = os.path.join(settings.PROCESSED_DIR, file_id)
|
|
|
|
if os.path.exists(processed_path):
|
|
source_path = processed_path
|
|
elif os.path.exists(upload_path):
|
|
source_path = upload_path
|
|
else:
|
|
raise FileNotFoundError(f"File not found: {file_id}")
|
|
|
|
# Tạo output filename
|
|
base_name = os.path.splitext(file_id)[0]
|
|
output_filename = f"{base_name}_exported.{format}"
|
|
output_path = os.path.join(settings.PROCESSED_DIR, output_filename)
|
|
|
|
result = export_audio(
|
|
input_path=source_path,
|
|
output_path=output_path,
|
|
format=format,
|
|
sample_rate=sample_rate,
|
|
bit_depth=bit_depth
|
|
)
|
|
|
|
result["output_file_id"] = output_filename
|
|
return result
|
|
|
|
|
|
@celery_app.task
|
|
def mix_multitrack_task(session_config: dict):
|
|
"""
|
|
Task xử lý hòa âm đa kênh (Multitrack Mixdown).
|
|
"""
|
|
session_id = session_config.get("session_id")
|
|
export_settings = session_config.get("export_settings", {})
|
|
tracks = session_config.get("tracks", [])
|
|
|
|
# Chuẩn bị metadata cho từng track
|
|
tracks_meta = []
|
|
for track in tracks:
|
|
if track.get("muted", False):
|
|
continue
|
|
|
|
file_id = track.get("file_id")
|
|
|
|
# Tìm file nguồn
|
|
upload_path = os.path.join(settings.UPLOADS_DIR, file_id)
|
|
processed_path = os.path.join(settings.PROCESSED_DIR, file_id)
|
|
|
|
if os.path.exists(processed_path):
|
|
file_path = processed_path
|
|
elif os.path.exists(upload_path):
|
|
file_path = upload_path
|
|
else:
|
|
raise FileNotFoundError(f"File not found: {file_id}")
|
|
|
|
tracks_meta.append({
|
|
"file_path": file_path,
|
|
"volume": track.get("volume", 1.0),
|
|
"muted": False
|
|
})
|
|
|
|
# Tạo tên file output
|
|
output_format = export_settings.get("format", "wav")
|
|
output_filename = f"{session_id}_mixed.{output_format}"
|
|
output_path = os.path.join(settings.PROCESSED_DIR, output_filename)
|
|
|
|
# Thực hiện mix
|
|
result = mix_multitrack_session(
|
|
tracks_meta=tracks_meta,
|
|
output_path=output_path,
|
|
sample_rate=export_settings.get("sample_rate", 44100),
|
|
bit_depth=export_settings.get("bit_depth", 16)
|
|
)
|
|
|
|
result["output_file_id"] = output_filename
|
|
result["session_id"] = session_id
|
|
|
|
return result
|
|
|
|
|
|
@celery_app.task
|
|
def process_multitrack_session_task(session_config: dict):
|
|
"""
|
|
Task xử lý toàn bộ session với nhiều tracks và clips.
|
|
Xử lý từng clip (với zero-crossing alignment), sau đó hòa âm tất cả tracks.
|
|
"""
|
|
session_id = session_config.get("session_id")
|
|
export_settings = session_config.get("export_settings", {})
|
|
tracks = session_config.get("tracks", [])
|
|
|
|
processed_tracks = []
|
|
|
|
# Xử lý từng track
|
|
for track in tracks:
|
|
if track.get("muted", False):
|
|
continue
|
|
|
|
track_id = track.get("track_id")
|
|
file_id = track.get("file_id")
|
|
clips = track.get("clips", [])
|
|
|
|
# Tìm file nguồn
|
|
upload_path = os.path.join(settings.UPLOADS_DIR, file_id)
|
|
processed_path = os.path.join(settings.PROCESSED_DIR, file_id)
|
|
|
|
if os.path.exists(processed_path):
|
|
source_path = processed_path
|
|
elif os.path.exists(upload_path):
|
|
source_path = upload_path
|
|
else:
|
|
raise FileNotFoundError(f"File not found: {file_id}")
|
|
|
|
# Xử lý clips nếu có
|
|
if clips:
|
|
clip = clips[0]
|
|
|
|
start_sec = clip.get("start_time_seconds")
|
|
end_sec = clip.get("end_time_seconds")
|
|
|
|
# Áp dụng zero-crossing alignment nếu được yêu cầu
|
|
if clip.get("apply_zero_crossing", True):
|
|
start_sec = find_nearest_zero_crossing_file(source_path, start_sec)
|
|
end_sec = find_nearest_zero_crossing_file(source_path, end_sec)
|
|
|
|
# Sử dụng hàm cut_and_loop_segment
|
|
processed_segment = cut_and_loop_segment(
|
|
file_path=source_path,
|
|
start_sec=start_sec,
|
|
end_sec=end_sec,
|
|
loop_count=clip.get("loop_count", 1),
|
|
fade_in_ms=clip.get("fade_in_ms", 150),
|
|
fade_out_ms=clip.get("fade_out_ms", 150),
|
|
volume_db_change=0.0
|
|
)
|
|
|
|
# Lưu segment đã xử lý
|
|
temp_filename = f"temp_{track_id}_{uuid.uuid4().hex[:8]}.wav"
|
|
temp_path = os.path.join(settings.PROCESSED_DIR, temp_filename)
|
|
processed_segment.export(temp_path, format="wav")
|
|
|
|
track_file_path = temp_path
|
|
else:
|
|
track_file_path = source_path
|
|
|
|
processed_tracks.append({
|
|
"file_path": track_file_path,
|
|
"volume": track.get("volume", 1.0),
|
|
"muted": False
|
|
})
|
|
|
|
# Hòa âm tất cả tracks
|
|
output_format = export_settings.get("format", "wav")
|
|
output_filename = f"{session_id}_final.{output_format}"
|
|
output_path = os.path.join(settings.PROCESSED_DIR, output_filename)
|
|
|
|
result = mix_multitrack_session(
|
|
tracks_meta=processed_tracks,
|
|
output_path=output_path,
|
|
sample_rate=export_settings.get("sample_rate", 44100),
|
|
bit_depth=export_settings.get("bit_depth", 16)
|
|
)
|
|
|
|
# Dọn dẹp các file tạm
|
|
for track in processed_tracks:
|
|
if "temp_" in os.path.basename(track["file_path"]):
|
|
try:
|
|
os.remove(track["file_path"])
|
|
except Exception as e:
|
|
logger.warning("Failed to remove temp file %s: %s", track["file_path"], e)
|
|
|
|
result["output_file_id"] = output_filename
|
|
result["session_id"] = session_id
|
|
|
|
return result
|
|
|
|
|
|
@celery_app.task
|
|
def cleanup_expired_files_task(max_age_hours: int = 24):
|
|
"""
|
|
Task tự động dọn dẹp các tệp kết xuất hết hạn (Week 5).
|
|
Xóa file trong thư mục processed cũ hơn max_age_hours giờ.
|
|
"""
|
|
now = time.time()
|
|
max_age_seconds = max_age_hours * 3600
|
|
|
|
cleaned_count = 0
|
|
cleaned_size = 0
|
|
|
|
for directory in [settings.PROCESSED_DIR]:
|
|
if not os.path.exists(directory):
|
|
continue
|
|
|
|
for filepath in glob.glob(os.path.join(directory, "*")):
|
|
if os.path.isfile(filepath):
|
|
file_age = now - os.path.getmtime(filepath)
|
|
if file_age > max_age_seconds:
|
|
file_size = os.path.getsize(filepath)
|
|
try:
|
|
os.remove(filepath)
|
|
cleaned_count += 1
|
|
cleaned_size += file_size
|
|
except Exception as e:
|
|
logger.warning("Failed to remove expired file %s: %s", filepath, e)
|
|
|
|
return {
|
|
"cleaned_files": cleaned_count,
|
|
"cleaned_size_mb": round(cleaned_size / (1024 * 1024), 2),
|
|
"max_age_hours": max_age_hours
|
|
}
|