如何在Celery中获取mega-cmd下载任务的进度？

你有三种可行的方案来获取mega-get在Celery任务中的下载进度，优先级从高到低如下：

1. 使用Mega官方/第三方Python SDK（最优方案） #

直接用Python SDK替代mega-cmd命令行工具，能在代码层面直接捕获下载进度，无需解析外部进程输出，稳定性和可控性更强。

以mega.py库为例，实现代码如下：

from celery import current_task
from mega import Mega

def download_via_mega_sdk(mega_url, save_path):
    # 初始化Mega客户端（需登录的话传入账号密码：mega.login(email, password)）
    mega_client = Mega()
    m = mega_client.login()
    
    # 获取目标文件对象
    target_file = m.get_file(mega_url)
    
    # 定义进度回调函数，实时更新Celery任务状态
    def progress_handler(downloaded_bytes, total_bytes):
        progress_percent = int((downloaded_bytes / total_bytes) * 100)
        current_task.update_state(
            state='PROGRESS',
            meta={
                'progress': progress_percent,
                'downloaded': downloaded_bytes,
                'total': total_bytes
            }
        )
    
    # 启动下载并绑定进度回调
    target_file.download(save_path, progress_callback=progress_handler)
    return {'status': 'completed', 'save_path': save_path}

注意：需先安装依赖pip install mega.py，部分SDK可能对Mega最新API的支持有延迟，使用前请确认版本兼容性。

2. 解析mega-get的输出（兼容性最好的方案） #

mega-get运行时会将进度信息输出到stderr（部分版本可能是stdout），通过subprocess.Popen实时捕获输出，用正则提取进度数据后更新Celery状态。

实现代码：

from celery import current_task
import subprocess
import re

def download_with_progress_parse(mega_url, save_path):
    # 启动mega-get进程，捕获错误输出（进度通常在这里）
    process = subprocess.Popen(
        ['mega-get', mega_url, save_path],
        stderr=subprocess.PIPE,
        stdout=subprocess.PIPE,
        text=True
    )
    
    # 适配mega-get的进度输出格式，正则匹配类似"Downloaded 123456 / 789012 bytes (15%)"的行
    progress_regex = re.compile(r'Downloaded (\d+) / (\d+) bytes \((\d+)%\)')
    
    # 实时读取输出并解析进度
    for line in iter(process.stderr.readline, ''):
        match_result = progress_regex.search(line)
        if match_result:
            downloaded = int(match_result.group(1))
            total = int(match_result.group(2))
            progress = int(match_result.group(3))
            
            current_task.update_state(
                state='PROGRESS',
                meta={
                    'progress': progress,
                    'downloaded': downloaded,
                    'total': total
                }
            )
    
    # 等待进程结束并检查返回码
    return_code = process.wait()
    if return_code != 0:
        raise Exception(f'mega-get执行失败，返回码：{return_code}')
    return {'status': 'completed', 'save_path': save_path}

注意：不同版本的mega-get输出格式可能略有差异，需根据实际输出调整正则表达式。

3. 监控下载文件的大小（兜底方案） #

如果前两种方案都无法实现，可以定期检查目标文件的大小，结合已知的文件总大小计算进度。这种方式精度和实时性稍差，但作为备选方案可行。

实现代码：

from celery import current_task
import subprocess
import os
import time
import re

def download_with_file_monitor(mega_url, save_path):
    # 先通过mega-info获取文件总大小
    info_process = subprocess.run(
        ['mega-info', mega_url],
        capture_output=True,
        text=True
    )
    size_regex = re.compile(r'Size: (\d+) bytes')
    size_match = size_regex.search(info_process.stdout)
    if not size_match:
        raise Exception('无法获取文件总大小')
    total_size = int(size_match.group(1))
    
    # 后台启动mega-get下载
    process = subprocess.Popen(
        ['mega-get', mega_url, save_path],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )
    
    # 每秒检查一次文件大小，更新进度
    while process.poll() is None:
        if os.path.exists(save_path):
            downloaded_size = os.path.getsize(save_path)
            progress = int((downloaded_size / total_size) * 100)
            current_task.update_state(
                state='PROGRESS',
                meta={
                    'progress': progress,
                    'downloaded': downloaded_size,
                    'total': total_size
                }
            )
        time.sleep(1)
    
    return_code = process.wait()
    if return_code != 0:
        raise Exception(f'mega-get执行失败，返回码：{return_code}')
    return {'status': 'completed', 'save_path': save_path}

注意：如果mega-get使用临时文件下载后再重命名，此方法会失效，需提前确认下载逻辑。

内容的提问来源于stack exchange，提问作者Daviid