|
|
import os |
|
|
import gradio as gr |
|
|
import tempfile |
|
|
import random |
|
|
import shutil |
|
|
from datetime import datetime |
|
|
|
|
|
try: |
|
|
from openai import OpenAI |
|
|
client = OpenAI(api_key=os.getenv("sk-ZYyxdh3NexC863OUdicSlKrunXEPlRlsLBMwc1vXMGEVwZdb")) if os.getenv("sk-ZYyxdh3NexC863OUdicSlKrunXEPlRlsLBMwc1vXMGEVwZdb") else None |
|
|
OPENAI_AVAILABLE = bool(client) |
|
|
except ImportError: |
|
|
OPENAI_AVAILABLE = False |
|
|
client = None |
|
|
|
|
|
try: |
|
|
from moviepy.editor import VideoFileClip, concatenate_videoclips, AudioFileClip, vfx |
|
|
MOVIEPY_AVAILABLE = True |
|
|
except ImportError: |
|
|
MOVIEPY_AVAILABLE = False |
|
|
|
|
|
try: |
|
|
from googleapiclient.discovery import build |
|
|
from googleapiclient.http import MediaFileUpload |
|
|
from google_auth_oauthlib.flow import InstalledAppFlow |
|
|
from google.auth.transport.requests import Request |
|
|
import pickle |
|
|
YOUTUBE_AVAILABLE = True |
|
|
except ImportError: |
|
|
YOUTUBE_AVAILABLE = False |
|
|
|
|
|
MAX_VIDEO_SIZE = 100 * 1024 * 1024 |
|
|
MAX_AUDIO_SIZE = 50 * 1024 * 1024 |
|
|
MAX_TOTAL_SIZE = 500 * 1024 * 1024 |
|
|
|
|
|
class YouTubeUploader: |
|
|
def __init__(self): |
|
|
self.SCOPES = ['https://www.googleapis.com/auth/youtube.upload'] |
|
|
self.client_secrets_file = 'client_secrets.json' |
|
|
self.credentials_file = 'youtube_token.pickle' |
|
|
self.youtube = None |
|
|
|
|
|
def authenticate(self): |
|
|
creds = None |
|
|
if os.path.exists(self.credentials_file): |
|
|
with open(self.credentials_file, 'rb') as token: |
|
|
creds = pickle.load(token) |
|
|
if not creds or not creds.valid: |
|
|
if creds and creds.expired and creds.refresh_token: |
|
|
creds.refresh(Request()) |
|
|
else: |
|
|
flow = InstalledAppFlow.from_client_secrets_file( |
|
|
self.client_secrets_file, self.SCOPES |
|
|
) |
|
|
creds = flow.run_local_server(port=0, open_browser=False) |
|
|
print("请复制下面的链接到浏览器打开并完成授权") |
|
|
with open(self.credentials_file, 'wb') as token: |
|
|
pickle.dump(creds, token) |
|
|
self.youtube = build('youtube', 'v3', credentials=creds) |
|
|
return True |
|
|
|
|
|
def upload_video(self, video_path, title, description="", tags="", privacy="private"): |
|
|
if not os.path.exists(video_path): |
|
|
return {"success": False, "error": f"视频文件 {video_path} 不存在"} |
|
|
|
|
|
body = { |
|
|
"snippet": { |
|
|
"title": title, |
|
|
"description": description, |
|
|
"tags": tags.split(",") if tags else [], |
|
|
"categoryId": "22" |
|
|
}, |
|
|
"status": { |
|
|
"privacyStatus": privacy, |
|
|
"selfDeclaredMadeForKids": False |
|
|
} |
|
|
} |
|
|
|
|
|
media = MediaFileUpload(video_path, chunksize=-1, resumable=True, mimetype="video/*") |
|
|
request = self.youtube.videos().insert(part=",".join(body.keys()), body=body, media_body=media) |
|
|
|
|
|
response = request.execute() |
|
|
video_id = response.get("id") |
|
|
url = f"https://www.youtube.com/watch?v={video_id}" |
|
|
return {"success": True, "video_id": video_id, "url": url, "title": title} |
|
|
|
|
|
def check_filesize(video_files, audio_file): |
|
|
total_size = 0 |
|
|
for v in video_files: |
|
|
size = os.path.getsize(v.name) |
|
|
if size > MAX_VIDEO_SIZE: |
|
|
return False, f"单个视频文件 {os.path.basename(v.name)} 过大，超过100MB限制" |
|
|
total_size += size |
|
|
if audio_file: |
|
|
audio_size = os.path.getsize(audio_file.name) |
|
|
if audio_size > MAX_AUDIO_SIZE: |
|
|
return False, f"音频文件 {os.path.basename(audio_file.name)} 过大，超过50MB限制" |
|
|
total_size += audio_size |
|
|
if total_size > MAX_TOTAL_SIZE: |
|
|
return False, f"上传文件总大小过大 {total_size/(1024*1024):.1f}MB，超过500MB限制" |
|
|
return True, "" |
|
|
|
|
|
def process_videos(video_files, audio_file, clip_duration, num_output): |
|
|
ok, msg = check_filesize(video_files, audio_file) |
|
|
if not ok: |
|
|
return msg, [], "" |
|
|
|
|
|
out_files = [] |
|
|
try: |
|
|
clips = [] |
|
|
for video in video_files: |
|
|
clip = VideoFileClip(video.name) |
|
|
duration = clip.duration |
|
|
|
|
|
for start in range(0, int(duration), int(clip_duration)): |
|
|
end = min(start + clip_duration, duration) |
|
|
subclip = clip.subclip(start, end).without_audio() |
|
|
clips.append(subclip) |
|
|
|
|
|
random.shuffle(clips) |
|
|
|
|
|
clips_per_video = max(1, len(clips) // num_output) |
|
|
output_dir = tempfile.mkdtemp() |
|
|
|
|
|
for i in range(num_output): |
|
|
start_idx = i * clips_per_video |
|
|
end_idx = len(clips) if i == num_output - 1 else (start_idx + clips_per_video) |
|
|
selected_clips = clips[start_idx:end_idx] |
|
|
if not selected_clips: |
|
|
continue |
|
|
|
|
|
final_clip = concatenate_videoclips(selected_clips) |
|
|
|
|
|
if audio_file: |
|
|
audio_clip = AudioFileClip(audio_file.name) |
|
|
min_duration = min(final_clip.duration, audio_clip.duration) |
|
|
final_clip = final_clip.subclip(0, min_duration) |
|
|
audio_clip = audio_clip.subclip(0, min_duration) |
|
|
final_clip = final_clip.set_audio(audio_clip) |
|
|
|
|
|
output_path = os.path.join(output_dir, f"mixcut_{i+1}.mp4") |
|
|
final_clip.write_videofile(output_path, codec="libx264", audio_codec="aac", verbose=False, logger=None) |
|
|
out_files.append(output_path) |
|
|
final_clip.close() |
|
|
|
|
|
shutil.rmtree(output_dir, ignore_errors=True) |
|
|
|
|
|
return f"成功生成 {len(out_files)} 个混剪视频", out_files, output_dir |
|
|
|
|
|
except Exception as e: |
|
|
return f"❌ 视频处理出错: {str(e)}", [], "" |
|
|
|
|
|
def generate_ai_analysis(video_count, file_names): |
|
|
if not OPENAI_AVAILABLE or video_count == 0: |
|
|
return "⚠️ 无视频文件或 AI 配置不可用。" |
|
|
try: |
|
|
file_info = ", ".join([f"'{name}'" for name in file_names[:3]]) |
|
|
prompt = f"""作为专业视频分析师，基于用户上传的 {video_count} 个视频文件（包括：{file_info} 等），请提供分析： |
|
|
1. 内容类型推测 |
|
|
2. 背景音乐建议 |
|
|
3. 剪辑策略 |
|
|
4. 目标用户 |
|
|
|
|
|
用中文回答，格式清晰。""" |
|
|
response = client.chat.completions.create( |
|
|
model="gpt-4o-mini", |
|
|
messages=[ |
|
|
{"role": "system", "content": "你是视频内容专家。"}, |
|
|
{"role": "user", "content": prompt} |
|
|
], |
|
|
max_tokens=600, |
|
|
temperature=0.7 |
|
|
) |
|
|
return response.choices[0].message.content |
|
|
except Exception as e: |
|
|
return f"❌ AI 分析失败: {str(e)}" |
|
|
|
|
|
def main_app(video_files, audio_file, clip_duration, num_output, upload_yt, yt_privacy): |
|
|
status, outfiles, _ = process_videos(video_files, audio_file, clip_duration, num_output) |
|
|
|
|
|
file_names = [os.path.basename(f.name) for f in (video_files or [])] |
|
|
ai_analysis = generate_ai_analysis(len(video_files or []), file_names) |
|
|
|
|
|
yt_result = "" |
|
|
if upload_yt: |
|
|
if not YOUTUBE_AVAILABLE: |
|
|
yt_result = "⚠️ 未安装 YouTube API 库" |
|
|
elif not os.path.exists("client_secrets.json"): |
|
|
yt_result = "⚠️ 缺少 client_secrets.json 文件" |
|
|
else: |
|
|
uploader = YouTubeUploader() |
|
|
if uploader.authenticate(): |
|
|
yt_result = "认证成功，上传功能请手动调用上传接口。" |
|
|
else: |
|
|
yt_result = "❌ 认证失败" |
|
|
|
|
|
return status, ai_analysis, yt_result |
|
|
|
|
|
try: |
|
|
import googleapiclient |
|
|
YOUTUBE_AVAILABLE = True |
|
|
except ImportError: |
|
|
YOUTUBE_AVAILABLE = False |
|
|
|
|
|
with gr.Blocks() as demo: |
|
|
gr.Markdown("## pyTovideo2 完整视频切割混剪 + YouTube上传（需预先配置）") |
|
|
|
|
|
video_files = gr.File(file_types=[".mp4", ".mov", ".avi"], label="上传视频文件", file_count="multiple") |
|
|
audio_file = gr.File(file_types=[".mp3", ".wav", ".m4a"], label="上传音频文件") |
|
|
clip_duration = gr.Number(value=2, label="切片时长 (秒)", minimum=1, maximum=10) |
|
|
num_output = gr.Number(value=3, label="输出视频数量", minimum=1, maximum=10) |
|
|
upload_yt = gr.Checkbox(label="处理后上传到 YouTube") |
|
|
yt_privacy = gr.Dropdown(choices=["public", "private", "unlisted"], value="private", label="选择 YouTube 视频隐私") |
|
|
btn = gr.Button("开始处理") |
|
|
|
|
|
status_out = gr.Textbox(label="处理状态", lines=4, interactive=False) |
|
|
ai_out = gr.Textbox(label="AI 内容分析", lines=12) |
|
|
yt_out = gr.Textbox(label="YouTube 上传状态", lines=6) |
|
|
|
|
|
btn.click( |
|
|
main_app, |
|
|
inputs=[video_files, audio_file, clip_duration, num_output, upload_yt, yt_privacy], |
|
|
outputs=[status_out, ai_out, yt_out] |
|
|
) |
|
|
|
|
|
demo.launch() |
|
|
|
