Simple web backend added with upload/status/process/download API

web/backend/.env

@@ -0,0 +1,4 @@
+MAX_UPLOAD_SIZE_MB=500
+TEMP_DIR=/tmp/audio_splitter_web
+CLEANUP_AFTER_SECONDS=3600
+ALLOW_ORIGINS=http://localhost:5173,http://localhost:3000

web/backend/README.md

@@ -0,0 +1,8 @@
+# From the web/ directory
+cd web
+
+# Install dependencies
+pip install -r requirements-web.txt
+
+# Run the server (note the module path: backend.main)
+uvicorn backend.main:app --reload --port 8000

web/backend/__init__.py

@@ -0,0 +1 @@
+"""Audio Splitter Web Backend"""

web/backend/api/__init__.py

@@ -0,0 +1 @@
+"""API route handlers."""

web/backend/api/download.py

@@ -0,0 +1,61 @@
+"""Download endpoint for split results."""
+
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import FileResponse
+from zipfile import ZipFile
+import os
+
+from backend.config import settings
+from backend.services.task_manager import task_manager
+from backend.models.response import TaskStatus
+
+router = APIRouter(prefix="/api", tags=["download"])
+
+
+def _prepare_download(task_id: str):
+    """Common logic to prepare and return the ZIP file."""
+    if not task_manager.has_task(task_id):
+        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
+
+    status = task_manager.get_status(task_id)
+    if status["status"] != TaskStatus.DONE:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Task {task_id} is not complete. Current status: {status['status']}"
+        )
+
+    output_dir = settings.temp_dir / task_id / "output"
+    if not output_dir.exists() or not any(output_dir.iterdir()):
+        raise HTTPException(status_code=404, detail="No output files found")
+
+    zip_path = settings.temp_dir / task_id / "splits.zip"
+    # Create ZIP if not already exists (or overwrite)
+    with ZipFile(zip_path, "w") as zipf:
+        for file_path in output_dir.iterdir():
+            if file_path.is_file():
+                zipf.write(file_path, arcname=file_path.name)
+
+    return zip_path
+
+
+@router.get("/download/{task_id}")
+async def download_results(task_id: str):
+    zip_path = _prepare_download(task_id)
+    return FileResponse(
+        path=zip_path,
+        media_type="application/zip",
+        filename=f"{task_id}_splits.zip",
+        headers={"Content-Disposition": f"attachment; filename={task_id}_splits.zip"}
+    )
+
+
+@router.get("/download/{task_id}/splits.zip")
+async def download_results_as_zip(task_id: str):
+    """Alias endpoint that provides a .zip suffix for easier curl usage."""
+    zip_path = _prepare_download(task_id)
+    return FileResponse(
+        path=zip_path,
+        media_type="application/zip",
+        filename="splits.zip",
+        headers={"Content-Disposition": "attachment; filename=splits.zip"}
+    )

web/backend/api/split.py

@@ -0,0 +1,39 @@
+"""Split task endpoint."""
+
+from fastapi import APIRouter, HTTPException, BackgroundTasks
+
+from backend.models.request import SplitRequest
+from backend.models.response import SplitResponse, TaskStatus
+from backend.services.splitter import run_split_task
+from backend.services.task_manager import task_manager
+
+router = APIRouter(prefix="/api", tags=["split"])
+
+
+@router.post("/split", response_model=SplitResponse)
+async def start_split(request: SplitRequest, background_tasks: BackgroundTasks):
+    task_id = request.task_id
+
+    if not task_manager.has_task(task_id):
+        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
+
+    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']}"
+        )
+
+    task_manager.update_task(
+        task_id,
+        status=TaskStatus.PROCESSING,
+        progress=0,
+        message="Preparing to split..."
+    )
+
+    background_tasks.add_task(run_split_task, task_id, request.tracklist, request.options)
+
+    return SplitResponse(
+        task_id=task_id,
+        status=TaskStatus.PROCESSING
+    )

web/backend/api/status.py

@@ -0,0 +1,25 @@
+"""Status query endpoint."""
+
+from fastapi import APIRouter, HTTPException
+
+from backend.models.response import StatusResponse
+from backend.services.task_manager import task_manager
+
+router = APIRouter(prefix="/api", tags=["status"])
+
+
+@router.get("/status/{task_id}", response_model=StatusResponse)
+async def get_status(task_id: str):
+    if not task_manager.has_task(task_id):
+        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
+
+    status = task_manager.get_status(task_id)
+
+    return StatusResponse(
+        task_id=task_id,
+        status=status["status"],
+        progress=status["progress"],
+        message=status["message"],
+        error=status.get("error"),
+        tracks=status.get("tracks", [])
+    )

web/backend/api/upload.py

@@ -0,0 +1,58 @@
+"""File upload endpoint."""
+
+import uuid
+import shutil
+from pathlib import Path
+
+from fastapi import APIRouter, UploadFile, File, HTTPException
+
+from backend.config import settings
+from backend.models.response import UploadResponse
+from backend.services.task_manager import task_manager
+
+router = APIRouter(prefix="/api", tags=["upload"])
+
+ALLOWED_EXTENSIONS = {
+    ".mp3", ".flac", ".wav", ".m4a", ".ogg", ".opus",
+    ".aac", ".wma", ".aiff", ".alac", ".ac3"
+}
+
+
+@router.post("/upload", response_model=UploadResponse)
+async def upload_file(file: UploadFile = File(...)):
+    extension = Path(file.filename).suffix.lower()
+    if extension not in ALLOWED_EXTENSIONS:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Unsupported file format. Allowed: {', '.join(sorted(ALLOWED_EXTENSIONS))}"
+        )
+
+    task_id = str(uuid.uuid4())[:8]
+    task_dir = settings.temp_dir / task_id
+    task_dir.mkdir(parents=True, exist_ok=True)
+
+    input_path = task_dir / f"input{extension}"
+    try:
+        with open(input_path, "wb") as buffer:
+            shutil.copyfileobj(file.file, buffer)
+    except Exception as e:
+        shutil.rmtree(task_dir, ignore_errors=True)
+        raise HTTPException(status_code=500, detail=f"Failed to save file: {str(e)}")
+
+    file_size = input_path.stat().st_size
+    max_size_bytes = settings.max_upload_size_mb * 1024 * 1024
+    if file_size > max_size_bytes:
+        shutil.rmtree(task_dir, ignore_errors=True)
+        raise HTTPException(
+            status_code=400,
+            detail=f"File too large. Maximum size: {settings.max_upload_size_mb} MB"
+        )
+
+    # Create task entry in the task manager
+    task_manager.create_task(task_id, file.filename, file_size)
+
+    return UploadResponse(
+        task_id=task_id,
+        filename=file.filename,
+        size=file_size
+    )

web/backend/config.py

@@ -0,0 +1,33 @@
+"""Configuration settings for the web backend."""
+
+import os
+from pathlib import Path
+
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    """Application settings loaded from environment variables."""
+
+    # Server
+    host: str = "0.0.0.0"
+    port: int = 8000
+    debug: bool = False
+
+    # File handling
+    max_upload_size_mb: int = 500
+    temp_dir: Path = Path("/tmp/audio_splitter_web")
+    cleanup_after_seconds: int = 3600  # 1 hour
+
+    # CORS
+    allow_origins: list[str] = ["http://localhost:5173", "http://localhost:3000"]
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+
+
+settings = Settings()
+
+# Ensure temp directory exists
+settings.temp_dir.mkdir(parents=True, exist_ok=True)

web/backend/main.py

@@ -0,0 +1,38 @@
+"""FastAPI application entry point."""
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from backend.config import settings
+from backend.api import upload, split, status, download
+
+app = FastAPI(
+    title="Audio Splitter Web API",
+    description="Web interface for splitting audio files using a tracklist",
+    version="0.1.0",
+)
+
+# CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=settings.allow_origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Include routers
+app.include_router(upload.router)
+app.include_router(split.router)
+app.include_router(status.router)
+app.include_router(download.router)
+
+
+@app.get("/")
+async def root():
+    return {"status": "ok", "service": "Audio Splitter Web API"}
+
+
+@app.get("/health")
+async def health():
+    return {"status": "healthy"}

web/backend/models/__init__.py

@@ -0,0 +1 @@
+"""Pydantic models for request/response validation."""

web/backend/models/request.py

@@ -0,0 +1,58 @@
+"""Request models for API endpoints."""
+
+from typing import Optional, List
+
+from pydantic import BaseModel, Field, validator
+
+
+class TracklistEntry(BaseModel):
+    """A single track entry from the tracklist."""
+
+    ts: str = Field(..., description="Timestamp (e.g., '00:00' or '00:00-01:30')")
+    tn: Optional[str] = Field("", description="Track name")
+    an: Optional[str] = Field("", description="Author/artist")
+    al: Optional[str] = Field("", description="Album")
+    date: Optional[str] = Field("", description="Date/year")
+    ext: Optional[str] = Field("", description="File extension")
+
+    @validator("ts")
+    def validate_timestamp(cls, v: str) -> str:
+        """Basic timestamp validation (format and range)."""
+        v = v.strip()
+        if not v:
+            raise ValueError("Timestamp cannot be empty")
+
+        # Check for range format (start-end)
+        if "-" in v:
+            parts = v.split("-", 1)
+            start = parts[0].strip()
+            end = parts[1].strip()
+            if not start or not end:
+                raise ValueError("Invalid range format. Expected 'start-end'")
+            # Validate each part with the same logic
+            for ts in [start, end]:
+                cls._validate_single_timestamp(ts)
+        else:
+            cls._validate_single_timestamp(v)
+
+        return v
+
+    @staticmethod
+    def _validate_single_timestamp(ts: str) -> None:
+        """Validate a single timestamp (mm:ss or HH:MM:SS)."""
+        parts = ts.split(":")
+        if len(parts) not in (2, 3):
+            raise ValueError(f"Invalid timestamp format: {ts}. Expected mm:ss or HH:MM:SS")
+        try:
+            for p in parts:
+                int(p)
+        except ValueError:
+            raise ValueError(f"Invalid timestamp: {ts}. Must contain only numbers.")
+
+
+class SplitRequest(BaseModel):
+    """Request model for the split endpoint."""
+
+    task_id: str = Field(..., description="Task ID from upload")
+    tracklist: List[TracklistEntry] = Field(..., description="List of tracks")
+    options: dict = Field(default_factory=dict, description="All CLI options")

web/backend/models/response.py

@@ -0,0 +1,38 @@
+"""Response models for API endpoints."""
+
+from typing import Optional, List
+from enum import Enum
+
+from pydantic import BaseModel, Field
+
+
+class TaskStatus(str, Enum):
+    PENDING = "pending"
+    PROCESSING = "processing"
+    DONE = "done"
+    ERROR = "error"
+
+
+class TrackInfo(BaseModel):
+    filename: str
+    size: int  # bytes
+
+
+class UploadResponse(BaseModel):
+    task_id: str
+    filename: str
+    size: int
+
+
+class SplitResponse(BaseModel):
+    task_id: str
+    status: TaskStatus
+
+
+class StatusResponse(BaseModel):
+    task_id: str
+    status: TaskStatus
+    progress: int = Field(0, ge=0, le=100)
+    message: str = ""
+    error: Optional[str] = None
+    tracks: List[TrackInfo] = []

web/backend/services/__init__.py

@@ -0,0 +1 @@
+"""Business logic services."""

web/backend/services/file_manager.py

@@ -0,0 +1,47 @@
+"""Temporary file management for web operations."""
+
+import shutil
+from pathlib import Path
+
+from backend.config import settings
+
+
+class FileManager:
+    @staticmethod
+    def get_task_dir(task_id: str) -> Path:
+        return settings.temp_dir / task_id
+
+    @staticmethod
+    def get_input_path(task_id: str) -> Path:
+        task_dir = FileManager.get_task_dir(task_id)
+        for f in task_dir.iterdir():
+            if f.name.startswith("input."):
+                return f
+        return None
+
+    @staticmethod
+    def get_output_dir(task_id: str) -> Path:
+        return FileManager.get_task_dir(task_id) / "output"
+
+    @staticmethod
+    def ensure_output_dir(task_id: str) -> Path:
+        output_dir = FileManager.get_output_dir(task_id)
+        output_dir.mkdir(parents=True, exist_ok=True)
+        return output_dir
+
+    @staticmethod
+    def cleanup_task(task_id: str) -> None:
+        task_dir = FileManager.get_task_dir(task_id)
+        if task_dir.exists():
+            shutil.rmtree(task_dir, ignore_errors=True)
+
+    @staticmethod
+    def get_output_files(task_id: str) -> list:
+        output_dir = FileManager.get_output_dir(task_id)
+        if not output_dir.exists():
+            return []
+        files = []
+        for f in output_dir.iterdir():
+            if f.is_file():
+                files.append({"filename": f.name, "size": f.stat().st_size})
+        return files

web/backend/services/splitter.py

@@ -0,0 +1,86 @@
+"""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
+
+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(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
+
+        args = SimpleNamespace(
+            format=options.get("format", "mp3"),
+            transcode_to=options.get("transcode_to", None),
+            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
+
+        task_manager.update_task(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(
+            task_id,
+            status=TaskStatus.DONE,
+            progress=100,
+            message="Split complete",
+            tracks=output_files
+        )
+
+    except Exception as e:
+        task_manager.update_task(
+            task_id,
+            status=TaskStatus.ERROR,
+            error=str(e),
+            message="Split failed"
+        )

web/backend/services/task_manager.py

@@ -0,0 +1,64 @@
+"""In-memory task state management."""
+
+from typing import Dict, Optional, List
+from datetime import datetime, timedelta
+import threading
+import time
+
+from backend.config import settings
+from backend.models.response import TaskStatus, TrackInfo
+
+
+class TaskManager:
+    def __init__(self):
+        self._tasks: Dict[str, dict] = {}
+        self._lock = threading.Lock()
+
+    def create_task(self, task_id: str, filename: str, file_size: int) -> None:
+        with self._lock:
+            self._tasks[task_id] = {
+                "task_id": task_id,
+                "filename": filename,
+                "file_size": file_size,
+                "status": TaskStatus.PENDING,
+                "progress": 0,
+                "message": "Upload complete",
+                "error": None,
+                "tracks": [],
+                "created_at": datetime.now(),
+                "updated_at": datetime.now()
+            }
+
+    def has_task(self, task_id: str) -> bool:
+        return task_id in self._tasks
+
+    def get_status(self, task_id: str) -> dict:
+        with self._lock:
+            return self._tasks.get(task_id, {}).copy()
+
+    def update_task(self, task_id: str, **kwargs) -> None:
+        with self._lock:
+            if task_id in self._tasks:
+                self._tasks[task_id].update(kwargs)
+                self._tasks[task_id]["updated_at"] = datetime.now()
+
+    def cleanup_old_tasks(self) -> None:
+        now = datetime.now()
+        timeout = timedelta(seconds=settings.cleanup_after_seconds)
+        with self._lock:
+            to_delete = []
+            for task_id, data in self._tasks.items():
+                if now - data["created_at"] > timeout:
+                    to_delete.append(task_id)
+            for task_id in to_delete:
+                self._delete_task_files(task_id)
+                del self._tasks[task_id]
+
+    def _delete_task_files(self, task_id: str) -> None:
+        import shutil
+        task_dir = settings.temp_dir / task_id
+        if task_dir.exists():
+            shutil.rmtree(task_dir, ignore_errors=True)
+
+
+task_manager = TaskManager()

web/backend/utils/__init__.py

@@ -0,0 +1 @@
+"""Utility functions for the web backend."""

web/backend/utils/validators.py

@@ -0,0 +1,23 @@
+"""Input validation utilities."""
+
+import re
+from typing import List, Tuple
+
+from ..models.request import TracklistEntry
+
+
+def validate_tracklist(tracklist: List[TracklistEntry]) -> Tuple[bool, List[str]]:
+    """
+    Validate a tracklist and return any errors.
+
+    Returns:
+        (is_valid, error_messages)
+    """
+    errors = []
+    for idx, entry in enumerate(tracklist, 1):
+        # Each entry must have a timestamp
+        if not entry.ts or not entry.ts.strip():
+            errors.append(f"Line {idx}: Missing timestamp (%ts)")
+        # Additional validation could be added here
+
+    return len(errors) == 0, errors

web/requirements-web.txt

@@ -0,0 +1,7 @@
+fastapi>=0.104.0
+uvicorn[standard]>=0.24.0
+python-multipart>=0.0.6
+aiofiles>=23.2.0
+pydantic>=2.5.0
+pydantic-settings>=2.0.0
+python-dotenv>=1.0.0