atlas/synthmind/nas_audio_extract_v3.py

#!/usr/bin/env python3
"""
NAS 视频音频批量提取 v3
流程: ssh cat .mp4 → 本地文件 → FFmpeg → shell重定向上传 → 删除本地
"""

import subprocess
import os
import time

NAS_BASE = "/volume2/work/Public Cloud Learning Sessions"
MAC_MP4  = os.path.expanduser("~/.openclaw/temp/xingshu/nas_mp4")
MAC_MP3  = os.path.expanduser("~/.openclaw/temp/xingshu/nas_mp3_out")
MAC_LOG  = os.path.expanduser("~/.openclaw/temp/xingshu/logs/nas_audio_v3.log")
PROGRESS = os.path.expanduser("~/.openclaw/temp/xingshu/nas_audio_v3.done")
SSH_USER = "shenwei"
NAS_HOST = "192.168.3.17"

os.makedirs(MAC_MP4, exist_ok=True)
os.makedirs(MAC_MP3, exist_ok=True)

FFMPEG_BIN = "/opt/homebrew/bin/ffmpeg"
FFMPEG_CMD = [FFMPEG_BIN, "-y", "-i", "pipe:0",
              "-vn", "-acodec", "libmp3lame", "-ab", "64k", "-ar", "22050", "-ac", "1",
              "-f", "mp3", "pipe:1"]

def log(msg):
    ts = time.strftime("%H:%M:%S")
    line = f"[{ts}] {msg}"
    print(line)
    with open(MAC_LOG, "a") as f:
        f.write(line + "\n")

def ssh_download(nas_path, local_path):
    """NAS 文件通过 ssh cat 落地到本地（1MB 分块）"""
    cmd = ["ssh", f"{SSH_USER}@{NAS_HOST}", f"cat '{nas_path}'"]
    with open(local_path, "wb") as fout:
        proc = subprocess.Popen(cmd, stdout=subprocess.PIPE)
        while True:
            chunk = proc.stdout.read(1024 * 1024)
            if not chunk:
                break
            fout.write(chunk)
        proc.stdout.close()
        proc.wait()
    return proc.returncode == 0

def ssh_upload(local_path, nas_path):
    """本地文件通过 shell 重定向上传到 NAS（本地路径需加引号防空格）"""
    cmd = f'ssh {SSH_USER}@{NAS_HOST} "cat > \'{nas_path}\'" < "{local_path}"'
    return subprocess.run(cmd, shell=True, stdout=subprocess.DEVNULL,
                          stderr=subprocess.DEVNULL).returncode == 0

def get_nas_mp3s():
    r = subprocess.run(["ssh", f"{SSH_USER}@{NAS_HOST}",
                       f"ls '{NAS_BASE}'/*.mp3 2>/dev/null"],
                      capture_output=True, text=True)
    if r.returncode != 0:
        return set()
    return set(l.strip() for l in r.stdout.split("\n") if l.strip())

def get_nas_mp4s():
    r = subprocess.run(["ssh", f"{SSH_USER}@{NAS_HOST}",
                       f"ls '{NAS_BASE}'/*.mp4 2>/dev/null"],
                      capture_output=True, text=True)
    if r.returncode != 0:
        return []
    return [l.strip() for l in r.stdout.split("\n") if l.strip()]

def load_done():
    if os.path.exists(PROGRESS):
        with open(PROGRESS) as f:
            return set(f.read().splitlines())
    return set()

def save_done(done_set):
    with open(PROGRESS, "w") as f:
        f.write("\n".join(done_set))

def main():
    log("=" * 60)
    log("🚀 v3 开始批量提取")
    
    done = load_done()
    nas_mp3s = get_nas_mp3s()
    all_mp4s = get_nas_mp4s()
    
    pending = []
    for v in all_mp4s:
        name = os.path.basename(v)
        mp3_name = name.replace(".mp4", ".mp3")
        nas_mp3_path = f"{NAS_BASE}/{mp3_name}"
        if name not in done and nas_mp3_path not in nas_mp3s:
            pending.append(name)
    
    total = len(all_mp4s)
    done_count = total - len(pending)
    log(f"📊 总体: {done_count}/{total} 已完成，{len(pending)} 待处理")
    
    if not pending:
        log("✅ 全部已完成")
        return
    
    success, failed = 0, []
    
    for i, video in enumerate(pending, 1):
        mp3_name = video.replace(".mp4", ".mp3")
        nas_mp4 = f"{NAS_BASE}/{video}"
        nas_mp3 = f"{NAS_BASE}/{mp3_name}"
        local_mp4 = f"{MAC_MP4}/{video}"
        local_mp3 = f"{MAC_MP3}/{mp3_name}"
        
        log(f"\n[{i}/{len(pending)}] ▶ {video}")
        
        try:
            # Step 1: 下载 .mp4 → 本地
            log(f"  📥 downloading...")
            if not ssh_download(nas_mp4, local_mp4):
                raise RuntimeError("download failed")
            size_mp4 = os.path.getsize(local_mp4)
            log(f"  ✅ downloaded {size_mp4//1024//1024}MB")
            
            # Step 2: FFmpeg 提取 .mp3
            log(f"  🔊 extracting...")
            with open(local_mp3, "wb") as fout:
                proc = subprocess.Popen(["/bin/cat", local_mp4],
                                       stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
                ffmpeg = subprocess.Popen(FFMPEG_CMD,
                                          stdin=proc.stdout, stdout=fout, stderr=subprocess.DEVNULL)
                proc.stdout.close()
                ret = ffmpeg.wait()
                proc.wait()
            
            if ret != 0:
                raise RuntimeError("ffmpeg failed")
            
            # 删除 .mp4 释放空间
            os.remove(local_mp4)
            
            size_mp3 = os.path.getsize(local_mp3)
            log(f"  ✅ extracted {size_mp3//1024//1024}MB ({size_mp3//1024}KB)")
            
            # Step 3: 上传 .mp3 → NAS
            log(f"  📤 uploading...")
            if not ssh_upload(local_mp3, nas_mp3):
                raise RuntimeError("upload failed")
            log(f"  ✅ uploaded to NAS")
            
            # 删除 .mp3
            os.remove(local_mp3)
            
            done.add(video)
            save_done(done)
            success += 1
            
            log(f"  📊 进度: {done_count + success}/{total}")
            
        except Exception as e:
            log(f"  ❌ {e}")
            failed.append(video)
            for p in [local_mp4, local_mp3]:
                if os.path.exists(p):
                    os.remove(p)
    
    log(f"\n{'='*60}")
    log(f"🏁 完成: {success}/{len(pending)}，失败: {len(failed)}")
    if failed:
        log(f"失败列表: {failed}")

if __name__ == "__main__":
    main()