add new video edit scripts for combine video and spliting video

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

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

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

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,

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}")
 

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."""
     
-        # Write all rows at once
-        writer.writerows(data)
+    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.")
+        
+        # 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."""

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"
+]

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)

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()