injectlib/CapCutAPI
1
0
Fork 0
mirror of https://github.com/sun-guannan/CapCutAPI.git synced 2025-11-25 03:15:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

add get duration local

Browse Source
This commit is contained in:
sun-guannan
2025-07-13 15:09:36 +08:00
parent a842875945
commit 1bb31941df
1 changed files with 38 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
get_duration_impl.py
View File

@@ -4,19 +4,19 @@ import time
def get_video_duration(video_url): def get_video_duration(video_url):
""" """
获取视频时长，支持超时重试。 Get video duration with timeout retry support.
:param video_url: 视频URL :param video_url: Video URL
:return: 视频时长（秒） :return: Video duration (seconds)
""" """
# 定义重试次数和每次重试的等待时间 # Define retry count and wait time for each retry
max_retries = 3 max_retries = 3
retry_delay_seconds = 1 # 每次重试间隔1秒 retry_delay_seconds = 1 # 1 second interval between retries
timeout_seconds = 10 # 设置每次尝试的超时时间 timeout_seconds = 10 # Set timeout for each attempt
for attempt in range(max_retries): for attempt in range(max_retries):
print(f"尝试获取视频时长 (第 {attempt + 1}/{max_retries} 次尝试) ...") print(f"Attempting to get video duration (Attempt {attempt + 1}/{max_retries}) ...")
result = {"success": False, "output": 0, "error": None} # 每次重试前重置结果 result = {"success": False, "output": 0, "error": None} # Reset result before each retry
try: try:
command = [ command = [
@@ -28,73 +28,73 @@ def get_video_duration(video_url):
video_url video_url
] ]
# 使用 subprocess.run 更灵活地处理超时和输出 # Use subprocess.run for more flexible handling of timeout and output
process = subprocess.run(command, process = subprocess.run(command,
capture_output=True, capture_output=True,
text=True, # 自动解码为文本 text=True, # Auto decode to text
timeout=timeout_seconds, # 使用变量设置超时时间 timeout=timeout_seconds, # Use variable to set timeout
check=True) # 如果返回非零退出码则抛出CalledProcessError check=True) # Raise CalledProcessError if non-zero exit code
info = json.loads(process.stdout) info = json.loads(process.stdout)
# 优先从流中获取时长，因为更精确 # Prioritize getting duration from streams because it's more accurate
media_streams = [s for s in info.get('streams', []) if 'duration' in s] media_streams = [s for s in info.get('streams', []) if 'duration' in s]
if media_streams: if media_streams:
duration = float(media_streams[0]['duration']) duration = float(media_streams[0]['duration'])
result["output"] = duration result["output"] = duration
result["success"] = True result["success"] = True
# 否则从格式信息中获取时长 # Otherwise get duration from format information
elif 'format' in info and 'duration' in info['format']: elif 'format' in info and 'duration' in info['format']:
duration = float(info['format']['duration']) duration = float(info['format']['duration'])
result["output"] = duration result["output"] = duration
result["success"] = True result["success"] = True
else: else:
result["error"] = "未找到音视频时长信息。" result["error"] = "Audio/video duration information not found."
# 如果成功获取时长，直接返回结果，不再重试 # If duration is successfully obtained, return result directly without retrying
if result["success"]: if result["success"]:
print(f"成功获取时长: {result['output']:.2f} 秒") print(f"Successfully obtained duration: {result['output']:.2f} seconds")
return result return result
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
result["error"] = f"获取视频时长超时（超过{timeout_seconds}秒）。" result["error"] = f"Getting video duration timed out (exceeded {timeout_seconds} seconds)."
print(f"尝试 {attempt + 1} 超时。") print(f"Attempt {attempt + 1} timed out.")
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
result["error"] = f"执行 ffprobe 命令出错 (退出码 {e.returncode}): {e.stderr.strip()}" result["error"] = f"Error executing ffprobe command (exit code {e.returncode}): {e.stderr.strip()}"
print(f"尝试 {attempt + 1} 失败。错误: {e.stderr.strip()}") print(f"Attempt {attempt + 1} failed. Error: {e.stderr.strip()}")
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
result["error"] = f"解析 JSON 数据出错: {e}" result["error"] = f"Error parsing JSON data: {e}"
print(f"尝试 {attempt + 1} 失败。JSON 解析错误: {e}") print(f"Attempt {attempt + 1} failed. JSON parsing error: {e}")
except FileNotFoundError: except FileNotFoundError:
result["error"] = "ffprobe 命令未找到。请确保 FFmpeg 已安装且在系统 PATH 中。" result["error"] = "ffprobe command not found. Please ensure FFmpeg is installed and in system PATH."
print("错误: ffprobe 命令未找到，请检查安装。") print("Error: ffprobe command not found, please check installation.")
return result # ffprobe本身找不到，无需重试 return result # No need to retry if ffprobe itself is not found
except Exception as e: except Exception as e:
result["error"] = f"发生未知错误: {e}" result["error"] = f"Unknown error occurred: {e}"
print(f"尝试 {attempt + 1} 失败。未知错误: {e}") print(f"Attempt {attempt + 1} failed. Unknown error: {e}")
# 每次失败后都尝试使用远程服务获取时长 # Try using remote service to get duration after each local failure
if not result["success"]: if not result["success"]:
print(f"本地获取失败") print(f"Local retrieval failed")
# try: # try:
# remote_duration = get_duration(video_url) # remote_duration = get_duration(video_url)
# if remote_duration is not None: # if remote_duration is not None:
# result["success"] = True # result["success"] = True
# result["output"] = remote_duration # result["output"] = remote_duration
# result["error"] = None # result["error"] = None
# print(f"远程服务成功获取时长: {remote_duration:.2f} 秒") # print(f"Remote service successfully obtained duration: {remote_duration:.2f} seconds")
# return result # 远程服务成功，直接返回 # return result # Remote service succeeded, return directly
# else: # else:
# print(f"远程服务也无法获取时长 (第 {attempt + 1} 次尝试)") # print(f"Remote service also unable to get duration (Attempt {attempt + 1})")
# except Exception as e: # except Exception as e:
# print(f"远程服务获取时长失败 (第 {attempt + 1} 次尝试): {e}") # print(f"Remote service failed to get duration (Attempt {attempt + 1}): {e}")
# 如果当前尝试失败且未达到最大重试次数，则等待并准备下一次重试 # If current attempt failed and max retries not reached, wait and prepare for next retry
if not result["success"] and attempt < max_retries - 1: if not result["success"] and attempt < max_retries - 1:
print(f"等待 {retry_delay_seconds} 秒后重试...") print(f"Waiting {retry_delay_seconds} seconds before retrying...")
time.sleep(retry_delay_seconds) time.sleep(retry_delay_seconds)
elif not result["success"] and attempt == max_retries - 1: elif not result["success"] and attempt == max_retries - 1:
print(f"已达到最大重试次数 {max_retries}，本地和远程服务都无法获取时长。") print(f"Maximum retry count {max_retries} reached, both local and remote services unable to get duration.")
return result # 所有重试都失败后返回最后一次的失败结果 return result # Return the last failure result after all retries fail