Tracs validation added. A warning is thrown if the input contains a video strem, but the requested output format doesn't support it

This commit is contained in:
2026-08-03 11:26:17 +05:00
parent 1d1f8563c0
commit dcdae41706
11 changed files with 330 additions and 52 deletions
+25
View File
@@ -0,0 +1,25 @@
"""Endpoint to expose format information to the frontend."""
from fastapi import APIRouter
from backend.constants import FORMAT_INFO
router = APIRouter(prefix="/api", tags=["formats"])
@router.get("/formats")
async def get_formats():
"""
Return the list of supported container formats with their properties.
"""
return {
"formats": [
{
"name": name,
"ffmpeg": info["ffmpeg"],
"extension": info["ext"],
"audio_only": info["audio_only"],
}
for name, info in FORMAT_INFO.items()
]
}
+35
View File
@@ -0,0 +1,35 @@
"""Endpoint to retrieve stream information for an uploaded file."""
from fastapi import APIRouter, HTTPException
from backend.config import settings
from backend.services.file_manager import FileManager
from backend.services.task_manager import task_manager
from backend.ffmpeg import get_stream_info
router = APIRouter(prefix="/api", tags=["info"])
@router.get("/info/{task_id}")
async def get_task_info(task_id: str):
"""
Return stream information (has_audio, has_video, has_subtitle) for the uploaded file.
"""
if not task_manager.has_task(task_id):
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
input_path = FileManager.get_input_path(task_id)
if not input_path or not input_path.exists():
raise HTTPException(status_code=404, detail="Input file not found")
try:
info = get_stream_info(str(input_path))
return {
"task_id": task_id,
"has_audio": info["has_audio"],
"has_video": info["has_video"],
"has_subtitle": info["has_subtitle"],
"audio_codec": info["audio_codec"],
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to retrieve stream info: {str(e)}")
+12
View File
@@ -0,0 +1,12 @@
FORMAT_INFO = {
'mp3': {'ffmpeg': 'mp3', 'ext': '.mp3', 'audio_only': True},
'm4a': {'ffmpeg': 'mp4', 'ext': '.m4a', 'audio_only': True},
'mp4': {'ffmpeg': 'mp4', 'ext': '.mp4', 'audio_only': False},
'mkv': {'ffmpeg': 'matroska', 'ext': '.mkv', 'audio_only': False},
'matroska': {'ffmpeg': 'matroska', 'ext': '.mkv', 'audio_only': False},
'ogg': {'ffmpeg': 'ogg', 'ext': '.ogg', 'audio_only': True},
'opus': {'ffmpeg': 'ogg', 'ext': '.opus', 'audio_only': True},
'flac': {'ffmpeg': 'flac', 'ext': '.flac', 'audio_only': True},
'wav': {'ffmpeg': 'wav', 'ext': '.wav', 'audio_only': True},
'aac': {'ffmpeg': 'adts', 'ext': '.aac', 'audio_only': True},
}
+79
View File
@@ -0,0 +1,79 @@
"""FFmpeg/FFprobe interaction utilities for the web backend."""
import subprocess
import json
def get_stream_info(input_file: str):
"""
Retrieve stream information (audio, video, subtitle presence) from a media file.
Returns a dict with keys: has_audio, has_video, has_subtitle, audio_codec.
"""
# Get audio codec (if any)
audio_codec = None
try:
cmd = [
'ffprobe', '-v', 'error',
'-select_streams', 'a:0',
'-show_entries', 'stream=codec_name',
'-of', 'default=noprint_wrappers=1:nokey=1',
input_file
]
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
codec = result.stdout.strip().lower()
if codec:
audio_codec = codec
except Exception:
pass
# Check for video stream
has_video = False
try:
cmd = [
'ffprobe', '-v', 'error',
'-select_streams', 'v',
'-show_entries', 'stream=codec_type',
'-of', 'default=noprint_wrappers=1:nokey=1',
input_file
]
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
has_video = bool(result.stdout.strip())
except Exception:
pass
# Check for subtitle stream
has_subtitle = False
try:
cmd = [
'ffprobe', '-v', 'error',
'-select_streams', 's',
'-show_entries', 'stream=codec_type',
'-of', 'default=noprint_wrappers=1:nokey=1',
input_file
]
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
has_subtitle = bool(result.stdout.strip())
except Exception:
pass
return {
'has_audio': audio_codec is not None,
'has_video': has_video,
'has_subtitle': has_subtitle,
'audio_codec': audio_codec,
}
def get_audio_duration(input_file: str) -> float:
"""Get the duration of the audio file in seconds."""
cmd = [
'ffprobe', '-v', 'error',
'-show_entries', 'format=duration',
'-of', 'default=noprint_wrappers=1:nokey=1',
input_file
]
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
try:
return float(result.stdout.strip())
except ValueError:
return 0.0
+4 -3
View File
@@ -5,8 +5,7 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from backend.config import settings
from backend.api import upload, split, status, download
from backend.api.websocket import router as websocket_router
from backend.api import upload, split, status, download, websocket, formats, info
from backend.services import progress_publisher
app = FastAPI(
@@ -36,7 +35,9 @@ app.include_router(upload.router)
app.include_router(split.router)
app.include_router(status.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(info.router) # new
@app.get("/")