87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
import json
|
|
import subprocess
|
|
import typer
|
|
from pathlib import Path
|
|
import re
|
|
import click
|
|
|
|
from models.streams import VideoFile, VideoStream, AudioStream, SubtitleStream, OriginalMediaType
|
|
from movie_details import get_movie_details
|
|
from lib.fs import write_shell_script
|
|
|
|
app = typer.Typer()
|
|
|
|
def parse_streams(stream_data: dict, VideoFile: VideoFile) -> VideoFile:
|
|
for stream in stream_data["streams"]:
|
|
if (
|
|
stream["codec_type"] == "video"
|
|
and stream["disposition"]["attached_pic"] == 0
|
|
):
|
|
VideoFile.video_streams.append(VideoStream(**stream))
|
|
elif stream["codec_type"] == "audio":
|
|
VideoFile.audio_streams.append(AudioStream(**stream))
|
|
elif stream["codec_type"] == "subtitle":
|
|
VideoFile.subtitle_streams.append(SubtitleStream(**stream))
|
|
return VideoFile
|
|
|
|
|
|
def validate_imdb(ctx: click.Context, param: click.Parameter, value: str) -> str:
|
|
"""Validate that an IMDb ID matches 'tt' followed by 7-8 digits."""
|
|
if value is None:
|
|
return value
|
|
if not re.match(r"^tt\d{7,8}$", value):
|
|
raise click.BadParameter("IMDb ID must be 'tt' followed by 7-8 digits (e.g. tt1234567)")
|
|
return value
|
|
|
|
|
|
@app.command()
|
|
def main(
|
|
filename: Path = typer.Argument(..., exists=True, help="Input video file"),
|
|
imdb: str = typer.Option(None, "-i", "--imdb", callback=validate_imdb, help="IMDB ID"),
|
|
nosubtitles: bool = typer.Option(False, "-ns", "--nosubtitles", help="Remove subtitles"),
|
|
preset: int = typer.Option(2, "-p", "--preset", help="Encoding preset"),
|
|
crf: int = typer.Option(30, "-c", "--crf", help="CRF value"),
|
|
original_media_type: OriginalMediaType = typer.Option(OriginalMediaType.DVD, "-o", "--original-media-type", case_sensitive=False, help="Source of the original media."),
|
|
film_grain: int = typer.Option(20, "-fg", "--film-grain", help="Film grain amount (default: 20)")
|
|
):
|
|
video_details = VideoFile(
|
|
imdb=imdb, video_path=filename, allow_subtitle=not nosubtitles
|
|
)
|
|
video_details = get_movie_details(video_details)
|
|
|
|
cmd = [
|
|
"ffprobe",
|
|
"-v",
|
|
"quiet",
|
|
"-print_format",
|
|
"json",
|
|
"-show_format",
|
|
"-show_streams",
|
|
filename,
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, check=True)
|
|
data = parse_streams(json.loads(result.stdout), video_details)
|
|
data.original_media_type = original_media_type
|
|
for stream in data.video_streams:
|
|
stream.preset = preset
|
|
stream.crf = crf
|
|
stream.original_media_type = original_media_type
|
|
stream.film_grain = film_grain
|
|
export_data = " ".join(data.cmd())
|
|
export_path = Path(filename).parent / "encode.sh"
|
|
# Confirm overwrite if script exists
|
|
if export_path.exists():
|
|
if not typer.confirm(f"'{export_path}' already exists. Overwrite? "):
|
|
typer.echo("Aborting.")
|
|
raise typer.Exit()
|
|
else:
|
|
export_path.unlink()
|
|
write_shell_script(export_data, export_path)
|
|
print(Path(filename).parent)
|
|
print(" ".join(data.cmd()))
|
|
# json.dump(data, sys.stdout, indent=4)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app()
|