Update app.py

app.py

@@ -1,226 +1,91 @@
 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  # 100MB
-MAX_AUDIO_SIZE = 50 * 1024 * 1024   # 50MB
-MAX_TOTAL_SIZE = 500 * 1024 * 1024  # 500MB
-
-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 文件"
+import pickle
+from googleapiclient.discovery import build
+from googleapiclient.http import MediaFileUpload
+from google_auth_oauthlib.flow import InstalledAppFlow
+from google.auth.transport.requests import Request
+
+CLIENT_SECRETS_FILE = 'client_secrets.json'
+CREDENTIALS_FILE = 'youtube_token.pickle'
+SCOPES = ['https://www.googleapis.com/auth/youtube.upload']
+
+def authenticate_youtube():
+    creds = None
+    if os.path.exists(CREDENTIALS_FILE):
+        with open(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:
-            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]
+            flow = InstalledAppFlow.from_client_secrets_file(
+                CLIENT_SECRETS_FILE, SCOPES)
+            # 手动认证：命令行打印链接，复制粘贴访问授权
+            creds = flow.run_console()
+        with open(CREDENTIALS_FILE, 'wb') as token:
+            pickle.dump(creds, token)
+    youtube = build('youtube', 'v3', credentials=creds)
+    return youtube
+
+def upload_video(youtube, video_path, title, description, tags=None, privacy_status="private"):
+    if not os.path.exists(video_path):
+        print(f"错误：找不到视频文件 {video_path}")
+        return None
+    
+    body = {
+        'snippet': {
+            'title': title,
+            'description': description,
+            'tags': tags or [],
+            'categoryId': '22'  # People & Blogs
+        },
+        'status': {
+            'privacyStatus': privacy_status,
+            'selfDeclaredMadeForKids': False
+        }
+    }
+    
+    media = MediaFileUpload(video_path, chunksize=-1, resumable=True, mimetype='video/*')
+    request = youtube.videos().insert(
+        part=','.join(body.keys()),
+        body=body,
+        media_body=media
     )
+    response = request.execute()
+    
+    print(f"上传成功！视频链接：https://www.youtube.com/watch?v={response['id']}")
+    return response['id']
+
+def main():
+    youtube = authenticate_youtube()
+    
+    # 在这里替换为你想上传的视频文件及信息
+    videos_to_upload = [
+        {
+            'path': 'your_video_1.mp4',
+            'title': '视频标题1',
+            'description': '这是视频描述1',
+            'tags': ['示例', '测试'],
+            'privacy_status': 'private'
+        },
+        {
+            'path': 'your_video_2.mp4',
+            'title': '视频标题2',
+            'description': '这是视频描述2',
+            'tags': ['样例', '上传'],
+            'privacy_status': 'unlisted'
+        }
+    ]
+    
+    for video_info in videos_to_upload:
+        upload_video(
+            youtube,
+            video_info['path'],
+            video_info['title'],
+            video_info['description'],
+            video_info.get('tags'),
+            video_info.get('privacy_status', 'private')
+        )
 
-demo.launch()
+if __name__ == '__main__':
+    main()