app.py

import gradio as gr
import gdown
import os
import requests
import zipfile
from pathlib import Path
from urllib.parse import urlparse
import re
from datetime import datetime

# 创建下载目录
DOWNLOAD_DIR = Path("download")
DOWNLOAD_DIR.mkdir(exist_ok=True)

def is_google_drive_link(url):
    """检查是否为Google Drive链接"""
    patterns = [
        r'drive\.google\.com',
        r'docs\.google\.com'
    ]
    return any(re.search(pattern, url) for pattern in patterns)

def get_file_size_str(size_bytes):
    """将字节转换为可读的文件大小"""
    for unit in ['B', 'KB', 'MB', 'GB']:
        if size_bytes < 1024.0:
            return f"{size_bytes:.2f} {unit}"
        size_bytes /= 1024.0
    return f"{size_bytes:.2f} TB"

def compress_folder(folder_path):
    """压缩文件夹为zip文件"""
    zip_path = f"{folder_path}.zip"
    with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
        for root, dirs, files in os.walk(folder_path):
            for file in files:
                file_path = os.path.join(root, file)
                arcname = os.path.relpath(file_path, os.path.dirname(folder_path))
                zipf.write(file_path, arcname)
    return zip_path

def download_file(url, headers_dict=None, progress=gr.Progress()):
    """下载文件的主函数"""
    try:
        progress(0, desc="开始下载...")
        
        # 解析请求头
        headers = {}
        if headers_dict:
            for key, value in headers_dict.items():
                if key and value:
                    headers[key] = value
        
        # 判断是否为Google Drive链接
        if is_google_drive_link(url):
            progress(0.3, desc="检测到Google Drive链接，使用gdown下载...")
            
            # 使用gdown下载
            output_path = str(DOWNLOAD_DIR) + '/'
            
            # 检测是否为文件夹链接
            is_folder = 'folders' in url
            
            if is_folder:
                downloaded = gdown.download_folder(url, output=output_path, quiet=False)
                if downloaded:
                    folder_path = os.path.dirname(downloaded[0])
                    progress(0.8, desc="压缩文件夹...")
                    zip_path = compress_folder(folder_path)
                    file_size = os.path.getsize(zip_path)
                    return (
                        zip_path,
                        f"✅ 下载成功！\n文件: {os.path.basename(zip_path)}\n大小: {get_file_size_str(file_size)}"
                    )
            else:
                output_file = gdown.download(url, output=output_path, quiet=False, fuzzy=True)
                if output_file:
                    file_size = os.path.getsize(output_file)
                    return (
                        output_file,
                        f"✅ 下载成功！\n文件: {os.path.basename(output_file)}\n大小: {get_file_size_str(file_size)}"
                    )
                else:
                    return None, "❌ Google Drive下载失败"
        
        else:
            progress(0.3, desc="使用HTTP方式下载...")
            
            # 普通HTTP下载
            response = requests.get(url, headers=headers, stream=True, timeout=30)
            response.raise_for_status()
            
            # 获取文件名
            content_disposition = response.headers.get('content-disposition')
            if content_disposition:
                filename = re.findall('filename="(.+?)"', content_disposition)
                filename = filename[0] if filename else None
            else:
                filename = os.path.basename(urlparse(url).path)
            
            if not filename:
                filename = f"download_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
            
            # 保存文件
            file_path = DOWNLOAD_DIR / filename
            total_size = int(response.headers.get('content-length', 0))
            
            downloaded = 0
            with open(file_path, 'wb') as f:
                for chunk in response.iter_content(chunk_size=256*1024):
                    if chunk:
                        f.write(chunk)
                        downloaded += len(chunk)
                        if total_size:
                            progress(0.3 + 0.6 * (downloaded / total_size), 
                                   desc=f"下载中... {get_file_size_str(downloaded)}/{get_file_size_str(total_size)}")
            
            file_size = os.path.getsize(file_path)
            return (
                str(file_path),
                f"✅ 下载成功！\n文件: {filename}\n大小: {get_file_size_str(file_size)}"
            )
    
    except Exception as e:
        return None, f"❌ 下载失败: {str(e)}"

def get_downloaded_files():
    """获取已下载的文件列表"""
    files = []
    for item in DOWNLOAD_DIR.iterdir():
        if item.is_file():
            size = item.stat().st_size
            modified = datetime.fromtimestamp(item.stat().st_mtime).strftime('%Y-%m-%d %H:%M:%S')
            files.append({
                'name': item.name,
                'path': str(item),
                'size': get_file_size_str(size),
                'modified': modified
            })
    return sorted(files, key=lambda x: x['modified'], reverse=True)

def delete_file(file_path):
    """删除文件"""
    try:
        if os.path.exists(file_path):
            os.remove(file_path)
            return f"✅ 已删除: {os.path.basename(file_path)}", refresh_file_list()
        else:
            return "❌ 文件不存在", refresh_file_list()
    except Exception as e:
        return f"❌ 删除失败: {str(e)}", refresh_file_list()

def refresh_file_list():
    """刷新文件列表"""
    files = get_downloaded_files()
    if not files:
        return "📁 暂无已下载的文件"
    
    html = "<div style='max-height: 400px; overflow-y: auto;'>"
    for file in files:
        html += f"""
        <div style='border: 1px solid #ddd; padding: 10px; margin: 5px 0; border-radius: 5px;'>
            <strong>📄 {file['name']}</strong><br>
            <small>大小: {file['size']} | 修改时间: {file['modified']}</small><br>
            <small style='color: #666;'>路径: {file['path']}</small>
        </div>
        """
    html += "</div>"
    return html

def parse_headers(headers_text):
    """解析请求头文本"""
    headers_dict = {}
    if headers_text.strip():
        for line in headers_text.strip().split('\n'):
            if ':' in line:
                key, value = line.split(':', 1)
                headers_dict[key.strip()] = value.strip()
    return headers_dict

# 创建Gradio界面
with gr.Blocks(title="文件下载器", theme=gr.themes.Soft()) as app:
    gr.Markdown("# 🌐 多功能文件下载器")
    gr.Markdown("支持Google Drive和普通HTTP链接下载")
    
    with gr.Tab("📥 下载文件"):
        with gr.Row():
            with gr.Column(scale=2):
                url_input = gr.Textbox(
                    label="下载链接",
                    placeholder="请输入Google Drive链接或其他下载链接...",
                    lines=2
                )
                
                with gr.Accordion("⚙️ 高级选项", open=False):
                    gr.Markdown("### 自定义请求头（每行一个，格式: Header: Value）")
                    headers_input = gr.Textbox(
                        label="请求头",
                        placeholder="Cookie: your_cookie_here\nAuthorization: Bearer your_token_here\nUser-Agent: Custom User Agent",
                        lines=5
                    )
                    gr.Markdown("**常用示例:**\n- `Cookie: session=xxx`\n- `Authorization: Bearer token123`\n- `User-Agent: Mozilla/5.0`")
                
                download_btn = gr.Button("🚀 开始下载", variant="primary", size="lg")
            
            with gr.Column(scale=1):
                status_output = gr.Textbox(label="下载状态", lines=4)
                file_output = gr.File(label="下载的文件")
        
        # 下载按钮事件
        download_btn.click(
            fn=lambda url, headers: download_file(url, parse_headers(headers)),
            inputs=[url_input, headers_input],
            outputs=[file_output, status_output]
        )
    
    with gr.Tab("📂 文件浏览器"):
        gr.Markdown("### 已下载的文件")
        
        with gr.Row():
            refresh_btn = gr.Button("🔄 刷新列表", size="sm")
        
        file_list_html = gr.HTML(value=refresh_file_list())
        
        gr.Markdown("### 文件操作")
        with gr.Row():
            file_path_input = gr.Textbox(
                label="文件路径",
                placeholder="粘贴文件路径或从上面复制...",
                scale=3
            )
            delete_btn = gr.Button("🗑️ 删除文件", variant="stop", scale=1)
        
        operation_status = gr.Textbox(label="操作状态", lines=2)
        download_file_output = gr.File(label="下载文件")
        
        # 刷新按钮事件
        refresh_btn.click(
            fn=refresh_file_list,
            outputs=[file_list_html]
        )
        
        # 删除按钮事件
        delete_btn.click(
            fn=delete_file,
            inputs=[file_path_input],
            outputs=[operation_status, file_list_html]
        )
        
        # 点击文件路径自动加载文件用于下载
        file_path_input.change(
            fn=lambda path: path if os.path.exists(path) else None,
            inputs=[file_path_input],
            outputs=[download_file_output]
        )
    
    gr.Markdown("""
    ---
    ### 📖 使用说明
    
    **下载功能:**
    - 自动识别Google Drive链接（支持文件和文件夹）
    - 支持自定义请求头（Cookie、Token等）
    - 文件夹会自动压缩为ZIP格式
    
    **文件浏览器:**
    - 查看所有已下载的文件
    - 直接下载文件到本地
    - 删除不需要的文件
    
    **支持的链接格式:**
    - Google Drive: `https://drive.google.com/file/d/xxx` 或 `https://drive.google.com/folders/xxx`
    - 普通HTTP/HTTPS下载链接
    """)

# 启动应用
if __name__ == "__main__":
    app.launch(
        auth=(os.environ.get('username'), os.environ.get('password')),
        ssr_mode=False
    )