Server-scripts/serverlibrary/core.py

from pathlib import Path
import shutil
import subprocess
import json
import re
from typing import Any


# -----------------------------
# Helpers
# -----------------------------
def safe_delete(path: Path):
    if path.is_dir():
        shutil.rmtree(path)
    else:
        path.unlink()

def safe_rename(src: Path, dst: Path, dry: bool=False, confirm: bool=False, silent=False):
    """
       Safely rename a file or directory without overwriting existing files.
       Automatically appends _1, _2, ... if needed.
       """

    src = Path(src)
    dst = Path(dst)

    # nothing to do
    if src.resolve() == dst.resolve():
        return

    directory = dst.parent
    base = dst.stem
    ext = dst.suffix

    final_dst = dst
    counter = 1

    # find free name
    while final_dst.exists():
        final_dst = directory / f"{base}_{counter}{ext}"
        counter += 1
    ret=f"{src} → {dst}"
    if not silent:
        print(ret)

    if dry:
        return

    if confirm:
        input("Press Enter to confirm...")

    dst.parent.mkdir(parents=True, exist_ok=True)
    src.rename(dst)
    return ret

def get_audio_tracks( source: str, title: int | None = None, handbrake: bool = False) -> list[dict]:
    """
    Return audio track information from a media file or DVD title.

    When ``handbrake`` is True, audio tracks are detected using
    HandBrakeCLI's scan output. This is useful for DVDs and other
    optical media sources where ffprobe may provide incomplete track
    information.

    When ``handbrake`` is False, ffprobe is used to inspect the media
    file and return all audio streams.

    Args:
        source:
            Path to the media file or DVD device/directory.

        title:
            DVD title number to scan when using HandBrakeCLI.
            Ignored when using ffprobe.

        handbrake:
            If True, use HandBrakeCLI for track detection.
            If False, use ffprobe.

    Returns:
        List of dictionaries containing:

        - number: audio track number/index
        - description: language, title, or track description

    Example:
        get_audio_tracks("movie.mkv")
        [{'number': 1, 'description': 'eng'}]

        get_audio_tracks("/dev/dvd", title=1, handbrake=True)
        [{'number': 1, 'description': 'Deutsch'}]
    """
    tracks: list[dict] = []

    if handbrake:
        result = subprocess.run(
            [
                "HandBrakeCLI",
                "-i", source,
                "-t", str(title),
                "--scan",
            ],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            check=True,
        )

        pattern = re.compile(
            r"\+\s+(\d+),\s+(.+?)\s+\((?:AC3|AAC|MP3|E-AC-3|DTS)"
        )

        for match in pattern.finditer(result.stdout):
            tracks.append({
                "number": int(match.group(1)),
                "description": match.group(2).strip(),
            })

    else:
        result = subprocess.run(
            [
                "ffprobe",
                "-v", "quiet",
                "-print_format", "json",
                "-show_streams",
                source,
            ],
            capture_output=True,
            text=True,
            check=True,
        )

        info = json.loads(result.stdout)

        for stream in info["streams"]:
            if stream.get("codec_type") != "audio":
                continue

            tracks.append({
                "number": stream["index"],
                "description": (
                    stream.get("tags", {}).get("title")
                    or stream.get("tags", {}).get("language")
                    or f"Track {stream['index']}"
                ),
                "primary": stream.get("codec_type") == "audio" and stream.get("disposition", {}).get("default") == 1
            })

    if not tracks:
        tracks.append({
            "number": 0,
            "description": "",
            "primary": False
        })

    return tracks

def get_subtitle_tracks(source):
    result = subprocess.run(
        [
            "ffprobe",
            "-v", "quiet",
            "-print_format", "json",
            "-show_streams",
            source,
        ],
        capture_output=True,
        text=True,
        check=True,
    )

    info = json.loads(result.stdout)

    tracks = []

    for stream in info["streams"]:
        if stream.get("codec_type") != "subtitle":
            continue

        tracks.append({
            "number": stream["index"],
            "description": (
                stream.get("tags", {}).get("title")
                or stream.get("tags", {}).get("language")
                or f"Subtitle {stream['index']}"
            ),
            "primary": (
                stream.get("disposition", {}).get("default") == 1
            ),
        })

    if not tracks:
        tracks.append({
            "number": 0,
            "description": "",
            "primary": False,
        })

    return tracks