119 lines
4.3 KiB
Python
119 lines
4.3 KiB
Python
"""Splitter service that calls the core audio_splitter logic."""
|
|
|
|
import os
|
|
import sys
|
|
import shutil
|
|
from pathlib import Path
|
|
from typing import List, Dict, Any
|
|
import time
|
|
|
|
from backend.config import settings
|
|
from backend.services.task_manager import task_manager
|
|
from backend.services.file_manager import FileManager
|
|
from backend.models.request import TracklistEntry
|
|
from backend.models.response import TaskStatus
|
|
|
|
|
|
def run_split_task(task_id: str, tracklist: List[TracklistEntry], options: Dict[str, Any]) -> None:
|
|
try:
|
|
task_manager.update_task_with_progress(
|
|
task_id, progress=5, message="Initializing..."
|
|
)
|
|
|
|
input_path = FileManager.get_input_path(task_id)
|
|
if not input_path or not input_path.exists():
|
|
raise RuntimeError(f"Input file not found for task {task_id}")
|
|
|
|
output_dir = FileManager.ensure_output_dir(task_id)
|
|
|
|
tracks = []
|
|
for entry in tracklist:
|
|
track_dict = {
|
|
"ts": entry.ts,
|
|
"tn": entry.tn or "",
|
|
"an": entry.an or "",
|
|
"al": entry.al or "",
|
|
"date": entry.date or "",
|
|
"ext": entry.ext or ""
|
|
}
|
|
tracks.append(track_dict)
|
|
|
|
from types import SimpleNamespace
|
|
|
|
# Build args namespace using the new fields
|
|
args = SimpleNamespace(
|
|
container=options.get("container"), # None -> auto-detect
|
|
audio_codec=options.get("audio_codec", "copy"),
|
|
video_codec=options.get("video_codec", "copy"),
|
|
subtitle_codec=options.get("subtitle_codec", "copy"),
|
|
video_quality=options.get("video_quality"),
|
|
drop_video=options.get("drop_video", False),
|
|
drop_subs=options.get("drop_subs", False),
|
|
number_tracks=options.get("number_tracks", False),
|
|
replace_bad_chars=options.get("replace_bad_chars", False),
|
|
replacement_char=options.get("replacement_char", "_"),
|
|
bad_chars=options.get("bad_chars", r'!@#№$;:%^&?*(){}[]\/<>+=~`\' '),
|
|
skip_existing=options.get("skip_existing", False),
|
|
output_template=options.get("output_template", "%an-%tn.%ext"),
|
|
album=options.get("album", None),
|
|
comment=options.get("comment", None),
|
|
no_comment=options.get("no_comment", False),
|
|
comment_stream=options.get("comment_stream", None),
|
|
merge_comments=options.get("merge_comments", False),
|
|
comment_separator=options.get("comment_separator", "; "),
|
|
delete_original=False,
|
|
)
|
|
|
|
project_root = Path(__file__).parent.parent.parent.parent
|
|
if str(project_root) not in sys.path:
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
from audio_splitter.core import split_audio
|
|
|
|
# Detect attached picture and extract if applicable
|
|
cover_image_path = None
|
|
if not args.drop_video:
|
|
from audio_splitter.ffmpeg import is_attached_picture, extract_cover_image, get_stream_info
|
|
input_path_str = str(input_path)
|
|
stream_info = get_stream_info(input_path_str)
|
|
if stream_info.get('has_video') and is_attached_picture(input_path_str):
|
|
cover_image_path = output_dir / 'cover.png'
|
|
if extract_cover_image(input_path_str, str(cover_image_path)):
|
|
print(f"Extracted cover image for task {task_id}")
|
|
else:
|
|
cover_image_path = None
|
|
|
|
# Attach cover image to args
|
|
args.cover_image = str(cover_image_path) if cover_image_path else None
|
|
|
|
task_manager.update_task_with_progress(
|
|
task_id, progress=10, message="Starting split..."
|
|
)
|
|
|
|
split_audio(str(input_path), str(output_dir), tracks, args)
|
|
|
|
output_files = FileManager.get_output_files(task_id)
|
|
|
|
task_manager.update_task_with_progress(
|
|
task_id,
|
|
progress=100,
|
|
message="Split complete",
|
|
status=TaskStatus.DONE
|
|
)
|
|
|
|
task_manager.update_task(
|
|
task_id,
|
|
tracks=output_files
|
|
)
|
|
|
|
time.sleep(0.5)
|
|
|
|
except Exception as e:
|
|
task_manager.update_task_with_progress(
|
|
task_id,
|
|
progress=0,
|
|
message="Split failed",
|
|
status=TaskStatus.ERROR,
|
|
error=str(e)
|
|
)
|