Add Base functionality

src/app.py

@@ -0,0 +1,50 @@
+import json
+import subprocess
+import click
+from pathlib import Path
+# from rich import print
+
+from models.streams import VideoFile, VideoStream, AudioStream, SubtitleStream
+from movie_details import get_movie_details
+
+
+def parse_streams(stream_data: dict, VideoFile: VideoFile) -> VideoFile:
+    for stream in stream_data["streams"]:
+        if stream["codec_type"] == "video":
+            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
+
+
+@click.command()
+@click.argument("filename", type=click.Path(exists=True))
+@click.option("-i", "--imdb", type=str, help="IMDB ID")
+@click.option("-ns", "--nosubtitles", is_flag=True, help="No subtitles")
+def main(filename: Path, imdb: str, nosubtitles: bool):
+    """Run ffprobe on a file and print the output in JSON format."""
+    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)
+    print(" ".join(data.cmd()))
+    # json.dump(data, sys.stdout, indent=4)
+
+
+if __name__ == "__main__":
+    main()

src/lib/fs.py

@@ -0,0 +1,5 @@
+def sanitize_filename(filename: str) -> str:
+    characters_not_allowed = ["<", ">", ":", '"', "/", "\\", "|", "?", "*", " ", "'"]
+    for char in characters_not_allowed:
+        filename = filename.replace(char, "")
+    return filename

src/models/streams.py

@@ -0,0 +1,134 @@
+from pydantic import BaseModel, Field
+from typing import Optional
+from datetime import datetime
+from pathlib import Path
+
+from lib.fs import sanitize_filename
+
+
+class Tags(BaseModel):
+    title: Optional[str] = Field(None)
+    language: Optional[str] = Field(None)
+
+
+class Disposition(BaseModel):
+    default: bool = Field(False)
+    comment: bool = Field(False)
+
+
+class VideoStream(BaseModel):
+    index: int = Field(None)
+    width: int = Field(None)
+    height: int = Field(None)
+    avg_frame_rate: str = Field(None, alias="avg_frame_rate")
+    codec_name: str = Field(None)
+    disposition: Disposition
+    tags: Tags
+
+    @property
+    def fps(self) -> float:
+        frame_numbers = [int(x) for x in self.avg_frame_rate.split("/")]
+        return frame_numbers[0] / frame_numbers[1]
+
+    def cmd_flags(self) -> list[str]:
+        return [
+            "-map",
+            f"0:{self.index}",
+            "-c:v",
+            "libsvtav1",
+            "-crf",
+            "30",
+            "-preset",
+            "6",
+            "-svtav1-params",
+            "tune=0:film-grain=20",
+            "-vf",
+            "'scale=iw*sar:ih,setsar=1,scale=-2:ih:lanczos'",
+            "-g",
+            f"{round(self.fps)*5}",
+        ]
+
+
+class AudioStream(BaseModel):
+    index: int
+    codec_name: str
+    sample_fmt: str
+    channel_layout: str
+    disposition: Disposition
+    tags: Tags
+
+    def cmd_flags(self, index: int) -> list[str]:
+        side_loaded = []
+        default_disposition = (
+            [f"-disposition:a:{index}", "default"] if self.disposition.default else []
+        )
+        if "side" in self.channel_layout:
+            side_loaded = [f"-filter:a:{index}", "'channelmap=channel_layout=5.1'"]
+        return (
+            ["-map", f"0:{self.index}", f"-c:a:{index}", "libopus"]
+            + side_loaded
+            + [f"-metadata:s:a:{index}", f"language={self.tags.language}"]
+            + [
+                f"-metadata:s:a:{index}",
+                f"title='{self.tags.title}'",
+            ]
+            + default_disposition
+        )
+
+
+class SubtitleStream(BaseModel):
+    index: int
+    codec_name: str
+    disposition: Disposition
+    tags: Tags
+
+    def cmd_flags(self, index: int) -> list[str]:
+        return ["-map", f"0:s:{index}", f"-c:s:{index}", "copy"] + [
+            f"-metadata:s:s:{index}",
+            f"language={self.tags.language}",
+        ]
+
+
+class VideoFile(BaseModel):
+    video_path: Path
+    title: str = Field(None)
+    date_released: Optional[datetime] = Field(None)
+    imdb: Optional[str] = Field(None)
+    video_streams: list[VideoStream] = Field([])
+    audio_streams: list[AudioStream] = Field([])
+    subtitle_streams: list[SubtitleStream] = Field([])
+    allow_subtitle: bool = Field(True)
+
+    def cmd(self) -> list[str]:
+        cmd = [
+            "ffmpeg",
+            "-hide_banner",
+            "-loglevel",
+            "error",
+            "-i",
+            str(self.video_path.absolute()),
+            "-map_metadata",
+            "-1",
+            "-metadata",
+            f"title='{self.title}'",
+            "-metadata",
+            f"date_released={self.date_released.strftime('%Y-%m-%d')}",
+            "-metadata",
+            f"imdb={self.imdb}",
+        ]
+        for stream in self.video_streams:
+            cmd.extend(stream.cmd_flags())
+        for i, stream in enumerate(self.audio_streams):
+            cmd.extend(stream.cmd_flags(i))
+        if self.allow_subtitle:
+            for i, stream in enumerate(self.subtitle_streams):
+                cmd.extend(stream.cmd_flags(i))
+
+        output_file = (
+            self.video_path.absolute().parent
+            / f"{sanitize_filename(self.title)}.{self.date_released.year}.mkv"
+        )
+
+        cmd.extend([str(output_file)])
+
+        return cmd

src/movie_details.py

@@ -0,0 +1,23 @@
+from models.streams import VideoFile
+from requests import get
+from dotenv import load_dotenv
+from os import getenv
+from datetime import datetime
+
+load_dotenv()
+
+
+def get_movie_details(VideoFile: VideoFile) -> VideoFile:
+    url = "https://www.omdbapi.com/"
+    params = {
+        "i": VideoFile.imdb,
+        "apikey": getenv("API_KEY"),
+    }
+    response = get(url, params=params)
+
+    if response.status_code == 200:
+        VideoFile.title = response.json()["Title"]
+        VideoFile.date_released = datetime.strptime(
+            response.json()["Released"], "%d %b %Y"
+        )
+        return VideoFile