mirror of
https://github.com/Arrowar/StreamingCommunity.git
synced 2025-06-08 04:25:24 +00:00
Restore github ...
This commit is contained in:
Lovi
parent
9c74f17e31
commit
e05ffbef25
@ -1,25 +1,18 @@
|
||||
# 31.01.24
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import logging
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
from typing import List, Dict
|
||||
|
||||
|
||||
# External libraries
|
||||
try: import ffmpeg # type: ignore
|
||||
except: pass
|
||||
|
||||
|
||||
# Internal utilities
|
||||
from Src.Util._jsonConfig import config_manager
|
||||
from Src.Util.os import check_file_existence, suppress_output
|
||||
from Src.Util.console import console
|
||||
from .util import has_audio_stream, need_to_force_to_ts, check_ffmpeg_input, check_duration_v_a
|
||||
from .util import need_to_force_to_ts, check_ffmpeg_input, check_duration_v_a
|
||||
from .capture import capture_ffmpeg_real_time
|
||||
from ..M3U8 import M3U8_Codec
|
||||
|
||||
@ -40,233 +33,6 @@ CHECK_OUTPUT_CONVERSION = config_manager.get_bool("M3U8_CONVERSION", "check_outp
|
||||
TQDM_USE_LARGE_BAR = config_manager.get_int('M3U8_DOWNLOAD', 'tqdm_use_large_bar')
|
||||
|
||||
|
||||
|
||||
# --> v 1.0 (deprecated)
|
||||
def __concatenate_and_save(file_list_path: str, output_filename: str, v_codec: str = None, a_codec: str = None, bandwidth: int = None, prefix: str = "segments", output_directory: str = None):
|
||||
"""
|
||||
Concatenate input files and save the output with specified decoding parameters.
|
||||
|
||||
Args:
|
||||
- file_list_path (str): Path to the file list containing the segments.
|
||||
- output_filename (str): Output filename for the concatenated video.
|
||||
- v_codec (str): Video decoding parameter (optional).
|
||||
- a_codec (str): Audio decoding parameter (optional).
|
||||
- bandwidth (int): Bitrate for the output video (optional).
|
||||
- prefix (str): Prefix to add at the end of output file name (default is "segments").
|
||||
- output_directory (str): Directory to save the output file. If not provided, defaults to the current directory.
|
||||
- codecs (str): Codecs for video and audio (optional).
|
||||
|
||||
Returns:
|
||||
output_file_path (str): Path to the saved output file.
|
||||
"""
|
||||
|
||||
try:
|
||||
|
||||
# Output arguments
|
||||
output_args = {
|
||||
'c': 'copy',
|
||||
'loglevel': DEBUG_FFMPEG,
|
||||
}
|
||||
|
||||
# Set up the output file name by modifying the video file name
|
||||
output_file_name = output_filename
|
||||
output_file_path = os.path.join(output_directory, output_file_name) if output_directory else output_file_name
|
||||
|
||||
# Concatenate input file list and output
|
||||
output = (
|
||||
ffmpeg.input(file_list_path, safe=0, f='concat')
|
||||
.output(output_file_path, **output_args)
|
||||
)
|
||||
|
||||
# Overwrite output file if exists
|
||||
output = ffmpeg.overwrite_output(output)
|
||||
|
||||
# Execute the process
|
||||
process = output.run()
|
||||
|
||||
except ffmpeg.Error as ffmpeg_error:
|
||||
logging.error(f"Error saving MP4: {ffmpeg_error.stderr.decode('utf-8')}")
|
||||
return ""
|
||||
|
||||
# Remove the temporary file list and folder and completely remove tmp folder
|
||||
os.remove(file_list_path)
|
||||
shutil.rmtree("tmp", ignore_errors=True)
|
||||
return output_file_path
|
||||
|
||||
def __join_audios(video_path: str, audio_tracks: List[Dict[str, str]], prefix: str = "merged") -> str:
|
||||
"""
|
||||
Join video with multiple audio tracks and sync them if there are matching segments.
|
||||
|
||||
Args:
|
||||
- video_path (str): Path to the video file.
|
||||
- audio_tracks (List[Dict[str, str]]): A list of dictionaries, where each dictionary contains 'audio_path'.
|
||||
- prefix (str, optional): Prefix to add at the beginning of the output filename. Defaults to "merged".
|
||||
|
||||
Returns:
|
||||
out_path (str): Path to the saved output video file.
|
||||
"""
|
||||
|
||||
try:
|
||||
|
||||
# Check if video_path exists
|
||||
if not os.path.exists(video_path):
|
||||
raise FileNotFoundError(f"Video file '{video_path}' not found.")
|
||||
|
||||
# Create input streams for video and audio using ffmpeg's.
|
||||
video_stream = ffmpeg.input(video_path)
|
||||
|
||||
# Create a list to store audio streams and map arguments
|
||||
audio_streams = []
|
||||
map_arguments = []
|
||||
|
||||
# Iterate through audio tracks
|
||||
for i, audio_track in enumerate(audio_tracks):
|
||||
audio_path = audio_track.get('path', '')
|
||||
|
||||
# Check if audio_path exists
|
||||
if audio_path:
|
||||
if not os.path.exists(audio_path):
|
||||
logging.warning(f"Audio file '{audio_path}' not found.")
|
||||
continue
|
||||
|
||||
audio_stream = ffmpeg.input(audio_path)
|
||||
audio_streams.append(audio_stream)
|
||||
map_arguments.extend(['-map', f'{i + 1}:a:0'])
|
||||
|
||||
# Set up a process to combine the video and audio streams and create an output file with .mp4 extension.
|
||||
output_file_name = f"{prefix}_{os.path.splitext(os.path.basename(video_path))[0]}.mp4"
|
||||
out_path = os.path.join(os.path.dirname(video_path), output_file_name)
|
||||
|
||||
# Output arguments
|
||||
output_args = {
|
||||
'vcodec': 'copy',
|
||||
'acodec': 'copy',
|
||||
'loglevel': DEBUG_FFMPEG
|
||||
}
|
||||
|
||||
# Combine inputs, map audio streams, and set output
|
||||
output = (
|
||||
ffmpeg.output(
|
||||
video_stream,
|
||||
*audio_streams,
|
||||
out_path,
|
||||
**output_args
|
||||
)
|
||||
.global_args(
|
||||
'-map', '0:v:0',
|
||||
*map_arguments,
|
||||
'-shortest',
|
||||
'-strict', 'experimental',
|
||||
)
|
||||
)
|
||||
|
||||
# Overwrite output file if exists
|
||||
output = ffmpeg.overwrite_output(output)
|
||||
|
||||
# Retrieve the command that will be executed
|
||||
command = output.compile()
|
||||
logging.info(f"Execute command: {command}")
|
||||
|
||||
# Execute the process
|
||||
process = output.run()
|
||||
logging.info("[M3U8_Downloader] Merge completed successfully.")
|
||||
|
||||
# Return
|
||||
return out_path
|
||||
|
||||
except ffmpeg.Error as ffmpeg_error:
|
||||
logging.error("[M3U8_Downloader] Ffmpeg error: %s", ffmpeg_error)
|
||||
return ""
|
||||
|
||||
def __transcode_with_subtitles(video: str, subtitles_list: List[Dict[str, str]], output_file: str, prefix: str = "transcoded") -> str:
|
||||
|
||||
"""
|
||||
Transcode a video with subtitles.
|
||||
|
||||
Args:
|
||||
- video (str): Path to the input video file.
|
||||
- subtitles_list (list[dict[str, str]]): List of dictionaries containing subtitles information.
|
||||
- output_file (str): Path to the output transcoded video file.
|
||||
- prefix (str): Prefix to add to the output file name. Default is "transcoded".
|
||||
|
||||
Returns:
|
||||
str: Path to the transcoded video file.
|
||||
"""
|
||||
|
||||
try:
|
||||
|
||||
# Check if the input video file exists
|
||||
if not os.path.exists(video):
|
||||
raise FileNotFoundError(f"Video file '{video}' not found.")
|
||||
|
||||
# Get input video from video path
|
||||
input_ffmpeg = ffmpeg.input(video)
|
||||
input_video = input_ffmpeg['v']
|
||||
input_audio = input_ffmpeg['a']
|
||||
|
||||
# List with subtitles path and metadata
|
||||
input_subtitles = []
|
||||
metadata = {}
|
||||
|
||||
# Iterate through subtitle tracks
|
||||
for idx, sub_dict in enumerate(subtitles_list):
|
||||
# Get path and name of subtitles
|
||||
sub_file = sub_dict.get('path')
|
||||
title = sub_dict.get('name')
|
||||
|
||||
# Check if the subtitle file exists
|
||||
if not os.path.exists(sub_file):
|
||||
raise FileNotFoundError(f"Subtitle file '{sub_file}' not found.")
|
||||
|
||||
# Append ffmpeg input to list
|
||||
input_ffmpeg_sub = ffmpeg.input(sub_file)
|
||||
input_subtitles.append(input_ffmpeg_sub['s'])
|
||||
|
||||
# Add metadata for title
|
||||
metadata[f'metadata:s:s:{idx}'] = f"title={title}"
|
||||
|
||||
# Check if the input video has an audio stream
|
||||
logging.info(f"There is audio: {has_audio_stream(video)}")
|
||||
|
||||
# Set up the output file name by adding the prefix
|
||||
output_filename = f"{prefix}_{os.path.splitext(os.path.basename(video))[0]}.mkv"
|
||||
output_file = os.path.join(os.path.dirname(output_file), output_filename)
|
||||
|
||||
# Configure ffmpeg output
|
||||
output = ffmpeg.output(
|
||||
input_video,
|
||||
*(input_audio,) if has_audio_stream(video) else (), # If there is no audio stream
|
||||
*input_subtitles,
|
||||
output_file,
|
||||
vcodec='copy',
|
||||
acodec='copy' if has_audio_stream(video) else (), # If there is no audio stream
|
||||
**metadata,
|
||||
loglevel=DEBUG_FFMPEG
|
||||
)
|
||||
|
||||
# Overwrite output file if exists
|
||||
output = ffmpeg.overwrite_output(output)
|
||||
|
||||
# Retrieve the command that will be executed
|
||||
command = output.compile()
|
||||
logging.info(f"Execute command: {command}")
|
||||
|
||||
# Run ffmpeg command
|
||||
ffmpeg.run(output, overwrite_output=True)
|
||||
|
||||
# Rename video from mkv -> mp4
|
||||
output_filename_mp4 = output_file.replace("mkv", "mp4")
|
||||
os.rename(output_file, output_filename_mp4)
|
||||
|
||||
return output_filename_mp4
|
||||
|
||||
except ffmpeg.Error as ffmpeg_error:
|
||||
print(f"Error: {ffmpeg_error}")
|
||||
return ""
|
||||
|
||||
|
||||
|
||||
# --> v 1.1 (new)
|
||||
def join_video(video_path: str, out_path: str, codec: M3U8_Codec = None):
|
||||
|
||||
"""
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user