FEATURE: allows the backend to take a tracklis text file

This commit is contained in:
2026-08-19 17:26:07 +05:00
parent 607836b280
commit 52d01f1bcd
4 changed files with 149 additions and 10 deletions
+1 -2
View File
@@ -1,4 +1,3 @@
"""API route handlers.""" """API route handlers."""
from . import upload, split, status, download from . import upload, split, split_file, status, download, formats, info
# websocket is imported directly in main.py to avoid circular import
+140
View File
@@ -0,0 +1,140 @@
"""Endpoint for splitting with a tracklist file upload."""
import json
from pathlib import Path
from fastapi import APIRouter, UploadFile, File, Form, HTTPException, BackgroundTasks
from backend.config import settings
from backend.services.task_manager import task_manager
from backend.services.splitter import run_split_task
from backend.models.request import TracklistEntry
from backend.models.response import SplitResponse, TaskStatus
router = APIRouter(prefix="/api", tags=["split"])
# Import CLI tracklist parsing functions
from audio_splitter.tracklist import parse_format, read_tracklist
@router.post("/split-file", response_model=SplitResponse)
async def split_from_file(
background_tasks: BackgroundTasks,
task_id: str = Form(...),
tracklist_file: UploadFile = File(...),
options: str = Form("{}"),
tracklist_format: str = Form("%ts %tn - %an"),
):
"""
Start a splitting task using a tracklist file.
This endpoint accepts a plain text tracklist file (like the CLI does)
and parses it using the same logic.
Args:
task_id: Task ID from upload.
tracklist_file: Tracklist file (text/plain).
options: JSON string of all CLI options.
tracklist_format: Format string for parsing (default: "%ts %tn - %an").
Returns:
SplitResponse with task_id and status.
"""
# Validate task exists
if not task_manager.has_task(task_id):
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
# Check if task is already processing
status = task_manager.get_status(task_id)
if status and status["status"] in (TaskStatus.PROCESSING, TaskStatus.DONE):
raise HTTPException(
status_code=409,
detail=f"Task {task_id} is already {status['status']}"
)
# Validate file type
if not tracklist_file.filename.endswith(('.txt', '.text')):
raise HTTPException(
status_code=400,
detail="Tracklist file must be a text file (.txt or .text)"
)
# Read and parse the tracklist file
try:
content = await tracklist_file.read()
text = content.decode('utf-8')
except UnicodeDecodeError:
raise HTTPException(
status_code=400,
detail="Tracklist file must be UTF-8 encoded"
)
if not text.strip():
raise HTTPException(
status_code=400,
detail="Tracklist file is empty"
)
# Parse the tracklist using CLI logic
try:
# Parse the format string into tokens
tokens = parse_format(tracklist_format)
# Write the content to a temporary file (read_tracklist expects a file path)
temp_tracklist_path = settings.temp_dir / task_id / "tracklist.txt"
temp_tracklist_path.parent.mkdir(parents=True, exist_ok=True)
temp_tracklist_path.write_text(text, encoding='utf-8')
# Parse the tracklist file using CLI logic
tracklist_dicts = read_tracklist(str(temp_tracklist_path), tokens)
except Exception as e:
raise HTTPException(
status_code=400,
detail=f"Failed to parse tracklist: {str(e)}"
)
if not tracklist_dicts:
raise HTTPException(
status_code=400,
detail="No valid tracks found in tracklist file"
)
# Convert dicts to TracklistEntry objects (same as JSON endpoint)
try:
tracklist_entries = [TracklistEntry(**entry) for entry in tracklist_dicts]
except Exception as e:
raise HTTPException(
status_code=400,
detail=f"Invalid tracklist data: {str(e)}"
)
# Parse options JSON
try:
options_dict = json.loads(options)
except json.JSONDecodeError:
raise HTTPException(
status_code=400,
detail="Invalid JSON in options field"
)
# Update task status
task_manager.update_task(
task_id,
status=TaskStatus.PROCESSING,
progress=0,
message="Preparing to split..."
)
# Start background task
background_tasks.add_task(
run_split_task,
task_id,
tracklist_entries, # now TracklistEntry objects
options_dict
)
return SplitResponse(
task_id=task_id,
status=TaskStatus.PROCESSING
)
+4 -3
View File
@@ -5,7 +5,7 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from backend.config import settings from backend.config import settings
from backend.api import upload, split, status, download, websocket, formats, info from backend.api import upload, split, split_file, status, download, websocket, formats, info
from backend.services import progress_publisher from backend.services import progress_publisher
app = FastAPI( app = FastAPI(
@@ -33,11 +33,12 @@ async def startup_event():
# Include routers # Include routers
app.include_router(upload.router) app.include_router(upload.router)
app.include_router(split.router) app.include_router(split.router)
app.include_router(split_file.router) # NEW
app.include_router(status.router) app.include_router(status.router)
app.include_router(download.router) app.include_router(download.router)
app.include_router(websocket.router) app.include_router(websocket.router)
app.include_router(formats.router) # new app.include_router(formats.router)
app.include_router(info.router) # new app.include_router(info.router)
@app.get("/") @app.get("/")
+4 -5
View File
@@ -40,8 +40,10 @@ def run_split_task(task_id: str, tracklist: List[TracklistEntry], options: Dict[
from types import SimpleNamespace from types import SimpleNamespace
# Build args namespace from options.
# IMPORTANT: format is None if not specified -> core will auto-detect.
args = SimpleNamespace( args = SimpleNamespace(
format=options.get("format", "mp3"), format=options.get("format"),
transcode_to=options.get("transcode_to", None), transcode_to=options.get("transcode_to", None),
drop_video=options.get("drop_video", False), drop_video=options.get("drop_video", False),
drop_subs=options.get("drop_subs", False), drop_subs=options.get("drop_subs", False),
@@ -70,13 +72,10 @@ def run_split_task(task_id: str, tracklist: List[TracklistEntry], options: Dict[
task_id, progress=10, message="Starting split..." task_id, progress=10, message="Starting split..."
) )
# Run the split
split_audio(str(input_path), str(output_dir), tracks, args) split_audio(str(input_path), str(output_dir), tracks, args)
# Get output files
output_files = FileManager.get_output_files(task_id) output_files = FileManager.get_output_files(task_id)
# Final status update
task_manager.update_task_with_progress( task_manager.update_task_with_progress(
task_id, task_id,
progress=100, progress=100,
@@ -84,7 +83,7 @@ def run_split_task(task_id: str, tracklist: List[TracklistEntry], options: Dict[
status=TaskStatus.DONE status=TaskStatus.DONE
) )
# Add tracks to the task state # Add tracks to the task state (for download and status queries)
task_manager.update_task( task_manager.update_task(
task_id, task_id,
tracks=output_files tracks=output_files