From 856ab439f702c67c2b929e7bad34c2cda6ae7774 Mon Sep 17 00:00:00 2001 From: SumGuyV5 Date: Tue, 22 Sep 2026 22:12:44 +0000 Subject: [PATCH] add new video edit scripts for combine video and spliting video --- main.py | 21 +++- package/__init__.py | 2 +- package/build_videos.py | 2 +- package/database/database.py | 1 - package/process_uploads.py | 2 +- package/utils.py | 43 ++++++- package/video_editing/__init__.py | 13 ++ package/{ => video_editing}/build_shorts.py | 0 .../video_editing}/combine_videos.py | 35 ++---- .../split_video_by_timestamps.py | 112 ++++++++++++++++++ 10 files changed, 199 insertions(+), 32 deletions(-) create mode 100644 package/video_editing/__init__.py rename package/{ => video_editing}/build_shorts.py (100%) rename {helper_files => package/video_editing}/combine_videos.py (81%) create mode 100644 package/video_editing/split_video_by_timestamps.py diff --git a/main.py b/main.py index c4f451b..64757c9 100644 --- a/main.py +++ b/main.py @@ -1,4 +1,7 @@ #!/usr/bin/env python3 +import os +import sys +import fcntl import asyncio import package.twitch.video_info as video_info @@ -8,8 +11,24 @@ import package.twitch.download_thumbnails as download_thumbnails import package.build_videos as build_videos import package.process_uploads as process_uploads +def lock_file(): + # Open a lock file + lock_file_path = "/tmp/python_script.lock" + fp = open(lock_file_path, 'w') + + try: + # Try to grab an exclusive lock without blocking (LOCK_NB) + fcntl.lockf(fp, fcntl.LOCK_EX | fcntl.LOCK_NB) + except IOError: + print("Another instance is already running. Exiting.") + sys.exit(1) + + # Your script logic goes here + print("Running exclusively on Linux/macOS...") + if __name__ == "__main__": - upload = True + lock_file() + upload = False if upload is False: # Get Twitch Video Info diff --git a/package/__init__.py b/package/__init__.py index eecf374..54b1a3a 100644 --- a/package/__init__.py +++ b/package/__init__.py @@ -14,7 +14,7 @@ os.makedirs(CACHE_DIR, exist_ok=True) # 3. Expose Database Tools (Added) # Replace 'DatabaseClient' and 'connect_to_db' with your actual class/function names -from .build_shorts import fit_to_9_16_letterbox +from .video_editing.build_shorts import fit_to_9_16_letterbox from .build_videos import build_videos from .config import CHANNEL_NAME, TZ, DB from .hash_tags import get_top_nouns diff --git a/package/build_videos.py b/package/build_videos.py index 4b991e2..7fe50ee 100644 --- a/package/build_videos.py +++ b/package/build_videos.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 from datetime import datetime -import package.build_shorts as yt_shorts +import package.video_editing.build_shorts as yt_shorts import package.transcribe_video as transcribe_video import package.twitch.chat_vod as chat_vod from package.config import DB as DB diff --git a/package/database/database.py b/package/database/database.py index def0c49..f45888b 100644 --- a/package/database/database.py +++ b/package/database/database.py @@ -214,7 +214,6 @@ class Repository: def get_youtube_info_all(self) -> list[YoutubeInfo]: return self.session.scalars(select(YoutubeInfo)).all() - def insert(self, data: TwitchVideo | Data | YoutubeInfo | Chapter): def insert(self, data: TwitchVideo | Data | YoutubeInfo | Chapter | list[YoutubeInfo] | list[Chapter]): """ Handles insertion/updating for any of the defined models, diff --git a/package/process_uploads.py b/package/process_uploads.py index 53aeecb..22194d4 100644 --- a/package/process_uploads.py +++ b/package/process_uploads.py @@ -75,7 +75,7 @@ def process_videos(video_type: bool = True, limit: int = 22): if srt_path.exists(): try: - tags.extend(get_top_nouns(str(srt_path))) + tags.extend(get_top_nouns(str(srt_path), (15 - len(tags)))) except Exception as e: print(f"⚠️ Could not get nouns from {srt_path}: {e}") diff --git a/package/utils.py b/package/utils.py index 10a93ee..11e57e2 100755 --- a/package/utils.py +++ b/package/utils.py @@ -45,12 +45,45 @@ def get_video_id_from_chat(json_path: str | Path) -> str | None: return None def write_csv(data: list, file_name: str): - # Open file with newline='' to prevent extra blank rows across platforms - with open(file_name, "w", newline="", encoding="utf-8") as file: - writer = csv.writer(file) + """Writes a list of dictionaries or objects to a CSV file.""" + + if not data: + print(f"⚠️ Warning: No data provided to write to {file_name}.") + return + + # Determine if the data contains objects that need serialization + first_item = data[0] + + if hasattr(first_item, '__dict__') and not isinstance(first_item, (str, int, float, bool)): + print("ℹ️ Detected object types. Converting objects to dictionary rows for CSV writing.") - # Write all rows at once - writer.writerows(data) + # Extract attributes/properties into a list of dictionaries + data_to_write = [] + # We assume all objects in the list have the same structure for simplicity + # A more robust solution would require knowing the schema. + for item in data: + row_dict = {} + # Export all instance attributes to the dictionary + for key, value in item.__dict__.items(): + row_dict[key] = value + data_to_write.append(row_dict) + + # Since csv.writer doesn't handle dicts directly, we list the keys from the first dict + fieldnames = list(data_to_write[0].keys()) + + with open(file_name, "w", newline="", encoding="utf-8") as file: + writer = csv.DictWriter(file, fieldnames=fieldnames) + + writer.writeheader() + writer.writerows(data_to_write) + + else: + # Assume data is already a list of lists/tuples + with open(file_name, "w", newline="", encoding="utf-8") as file: + writer = csv.writer(file) + + # Write all rows at once + writer.writerows(data) def run_command(cmd_str: str, progress_prefix: str = "Progress", look_for: list = ["frame=", "time=", "fps=", "Rendering frame"]) -> bool: """Runs a system command, streams its output live, and reports errors on failure.""" diff --git a/package/video_editing/__init__.py b/package/video_editing/__init__.py new file mode 100644 index 0000000..e83bd91 --- /dev/null +++ b/package/video_editing/__init__.py @@ -0,0 +1,13 @@ +# package/video_editing/__init__.py + +# Expose components from individual modules using relative imports +from .combine_videos import merge_videos +from .split_video_by_timestamps import split_video +from .build_shorts import fit_to_9_16_letterbox + +# Explicitly define what is exposed when importing the subpackage +__all__ = [ + "merge_videos", + "split_video", + "fit_to_9_16_letterbox" +] \ No newline at end of file diff --git a/package/build_shorts.py b/package/video_editing/build_shorts.py similarity index 100% rename from package/build_shorts.py rename to package/video_editing/build_shorts.py diff --git a/helper_files/combine_videos.py b/package/video_editing/combine_videos.py similarity index 81% rename from helper_files/combine_videos.py rename to package/video_editing/combine_videos.py index f2a981b..3c16b62 100644 --- a/helper_files/combine_videos.py +++ b/package/video_editing/combine_videos.py @@ -53,37 +53,28 @@ def concat_videos_reencode(input_files, output_path): for i, video_path in enumerate(input_files): inputs.extend(["-i", str(video_path)]) + # Use semicolons to separate different input streams in the filter graph filter_parts.append(f"[{i}:v]scale=trunc(iw/2)*2:trunc(ih/2)*2,setpts=PTS-STARTPTS[v{i}]") audio_parts.append(f"[{i}:a]aresample=async=1[a{i}]") - concat_filter = ",".join(filter_parts + audio_parts) - map_parts = [] - for i in range(len(input_files)): - map_parts.append(f"[v{i}]") - map_parts.append(f"[a{i}]") - - # Join the video and audio streams in order and concatenate them. - # This works well when the clips are similar enough in format. - filter_complex = f"{','.join(filter_parts)};{';'.join(audio_parts)};" - filter_complex += "".join(f"[{i}:v][{i}:a]" for i in range(len(input_files))) + # Correctly separate individual stream filters with semicolons + filter_complex = f"{';'.join(filter_parts)};{';'.join(audio_parts)};" + filter_complex += "".join(f"[v{i}][a{i}]" for i in range(len(input_files))) filter_complex += f"concat=n={len(input_files)}:v=1:a=1[outv][outa]" cmd = [ "ffmpeg", "-y", *inputs, - "-filter_complex", - filter_complex, - "-map", - "[outv]", - "-map", - "[outa]", - "-c:v", - "libx264", - "-c:a", - "aac", - "-movflags", - "+faststart", + "-filter_complex", filter_complex, + "-map", "[outv]", + "-map", "[outa]", + "-c:v", "libx264", + "-crf", "18", # High quality video + "-preset", "slow", # High efficiency + "-c:a", "aac", + "-b:a", "256k", # High quality audio + "-movflags", "+faststart", str(output_path), ] run_ffmpeg(cmd) diff --git a/package/video_editing/split_video_by_timestamps.py b/package/video_editing/split_video_by_timestamps.py new file mode 100644 index 0000000..49d4ecd --- /dev/null +++ b/package/video_editing/split_video_by_timestamps.py @@ -0,0 +1,112 @@ +import subprocess +import os +from typing import List, Tuple + +def split_video(input_file: str, output_dir: str, timestamps: List[Tuple[float, float]], frame_perfect: bool = False) -> None: + """ + Splits a video file into multiple segments based on specified time ranges. + + Args: + input_file (str): The path to the video file to be split. + output_dir (str): The directory where the split segments will be saved. + timestamps (List[Tuple[float, float]]): A list of tuples representing (start_time, end_time). + frame_perfect (bool): If True, skips fast-copy and forces high-quality re-encoding + to ensure cuts happen exactly on the requested frame. + """ + if not os.path.exists(input_file): + print(f"Error: Input file not found at {input_file}") + return + + os.makedirs(output_dir, exist_ok=True) + print(f"Starting video split process for {input_file}...") + + for i, (start, end) in enumerate(timestamps): + segment_name = f"segment_{i+1:02d}_{start:.2f}s_to_{end:.2f}s.mp4" + output_file = os.path.join(output_dir, segment_name) + + print(f"\n-> Processing segment {i+1}/{len(timestamps)}: {segment_name}") + + # Method 1: Fast Copy (Lossless, no re-encoding) + # Moving -ss BEFORE -i fixes the frozen/corrupted initial frames. + fast_command = [ + "ffmpeg", "-y", + "-ss", str(start), + "-i", input_file, + "-to", str(end), + "-c", "copy", + output_file + ] + + # Method 2: High-Quality Re-encode (Frame-perfect fallback) + # Used if fast copy fails or if frame-accuracy is explicitly requested. + reencode_command = [ + "ffmpeg", "-y", + "-ss", str(start), + "-i", input_file, + "-to", str(end), + "-c:v", "libx264", + "-crf", "18", # High visual quality + "-preset", "slow", # Better compression efficiency + "-c:a", "aac", + "-b:a", "256k", # High-fidelity audio + output_file + ] + + try: + if frame_perfect: + print("Running in frame-perfect mode (re-encoding)...") + subprocess.run(reencode_command, check=True, capture_output=True, text=True) + else: + try: + print("Attempting lightning-fast stream copy...") + subprocess.run(fast_command, check=True, capture_output=True, text=True) + except subprocess.CalledProcessError: + print("⚠️ Fast copy failed. Falling back to high-quality re-encode...") + subprocess.run(reencode_command, check=True, capture_output=True, text=True) + + print(f"Successfully created segment: {output_file}") + + except subprocess.CalledProcessError as e: + print(f"❌ Error splitting video segment {i+1}:") + print(f"Command failed with return code {e.returncode}") + print(f"STDERR: {e.stderr}") + except FileNotFoundError: + print("-" * 50) + print("CRITICAL ERROR: FFmpeg executable not found.") + print("Please ensure 'ffmpeg' is installed and accessible in your system's PATH.") + print("-" * 50) + return + +def main(): + """ + Example usage: + Assumes 'input.mp4' exists and a 'split_videos' directory is ready. + """ + # --- CONFIGURATION --- + INPUT_VIDEO_PATH = "path/to/your/input_video.mp4" # <<< CHANGE THIS + OUTPUT_SEGMENTS_DIR = "output_splits" # <<< CHANGE THIS + + # Define time segments: [(start_time, end_time), ...] + # Times should be floats representing seconds. + # Example 1: 0s to 10s, then 10s to 20s + TIMESTAMPS: List[Tuple[float, float]] = [ + (0.0, 10.0), + (15.0, 25.0), + (30.0, 35.0) + ] + # --------------------- + + if not os.path.exists(INPUT_VIDEO_PATH): + print("!!! IMPORTANT !!!") + print(f"Please update the 'INPUT_VIDEO_PATH' variable in the main() function with the correct path to your video.") + print("!!! Script execution aborted. !!!") + return + + split_video( + input_file=INPUT_VIDEO_PATH, + output_dir=OUTPUT_SEGMENTS_DIR, + timestamps=TIMESTAMPS + ) + +if __name__ == "__main__": + main() \ No newline at end of file