vedit-mcp

import os
import shutil
import argparse
import subprocess
from pathlib import Path
from mcp.server.fastmcp import FastMCP
from loguru import logger
from logging import INFO
from typing import Optional
# The description of the argument is as follows.
# 1. `kb_dir`：（Required）It is used to set the base path where various video files for operation are located. (For details, please check the code.)
# 2. `using_logger`: (Optional) It is used to set whether to use the logger. If it is set to "True", 
#                   the output will be generated. If it is not set or set to other values, no log will be output.
# 3. `logger_level`:(Optional) The log level. By default, it is set to the DEBUG level. You can set it 
#                   to any one of "DEBUG", "INFO", "ERROR", and "WARNING" by yourself.
# 4. `logger_file_dir`: (Required when `using_logger="True"`), The directory for log output. It is required that this directory must already exist. 

# -----------------------------------------------------------------------------
# Setting Arguments Variable
# Logging Configuration
parser = argparse.ArgumentParser(description='Configure script parameters via command-line arguments')
parser.add_argument('--using_logger', choices=['True', 'False'], default='False',
                    help='Whether to use the logger. Valid values are True or False, default is False')
parser.add_argument('--logger_file_dir', default=None,
                    help='Directory for the log file. If the logger is used, this directory must be specified')
parser.add_argument('--logger_level', default='DEBUG',
                    help='Logging level for the logger, default is DEBUG')
parser.add_argument('--kb_dir', default=None,
                    help='Base path for the video folder')
args = parser.parse_args()

if args.using_logger == "True":
    USING_LOGGER = True
    LOGGER_FILE_DIR = args.logger_file_dir
else:
    USING_LOGGER = False        # default
    LOGGER_FILE_DIR = None
# ---
LOGGER_LEVEL = args.logger_level

# LOGGER_FILE_DIR: this folder must：
# This directory must already exist, and an mcp.log file will be created here to record logs. 
# If you have any other ideas, please modify the code yourself.


# Video Folder Base
KB_DIR = args.kb_dir
KB_CLIP = "clip"
KB_MERGE = "merge"
KB_RESULT = "result"
KB_ADD = "add"      # add bgm, and add ... what i don't know.....

# -----------------------------------------------------------------------------


# Config Logger
def get_logger():
    if not USING_LOGGER:
        logger.remove()
        return logger
    
    if LOGGER_LEVEL not in {"DEBUG", "INFO", "ERROR", "CRITICAL"}:
        raise ValueError(f"`logger_level` Error: the logger level is not exists: {LOGGER_LEVEL}")
    

    if LOGGER_FILE_DIR is None:
        raise ValueError("`logger_file_dir` Error: If you set `using_logger` to `True`, then you must configure " \
        "the argument variable `logger_file_dir` and " \
        "ensure that this directory already exists. ")

    elif not os.path.exists(os.path.abspath(LOGGER_FILE_DIR)):
        raise ValueError(f"`logger_file_dir` Error: The argument `logger_file_dir`={LOGGER_FILE_DIR} must already" \
                         "exist, we can not find it now.")
    

    log_path = os.path.join(LOGGER_FILE_DIR, 'logs' ,'mcp.log')   
    logger.add(log_path, rotation="500 MB", retention="10 days", level=LOGGER_LEVEL,
               format="{time} | {level} | " + "__VEDIO_EDITOR_SERVER__" + ":{function}:{line} - {message}")
    
    return logger

# Check the path
def check_paths():
    if KB_DIR is None:
        raise ValueError("`kb_dir` Error: KB_DIR is None, you must configure the argument variable " \
        "KB_DIR and ensure that this path truly exists. The function of this path is to " \
        "store the original video files, temporary files, and result files.")
    
    elif not os.path.exists(os.path.abspath(KB_DIR)):
        raise ValueError(
            f"`kb_dir` Error: The argument `kb_dir`={KB_DIR} must already exist, we can not find it now.")



logger = get_logger()
check_paths()
# ------------------------------------------------------------------------------
# ------------------------------------------------------------------------------
# start of functions -----------------------------------------------------------
# ------------------------------------------------------------------------------
# Underlying Implementation ----------------------------------------------------
# """
# ## Plan
# Actually, at a cursory glance, video editing can still be divided into the following operations: create, read, update, and delete.
# Create: Add filters, add subtitles, add background music, etc.
# Delete: Delete specified segments, delete subtitles, delete audio tracks, etc.
# Update: Merge multiple videos, cut out a small segment from the original video.
# Read: Query basic information of the original video, such as duration, frame rate, etc. (It seems this isn't very useful).

# ## What has been achieved so far
# 1. clip_video_tool: cut out a small segment from the original video.
# 2. move_videos_tool: merge multiple videos.
# 3. add_bgm_tool: add background music.
# """
# -----------------------------------------------------------------------------


# clip_video:
def clip_video(
        original_video_path: str,
        save_folder: str,
        start_time: int,
        stop_time: int,
        title: str,
) -> tuple[bool, str, str]:
    """
    Cut a video and save it to a specified folder.
    :param original_video_path: The path of the original video file.
    :param save_folder: The path of the folder where the cut video will be saved.
    :param start_time: The start time for cutting (in seconds; this time unit is sufficient for most operations).
    :param stop_time: The end time for cutting (in seconds).
    :return: A tuple where the first element is a boolean indicating whether the operation was successful, 
                the second element is the output location, and the third element is the log information.
    """
    logger.debug("-----------------------------------------------------------------------------")
    logger.debug("Parameter check <clip_video> ------------------------------------------------")
    logger.debug(f"origin_video_path: {original_video_path}")
    logger.debug(f"save_folder: {save_folder}")
    logger.debug(f"start_time: {start_time}")
    logger.debug(f"stop_time: {stop_time}")
    logger.debug("-----------------------------------------------------------------------------")


    # Check if the original video file exists
    if not os.path.isfile(original_video_path):
        error_msg = f"Error: The original video file does not exist."
        logger.error(error_msg)
        return False, "", error_msg
    
    _, file_extension = os.path.splitext(original_video_path)

    # Check if the target folder exists. If it doesn't exist, create it. 
    if not os.path.exists(save_folder):
        try:
            os.makedirs(save_folder)
        except OSError as e:
            error_msg = f"Error: Failed to create the folder. Error message: {e}"
            logger.error(error_msg)
            return False, "", error_msg

    # Generate the output file path
    output_path = os.path.join(save_folder, f"{title}{file_extension}")

    try:
        # Build the FFmpeg command
        command = [
            'ffmpeg',
            '-y',
            '-ss', str(start_time),
            '-to', str(stop_time),
            '-i', original_video_path,
            '-c', 'copy',
            output_path
        ]
        # Execute the FFmpeg command
        subprocess.run(command, check=True)
        success_msg = f"The video has been successfully cut and is being saved. "
        logger.info(success_msg)
        return True, output_path, success_msg
    except subprocess.CalledProcessError as e:
        error_msg = f"Error: An error occurred while executing the FFmpeg command. Error message: {e}"
        logger.error(error_msg)
        return False, "", error_msg
    
# merge_videos
def merge_videos(video_paths: list[str], save_folder: str) -> tuple[bool, str, str]:
    """
    Merge multiple local video files.
    :param video_paths: A list containing the paths of video files.
    :param save_folder: The folder where the merged video will be saved.
    :return: A tuple where the first element is a boolean indicating whether the operation was successful, 
                the second element is the output path, and the third element is the log information.
    """
    logger.debug("-----------------------------------------------------------------------------")
    logger.debug("Parameter check <merge_videos> ----------------------------------------------")
    logger.debug(f"video_path: {str(video_paths)}")
    logger.debug(f"save_folder: {save_folder}")
    logger.debug("-----------------------------------------------------------------------------")
    # Check if all video files exist
    for path in video_paths:
        if not os.path.isfile(path):
            error_msg = f"Error: The video file does not exist."
            logger.error(error_msg)
            return False, "", error_msg

    # Check if the target folder exists. If not, create it.
    if not os.path.exists(save_folder):
        try:
            os.makedirs(save_folder)
        except OSError as e:
            error_msg = f"Error: Failed to create the folder. Error message: {e}"
            logger.error(error_msg)
            return False, "", error_msg

    # Create a temporary file list
    temp_file_list = 'temp_file_list.txt'
    try:
        with open(temp_file_list, 'w', encoding='utf-8') as f:
            for path in video_paths:
                f.write(f"file '{path}'\n")

        # Generate the output file path
        output_path = os.path.join(save_folder, f'result.mp4')

        # Build the FFmpeg command
        command = [
            'ffmpeg',
            '-y',
            '-f', 'concat',
            '-safe', '0',
            '-i', temp_file_list,
            '-c', 'copy',
            output_path
        ]
        # Execute the FFmpeg command
        subprocess.run(command, check=True)
        success_msg = f"The videos have been successfully merged and saved."
        logger.info(success_msg)
        return True, output_path, success_msg
    except subprocess.CalledProcessError as e:
        error_msg = f"Error: An error occurred while executing the FFmpeg command. Error message: {e}"
        logger.error(error_msg)
        return False, "", error_msg
    except Exception as e:
        error_msg = f"An unknown error occurred: {e}"
        logger.error(error_msg)
        return False, "", error_msg
    finally:
        # Delete the temporary file list
        if os.path.exists(temp_file_list):
            os.remove(temp_file_list)


# add_audio_to_video
def add_audio_to_video(video_path: str, audio_path: str, output_path: str, start_time: int = 0, audio_duration: Optional[int] = None) -> tuple[bool, str]:
    logger.debug("-----------------------------------------------------------------------------")
    logger.debug("Parameter check <add_audio_to_video> ----------------------------------------")
    logger.debug(f"video_path: {video_path}")
    logger.debug(f"audio_path: {audio_path}")
    logger.debug(f"output_path: {output_path}")
    logger.debug(f"start_time: {start_time}")
    logger.debug(f"audio_duration: {audio_duration}")
    logger.debug("-----------------------------------------------------------------------------")

    # Get the duration of the video
    ffprobe_cmd = [
        'ffprobe',
        '-v', 'error',
        '-show_entries', 'format=duration',
        '-of', 'default=noprint_wrappers=1:nokey=1',
        video_path
    ]
    try:
        video_duration = float(subprocess.check_output(ffprobe_cmd).decode().strip())
    except subprocess.CalledProcessError as e:
        logger.error(f"Error getting video duration: {e}")
        return False, str(e)
    except ValueError:
        logger.error("Error parsing video duration.")
        return False, str(e)

    # If the audio duration is not specified, use the video duration
    if audio_duration is None:
        audio_duration = video_duration

    # Build the FFmpeg command
    ffmpeg_cmd = [
        'ffmpeg',
        '-i', video_path,
        '-i', audio_path,
        '-ss', str(start_time),
        '-t', str(min(audio_duration, video_duration)),
        '-filter_complex', '[0:a]volume=1[a1];[1:a]volume=1[a2];[a1][a2]amix=inputs=2:duration=first:dropout_transition=0[aout]',
        '-map', '0:v',
        '-map', '[aout]',
        '-c:v', 'copy',
        '-c:a', 'aac',
        output_path
    ]

    try:
        # Execute the FFmpeg command
        subprocess.run(ffmpeg_cmd, check=True)
        logger.info(f"Successfully added audio to video. Output saved to {output_path}")
        return True, "success"
    except subprocess.CalledProcessError as e:
        logger.error(f"Error adding audio to video: {e}")
        return False, str(e)


# copy file and rename
def copy_file(source_file: str, target_folder: str, rename: str) -> tuple[bool, str]:
    if not os.path.isfile(source_file):
        return False, f"it is not a file, path {source_file}"

    if not os.path.exists(target_folder):
        os.makedirs(target_folder)

    # 提取源文件的后缀
    _, file_extension = os.path.splitext(source_file)
    # 构建包含后缀的目标文件名
    target_file = os.path.join(target_folder, rename + file_extension)

    try:
        # 复制文件
        shutil.copy2(source_file, target_file)
        return True, "success"
    except Exception as e:
        return False, f"Error occurred while copying: {str(e)}"

# end of functions -------------------------------------------------------------
# ------------------------------------------------------------------------------
# ------------------------------------------------------------------------------



# MCP Service  --------------------
mcp = FastMCP(
    name="VideoEditorMCP",
    description="A video editing MCP tool service that has implemented " \
    "the basic functions among the fundamental functions.",
    version="0.1.4",

)

# --------------------------------
# MCP Tools ----------------------

# --------------------------------------------------------------------------
# mcp-tools注册
@mcp.tool()
def clip_video_tool(
        original_video_path: str,
        task_id: str,
        start_time: int,
        stop_time: int,
        title: str,
) -> dict:
    """
    Clip a video based on the given start and stop times.
    
    Parameters:
    original_video_path (str): The path of the original video file.
    task_id (str): The unique identifier for the clipping task.
    start_time (int): The start time (in some appropriate unit) for the clipping.
    stop_time (int): The stop time (in some appropriate unit) for the clipping.
    title (str): title: the title of the clipping.(This item does not include suffix names)
    Returns:
    dict: A dictionary containing the result of the clipping operation.
          The dictionary has the following keys:
          - "success": A boolean indicating whether the operation was successful.
          - "message": A string providing additional information about the operation.
          - "output_path": The path of the clipped video file.
    """
    # In fact, both the `original_video_path` and `output_path` are based on the `KB_DIR`, 
    # that is, in the form of `$KB_DIR/xxxxx`. 
    # 
    # It is necessary to ensure that each input is based on the `KB_DIR`, 
    # and each output is also based on the KB. 
    # The reason for this design is mainly due to the concern that errors may occur 
    # when generating paths and other operations. 
    # Therefore, every effort is made to avoid such instability.  
    _original_video_path = os.path.join(KB_DIR, original_video_path)
    _save_folder = os.path.join(KB_DIR, KB_CLIP, task_id)
    success, output_path, message = clip_video(
        _original_video_path, _save_folder, start_time, stop_time, title)
    return {"success": success, "message": message, "output_path": output_path[len(KB_DIR)+1:]}



@mcp.tool()
def merge_videos_tool(video_paths: list[str], task_id: str) -> dict:
    """
    Merge multiple videos into one.

    Parameters:
    video_paths (list[str]): A list of paths of the video files to be merged.
    task_id (str): The unique identifier for the merging task.

    Returns:
    dict: A dictionary containing the result of the merging operation.
          The dictionary has the following keys:
          - "success": A boolean indicating whether the operation was successful.
          - "message": A string providing additional information about the operation.
          - "output_path": The path of the merged video file.
    """
    # the `video_paths` and `output_path` relative to the `KB_DIR``
    _video_paths = []
    for path in video_paths:
        _video_paths.append(os.path.join(KB_DIR, path))
        pass
    _save_folder = os.path.join(KB_DIR, KB_MERGE, task_id)
    success, output_path, message = merge_videos(_video_paths, _save_folder)

    return {"success": success, "message": message, "output_path": output_path[len(KB_DIR)+1:]}


@mcp.tool()
def add_bgm_tool(
    video_path: str, audio_path: str, start_time: int=0, audio_duration: Optional[int]=None
    ) -> dict:
    """
    This function is used to add background music to a video.

    Parameters:
    video_path (str): The path to the video file, which should be a string.
    audio_path (str): The path to the audio file, which should be a string.
    start_time (int, optional): The start time (in seconds) from which the audio will be added to the video. The default value is 0.
    audio_duration (Optional[int], optional): The duration (in seconds) for which the audio will be added to the video. 
                    If it is None, the full duration of the audio will be used.

    Returns:
    str: Returns "success" if successful, or an error message if failed. 
         If an error occurs, notify the user of the reason for the error and apologize sincerely.
    """
    _video_path = os.path.join(KB_DIR, video_path)
    _audio_path = os.path.join(KB_DIR, audio_path)
    _output_path = os.path.join(KB_DIR, KB_ADD)
    
    _, msg = add_audio_to_video(_video_path, _audio_path, _output_path, start_time, audio_duration)
    return msg

@mcp.tool()
def task_endding(task_id: str, source_file: str, title: str = "") -> str:
    """
    This function should be called every time a task ends to push the result document after task processing to the result folder.
    Parameters:
    task_id (str): uniquely identifying the current task.
    source_file(str): Indicates the location of the result file to be pushed, 
    which is the file location provided after the previous process of this task ends.
    title(str): If you need to modify the file name (note: including the file extension), use this parameter.
                Otherwise, keep the file name the same as that of the source_file or simply don't input this parameter as there is a default parameter here.
                (This item does not include suffix names)
    Returns:
    str: Returns "success" if successful, or an error message if failed. 
         If an error occurs, notify the user of the reason for the error and apologize sincerely.
    """
    _source_file = os.path.join(KB_DIR, source_file)
    _target_dir = os.path.join(KB_DIR, KB_RESULT, task_id)
    if not os.path.exists(_target_dir):
        os.makedirs(_target_dir)

    if not os.path.exists(_source_file):
        return "This file does not exist. Please check if the path is correct."

    if len(title) == 0:
        _title, _ = os.path.splitext(os.path.basename(source_file))
    else:
        _title = title

    try:
        _, msg = copy_file(_source_file, _target_dir, _title)
        return msg
    except Exception as e:
        return f"Error: {str(e)}"
    



if __name__ == "__main__":
    logger.info("Video Edit MCP Server Running......")
    mcp.run(transport='stdio')