103 lines
2.8 KiB
Python
103 lines
2.8 KiB
Python
import os
|
|
from typing import Dict
|
|
from typing import Optional
|
|
|
|
from ytdl_sub.entries.entry import Entry
|
|
from ytdl_sub.entries.variables.kwargs import EXT
|
|
from ytdl_sub.plugins.plugin import Plugin
|
|
from ytdl_sub.plugins.plugin import PluginOptions
|
|
from ytdl_sub.utils.exceptions import FileNotDownloadedException
|
|
from ytdl_sub.utils.file_handler import FileMetadata
|
|
from ytdl_sub.validators.audo_codec_validator import VideoCodecTypeValidator
|
|
|
|
|
|
class FileConvertOptions(PluginOptions):
|
|
"""
|
|
Converts media files from one extension to another.
|
|
|
|
Usage:
|
|
|
|
.. code-block:: yaml
|
|
|
|
presets:
|
|
my_example_preset:
|
|
file_converter:
|
|
convert_to: "mp4"
|
|
"""
|
|
|
|
_required_keys = {"convert_to"}
|
|
|
|
def __init__(self, name, value):
|
|
super().__init__(name, value)
|
|
self._convert_to = self._validate_key(
|
|
key="convert_to", validator=VideoCodecTypeValidator
|
|
).value
|
|
|
|
@property
|
|
def convert_to(self) -> str:
|
|
"""
|
|
Convert to a desired extension
|
|
"""
|
|
return self._convert_to
|
|
|
|
|
|
class FileConvertPlugin(Plugin[FileConvertOptions]):
|
|
plugin_options_type = FileConvertOptions
|
|
|
|
def ytdl_options(self) -> Optional[Dict]:
|
|
"""
|
|
Returns
|
|
-------
|
|
ffmpeg video remuxing post processing dict
|
|
"""
|
|
return {
|
|
"postprocessors": [
|
|
{
|
|
"key": "FFmpegVideoRemuxer",
|
|
"when": "post_process",
|
|
"preferedformat": self.plugin_options.convert_to,
|
|
}
|
|
]
|
|
}
|
|
|
|
def modify_entry(self, entry: Entry) -> Optional[Entry]:
|
|
"""
|
|
Parameters
|
|
----------
|
|
entry
|
|
Entry with extracted audio
|
|
|
|
Returns
|
|
-------
|
|
Entry with updated 'ext' source variable
|
|
|
|
Raises
|
|
------
|
|
FileNotDownloadedException
|
|
If the audio file is not found
|
|
"""
|
|
new_ext = self.plugin_options.convert_to
|
|
converted_video_file = entry.get_download_file_path().removesuffix(entry.ext) + new_ext
|
|
if not self.is_dry_run:
|
|
if not os.path.isfile(converted_video_file):
|
|
raise FileNotDownloadedException("Failed to find the converted video file")
|
|
|
|
if entry.ext != new_ext:
|
|
entry.add_kwargs(
|
|
{
|
|
"__converted_from": entry.ext,
|
|
}
|
|
)
|
|
|
|
entry.add_kwargs({EXT: new_ext})
|
|
|
|
return entry
|
|
|
|
def post_process_entry(self, entry: Entry) -> Optional[FileMetadata]:
|
|
"""
|
|
Add metadata about conversion if it happened
|
|
"""
|
|
if converted_from := entry.kwargs_get("__converted_from"):
|
|
return FileMetadata(f"Converted from {converted_from}")
|
|
|
|
return None
|