Create app.py

app.py

@@ -0,0 +1,563 @@
+import gradio as gr
+import yt_dlp
+import os
+import re
+import json
+from pathlib import Path
+import tempfile
+import shutil
+from urllib.parse import urlparse, parse_qs
+import threading
+from concurrent.futures import ThreadPoolExecutor
+import time
+
+SUPPORTED_PLATFORMS = {
+    "抖音": r'(https?://)?(v\.douyin\.com|www\.douyin\.com)',
+    "快手": r'(https?://)?(v\.kuaishou\.com|www\.kuaishou\.com)',
+    "哔哩哔哩": r'(https?://)?(www\.bilibili\.com|b23\.tv)',
+    "YouTube": r'(https?://)?(www\.youtube\.com|youtu\.be)',
+    "小红书": r'(https?://)?(www\.xiaohongshu\.com|xhslink\.com)',
+    "微博": r'(https?://)?(weibo\.com|t\.cn)',
+    "西瓜视频": r'(https?://)?(www\.ixigua\.com)',
+    "腾讯视频": r'(https?://)?(v\.qq\.com)'
+}
+
+def get_platform_from_url(url):
+    """
+    自动识别URL所属平台
+    """
+    if not url:
+        return None
+    
+    for platform, pattern in SUPPORTED_PLATFORMS.items():
+        if re.search(pattern, url):
+            return platform
+    return None
+
+def get_platform_config(url, format_id=None):
+    """
+    根据URL返回对的配置
+    """
+    platform = get_platform_from_url(url)
+    if not platform:
+        return None
+        
+    # 基础配置
+    base_config = {
+        'format': format_id if format_id else 'best',
+        'merge_output_format': 'mp4',
+        # 网络相关设置
+        'socket_timeout': 10,  # 减少超时时间
+        'retries': 2,  # 减少重试次数
+        'fragment_retries': 2,
+        'retry_sleep': 2,  # 减少重试等待时间
+        'concurrent_fragment_downloads': 8,
+    }
+    
+    configs = {
+        "抖音": {
+            **base_config,
+            'format': format_id if format_id else 'best',
+        },
+        "快手": {
+            **base_config,
+            'format': format_id if format_id else 'best',
+        },
+        "哔哩哔哩": {
+            **base_config,
+            'format': format_id if format_id else 'bestvideo+bestaudio/best',
+            # B站特定设置
+            'concurrent_fragment_downloads': 16,
+            'file_access_retries': 2,
+            'extractor_retries': 2,
+            'fragment_retries': 2,
+            'retry_sleep': 2,
+        },
+        "YouTube": {
+            **base_config,
+            'format': format_id if format_id else 'bestvideo+bestaudio/best',
+        },
+        "小红书": {
+            **base_config,
+            'format': format_id if format_id else 'best',
+        },
+        "微博": {
+            **base_config,
+            'format': format_id if format_id else 'best',
+        },
+        "西瓜视频": {
+            **base_config,
+            'format': format_id if format_id else 'best',
+        },
+        "腾讯视频": {
+            **base_config,
+            'format': format_id if format_id else 'best',
+        }
+    }
+    
+    return configs.get(platform)
+
+def validate_url(url):
+    """
+    验证URL是否符合支持的平台格式
+    """
+    if not url:
+        return False, "请输入视频链接"
+    
+    platform = get_platform_from_url(url)
+    if not platform:
+        return False, "不支持的平台或链接格式不正确"
+    
+    return True, f"识别为{platform}平台"
+
+def format_filesize(bytes):
+    """
+    格式化文件大小显示
+    """
+    if not bytes:
+        return "未知大小"
+    
+    for unit in ['B', 'KB', 'MB', 'GB']:
+        if bytes < 1024:
+            return f"{bytes:.1f} {unit}"
+        bytes /= 1024
+    return f"{bytes:.1f} TB"
+
+def parse_video_info(url):
+    """
+    解析视频信息
+    """
+    try:
+        # 验证URL
+        is_valid, message = validate_url(url)
+        if not is_valid:
+            return {"status": "error", "message": message}
+        
+        # 获取平台特定配置
+        ydl_opts = get_platform_config(url)
+        if not ydl_opts:
+            return {"status": "error", "message": "不支持的平台"}
+            
+        ydl_opts.update({
+            'quiet': True,
+            'no_warnings': True,
+        })
+        
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(url, download=False)
+            if not info:
+                return {"status": "error", "message": "无法获取视频信息"}
+            
+            # 获取可用的格式
+            formats = []
+            seen_resolutions = set()  # 用于去重
+            if 'formats' in info:
+                # 过滤和排序格式
+                video_formats = []
+                for f in info['formats']:
+                    # 过滤音频格式和没有视频编码的格式
+                    if f.get('vcodec') == 'none' or not f.get('vcodec'):
+                        continue
+                    
+                    # 获取分辨率
+                    width = f.get('width', 0)
+                    height = f.get('height', 0)
+                    resolution = f.get('resolution', 'unknown')
+                    if width and height:
+                        resolution = f"{width}x{height}"
+                    
+                    # 获取格式说明
+                    format_note = f.get('format_note', '')
+                    if not format_note and resolution != 'unknown':
+                        if height:
+                            format_note = f"{height}p"
+                    
+                    # 创建唯一标识用于去重
+                    resolution_key = f"{height}_{width}" if height and width else resolution
+                    
+                    # 如果这个分辨率已经存在，跳过
+                    if resolution_key in seen_resolutions:
+                        continue
+                    seen_resolutions.add(resolution_key)
+                    
+                    # 创建格式信息
+                    format_info = {
+                        'format_id': f.get('format_id', ''),
+                        'ext': f.get('ext', ''),
+                        'resolution': resolution,
+                        'format_note': format_note,
+                        'quality': height or 0  # 用于排序
+                    }
+                    video_formats.append(format_info)
+                
+                # 按质量排序
+                video_formats.sort(key=lambda x: x['quality'], reverse=True)
+                formats = video_formats
+            
+            # 获取��览图
+            thumbnail = info.get('thumbnail', '')
+            if not thumbnail and 'thumbnails' in info:
+                thumbnails = info['thumbnails']
+                if thumbnails:
+                    thumbnail = thumbnails[-1]['url']
+            
+            platform = get_platform_from_url(url)
+            return {
+                "status": "success",
+                "message": "解析成功",
+                "platform": platform,
+                "title": info.get('title', '未知标题'),
+                "duration": info.get('duration', 0),
+                "formats": formats,
+                "thumbnail": thumbnail,
+                "description": info.get('description', ''),
+                "webpage_url": info.get('webpage_url', url),
+            }
+            
+    except Exception as e:
+        return {"status": "error", "message": f"解析失败: {str(e)}"}
+
+class DownloadProgress:
+    def __init__(self):
+        self.progress = 0
+        self.status = "准备下载"
+        self.lock = threading.Lock()
+    
+    def update(self, d):
+        with self.lock:
+            if d.get('status') == 'downloading':
+                total = d.get('total_bytes')
+                downloaded = d.get('downloaded_bytes')
+                if total and downloaded:
+                    self.progress = (downloaded / total) * 100
+                self.status = f"下载中: {d.get('_percent_str', '0%')} of {d.get('_total_bytes_str', 'unknown')}"
+            elif d.get('status') == 'finished':
+                self.progress = 100
+                self.status = "下载完成，正在处理..."
+
+def get_downloads_dir():
+    """
+    获取用户的下载目录
+    """
+    # 获取用户主目录
+    home = str(Path.home())
+    # 获取下载目录
+    downloads_dir = os.path.join(home, "Downloads")
+    # 如果下载目录不存在，则创建
+    if not os.path.exists(downloads_dir):
+        downloads_dir = home
+    return downloads_dir
+
+def clean_filename(title, platform):
+    """
+    清理并格式化文件名
+    """
+    # 移除非法字符
+    illegal_chars = r'[<>:"/\\|?*\n\r\t]'
+    clean_title = re.sub(illegal_chars, '', title)
+    
+    # 移除多余的空格和特殊符号
+    clean_title = re.sub(r'\s+', ' ', clean_title).strip()
+    clean_title = re.sub(r'[,.，。！!@#$%^&*()（）+=\[\]{};:]+', '', clean_title)
+    
+    # 移除表情符号
+    clean_title = re.sub(r'[\U0001F300-\U0001F9FF]', '', clean_title)
+    
+    # 添加平台标识
+    platform_suffix = {
+        "抖音": "抖音",
+        "快手": "快手",
+        "哔哩哔哩": "B站",
+        "YouTube": "YT",
+        "小红书": "XHS",
+        "微博": "微博",
+        "西瓜视频": "西瓜",
+        "腾讯视频": "腾讯"
+    }
+    
+    # 限制标题长度（考虑到平台标识的长度）
+    max_length = 50
+    if len(clean_title) > max_length:
+        clean_title = clean_title[:max_length-3] + '...'
+    
+    # 添加时间戳和平台标识
+    timestamp = time.strftime("%Y%m%d", time.localtime())
+    suffix = platform_suffix.get(platform, "视频")
+    
+    # 最终文件名格式：标题_时间_平台.mp4
+    final_name = f"{clean_title}_{timestamp}_{suffix}"
+    
+    return final_name
+
+def download_single_video(url, format_id, progress_tracker):
+    """
+    下载单个视频
+    """
+    try:
+        # 创建临时目录
+        temp_dir = tempfile.mkdtemp()
+        
+        # 获取平台信息
+        platform = get_platform_from_url(url)
+        if not platform:
+            shutil.rmtree(temp_dir, ignore_errors=True)
+            return {"status": "error", "message": "不支持的平台"}
+        
+        # 获取视频信息
+        with yt_dlp.YoutubeDL({'quiet': True}) as ydl:
+            info = ydl.extract_info(url, download=False)
+            # 清理并格式化文件名
+            clean_title = clean_filename(info.get('title', 'video'), platform)
+        
+        ydl_opts = get_platform_config(url, format_id)
+        if not ydl_opts:
+            shutil.rmtree(temp_dir, ignore_errors=True)
+            return {"status": "error", "message": "不支持的平台"}
+        
+        # 更新下载配置
+        ydl_opts.update({
+            'quiet': False,
+            'no_warnings': False,
+            'extract_flat': False,
+            'paths': {'home': temp_dir},
+            'progress_hooks': [progress_tracker.update],
+            'outtmpl': clean_title + '.%(ext)s',  # 不使用绝对路径
+            'ignoreerrors': True,  # 忽略部分错误继续下载
+            'noprogress': False,  # 显示进度
+            'continuedl': True,  # 支持断点续传
+            'retries': float('inf'),  # 无限重试
+            'fragment_retries': float('inf'),  # 片段无限重试
+            'skip_unavailable_fragments': True,  # 跳过不可用片段
+            'no_abort_on_error': True,  # 发生错误时不中止
+        })
+        
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            try:
+                info = ydl.extract_info(url, download=True)
+                if 'requested_downloads' in info:
+                    file_path = info['requested_downloads'][0]['filepath']
+                else:
+                    file_path = os.path.join(temp_dir, f"{clean_title}.mp4")
+                
+                if os.path.exists(file_path):
+                    # 检查文件大小
+                    file_size = os.path.getsize(file_path)
+                    if file_size == 0:
+                        shutil.rmtree(temp_dir, ignore_errors=True)
+                        return {"status": "error", "message": "下载的文件大小为0，可能下载失败"}
+                    
+                    # 创建一个新的临时文件
+                    temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.mp4')
+                    temp_file.close()
+                    shutil.copy2(file_path, temp_file.name)
+                    
+                    # 清理原始临时目录
+                    shutil.rmtree(temp_dir, ignore_errors=True)
+                    
+                    return {
+                        "status": "success",
+                        "file_path": temp_file.name,
+                        "title": clean_title,
+                        "ext": "mp4"
+                    }
+                else:
+                    shutil.rmtree(temp_dir, ignore_errors=True)
+                    return {"status": "error", "message": "下载文件不存在"}
+            except Exception as e:
+                error_msg = str(e)
+                # 如果是超时错误且进度不为0，继续下载
+                if ("timed out" in error_msg or "timeout" in error_msg) and progress_tracker.progress > 0:
+                    return {
+                        "status": "success",
+                        "file_path": file_path if 'file_path' in locals() else None,
+                        "title": clean_title,
+                        "ext": "mp4"
+                    }
+                shutil.rmtree(temp_dir, ignore_errors=True)
+                return {"status": "error", "message": f"下载过程中出错: {error_msg}"}
+                
+    except Exception as e:
+        if 'temp_dir' in locals():
+            shutil.rmtree(temp_dir, ignore_errors=True)
+        return {"status": "error", "message": str(e)}
+
+def download_video(urls, format_id=None):
+    """
+    下载视频并返回文件
+    """
+    if isinstance(urls, str):
+        urls = [url.strip() for url in urls.split('\n') if url.strip()]
+    
+    if not urls:
+        return "请输入至少一个视频链接", None, 0, "未开始下载"
+    
+    progress_tracker = DownloadProgress()
+    result = download_single_video(urls[0], format_id, progress_tracker)
+    
+    if result["status"] == "success":
+        try:
+            # 返回文件路径供Gradio处理下载
+            return "下载成功，正在传输...", result["file_path"], 100, "下载完成"
+        except Exception as e:
+            return f"文件处理失败: {str(e)}", None, 0, "下载失败"
+    else:
+        return f"下载失败: {result.get('message', '未知错误')}", None, 0, "下载失败"
+
+# 创建Gradio界面
+with gr.Blocks(title="视频下载工具", theme=gr.themes.Soft()) as demo:
+    # 存储视频信息的状态变量
+    video_info_state = gr.State({})
+    
+    with gr.Column(elem_id="header"):
+        gr.Markdown("""
+        # 🎥 视频下载工具
+        
+        一键下载各大平台视频，支持以下平台：
+        """)
+        
+        with gr.Row():
+            for platform in SUPPORTED_PLATFORMS.keys():
+                gr.Markdown(f"<span class='platform-badge'>{platform}</span>", elem_classes="platform")
+    
+    with gr.Row():
+        with gr.Column(scale=2):
+            # 输入部分
+            url_input = gr.Textbox(
+                label="视频链接",
+                placeholder="请输入视频链接，支持批量下载（每行一个链接）...",
+                lines=3,
+                info="支持多个平台的视频链接，自动识别平台类型"
+            )
+            parse_btn = gr.Button("解析视频", variant="secondary", size="lg")
+            
+            # 视频信息显示（使用Accordion组件）
+            with gr.Accordion("视频详细信息", open=False, visible=False) as video_info_accordion:
+                video_info = gr.JSON(show_label=False)
+            
+            format_choice = gr.Dropdown(
+                label="选择清晰度",
+                choices=[],
+                interactive=True,
+                visible=False
+            )
+            
+            download_btn = gr.Button("开始下载", variant="primary", size="lg", interactive=False)
+            
+        with gr.Column(scale=3):
+            # 预览和输出部分
+            with gr.Row():
+                preview_image = gr.Image(label="视频预览", visible=False)
+            with gr.Row():
+                progress = gr.Slider(
+                    minimum=0,
+                    maximum=100,
+                    value=0,
+                    label="下载进度",
+                    interactive=False
+                )
+                status = gr.Textbox(
+                    label="状态信息",
+                    value="等待开始下载...",
+                    interactive=False
+                )
+            # 使用File组件来处理下载
+            output_file = gr.File(label="下载文件")
+    
+    # 添加自定义CSS
+    gr.Markdown("""
+    <style>
+    #header {
+        text-align: center;
+        margin-bottom: 2rem;
+    }
+    .platform-badge {
+        display: inline-block;
+        padding: 0.5rem 1rem;
+        margin: 0.5rem;
+        border-radius: 2rem;
+        background-color: #2196F3;
+        color: white;
+        font-weight: bold;
+    }
+    .gradio-container {
+        max-width: 1200px !important;
+    }
+    .contain {
+        margin: 0 auto;
+        padding: 2rem;
+    }
+    .download-link {
+        display: inline-block;
+        padding: 0.8rem 1.5rem;
+        background-color: #4CAF50;
+        color: white;
+        text-decoration: none;
+        border-radius: 0.5rem;
+        margin-top: 1rem;
+        font-weight: bold;
+        transition: background-color 0.3s;
+    }
+    .download-link:hover {
+        background-color: #45a049;
+    }
+    </style>
+    """)
+    
+    def update_video_info(url):
+        """更新视频信息"""
+        # 只解析第一个链接
+        first_url = url.split('\n')[0].strip()
+        info = parse_video_info(first_url)
+        
+        if info["status"] == "success":
+            # 准备清晰度选项
+            format_choices = []
+            for fmt in info["formats"]:
+                # 构建格式标签
+                label_parts = []
+                if fmt['format_note']:
+                    label_parts.append(fmt['format_note'])
+                if fmt['resolution'] != 'unknown':
+                    label_parts.append(fmt['resolution'])
+                
+                label = " - ".join(filter(None, label_parts))
+                if not label:
+                    label = f"格式 {fmt['format_id']}"
+                
+                format_choices.append((label, fmt['format_id']))
+            
+            return [
+                gr.update(visible=True, value=info),  # video_info
+                gr.update(visible=True, choices=format_choices, value=format_choices[0][1] if format_choices else None),  # format_choice
+                gr.update(interactive=True),  # download_btn
+                gr.update(visible=True, value=info["thumbnail"]),  # preview_image
+                f"解析成功: {info['title']} ({info['platform']})",  # status
+                gr.update(visible=True)  # video_info_accordion
+            ]
+        else:
+            return [
+                gr.update(visible=False),  # video_info
+                gr.update(visible=False),  # format_choice
+                gr.update(interactive=False),  # download_btn
+                gr.update(visible=False),  # preview_image
+                info["message"],  # status
+                gr.update(visible=False)  # video_info_accordion
+            ]
+    
+    # 绑定解析按钮事件
+    parse_btn.click(
+        fn=update_video_info,
+        inputs=[url_input],
+        outputs=[video_info, format_choice, download_btn, preview_image, status, video_info_accordion]
+    )
+    
+    # 绑定下载按钮事件
+    download_btn.click(
+        fn=download_video,
+        inputs=[url_input, format_choice],
+        outputs=[status, output_file, progress, status]
+    )
+
+# 启���应用
+if __name__ == "__main__":
+    demo.launch()