#!/usr/bin/env python3 import os import subprocess import re import argparse from datetime import datetime # -------------------------- 配置参数（不变） -------------------------- LOCAL_DIR = "/home/nick/.wine/drive_c/users/nick/Documents/wechat" S3_BUCKET = "www.staroceans.org" S3_DIR = "video/wechat/" MP3_LOCAL_SUBDIR = "mp3" MP3_S3_SUBDIR = "mp3/" LAST_RUN_TIME_FILE = "/home/nick/diabloforum/tools/data/last_run_time.txt" DRY_RUN = False LINK_FILE = "/home/nick/diabloforum/tools/data/s3_video_links.txt" MP3_LINK_FILE = "/home/nick/diabloforum/tools/data/s3_audio_links.txt" TOP_FILE = "/home/nick/diabloforum/tools/data/top.txt" BOTTOM_FILE = "/home/nick/diabloforum/tools/data/bottom.txt" VIDEO_FILE = "/home/nick/diabloforum/tools/data/video_list.html" MP3_VIDEO_FILE = "/home/nick/diabloforum/tools/data/audio_list.html" # -------------------------- 工具函数（不变） -------------------------- def safe_exec(cmd, shell=True, check=True): print(f"[{'DRY RUN' if DRY_RUN else '实际执行'}] {cmd}") if not DRY_RUN: return subprocess.run(cmd, shell=shell, check=check, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) return None def get_file_md5(file_path): result = safe_exec(f"md5sum {file_path}", check=True) return result.stdout.split()[0] if result else "模拟MD5" def s3_file_exists(s3_path): result = safe_exec(f"s3cmd info {s3_path}", check=False) return False if DRY_RUN else result.returncode == 0 def get_s3_md5(s3_path): result = safe_exec(f"s3cmd info {s3_path}", check=True) if result: for line in result.stdout.splitlines(): if "MD5 sum" in line: return line.split(": ")[1].strip() return "模拟S3 MD5" def get_file_timestamp(file_path): if DRY_RUN: return datetime.now().strftime("%Y-%m-%d %H:%M:%S") if not os.path.exists(file_path): raise FileNotFoundError(f"文件不存在: {file_path}") return datetime.fromtimestamp(os.path.getmtime(file_path)).strftime("%Y-%m-%d %H:%M:%S") def natural_sort_key(filename): name_without_ext = os.path.splitext(filename)[0] match = re.match(r'(.*?)(\d+)$', name_without_ext) return (match.group(1), int(match.group(2))) if match else (name_without_ext, 0) def get_reference_time(time_arg, parser): if time_arg: try: return datetime.strptime(time_arg, "%Y-%m-%d %H:%M:%S") except ValueError: parser.error(f"无效的时间格式: '{time_arg}'，正确格式应为: YYYY-MM-DD HH:MM:SS") if os.path.exists(LAST_RUN_TIME_FILE): try: with open(LAST_RUN_TIME_FILE, 'r') as f: return datetime.strptime(f.read().strip(), "%Y-%m-%d %H:%M:%S") except (ValueError, IOError) as e: print(f"⚠️ 读取上次运行时间失败: {e}，将使用纪元时间作为参考") return datetime.fromtimestamp(0) def save_current_run_time(): if DRY_RUN: print(f"[DRY RUN] 模拟保存当前时间到 {LAST_RUN_TIME_FILE}") return current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") try: with open(LAST_RUN_TIME_FILE, 'w') as f: f.write(current_time) print(f"✅ 已保存当前运行时间: {current_time}") except IOError as e: print(f"⚠️ 保存运行时间失败: {e}") # -------------------------- 处理函数（添加skip_uploading参数） -------------------------- def process_mp4(local_mp4, skip_uploading): """处理MP4文件：skip_uploading=True时仅生成列表，跳过上传""" filename = os.path.basename(local_mp4) s3_mp4_path = f"s3://{S3_BUCKET}/{S3_DIR}{filename}" s3_mp4_url = f"http://{S3_BUCKET}/{S3_DIR}{filename}" print(f"\n=== 处理MP4文件：{filename} ===") # 仅当不跳过上传时，执行检查和上传逻辑 if not skip_uploading: print(f"[步骤1] 检查S3是否存在: {s3_mp4_path}") s3_exists = s3_file_exists(s3_mp4_path) s3_md5_check = False if s3_exists and s3_md5_check: local_md5 = get_file_md5(local_mp4) s3_md5 = get_s3_md5(s3_mp4_path) print(f"[对比] 本地MD5: {local_md5} | S3 MD5: {s3_md5}") if local_md5 != s3_md5: print("[结论] MD5不一致，需要重新上传") s3_exists = False else: print("[结论] MD5一致，跳过上传") if not s3_exists: print("[步骤2] 开始上传到S3") safe_exec(f"s3cmd put --mime-type=\"video/mp4\" --acl-public --continue --check-md5 '{local_mp4}' '{s3_mp4_path}'") else: print("[跳过上传] 仅生成列表链接") # 无论是否跳过上传，均生成列表链接（核心需求） timestamp = get_file_timestamp(local_mp4) link_line = f"
  • {filename} ({timestamp})
    • " safe_exec(f"echo '{link_line}' >> {LINK_FILE}") print("[步骤3] 已添加链接到列表") def process_mp3(local_mp4, skip_uploading): """处理MP3文件：skip_uploading=True时仅生成列表，跳过转码和上传""" filename = os.path.basename(local_mp4) filename_noext = os.path.splitext(filename)[0] mp3_filename = f"{filename_noext}.mp3" local_mp3_dir = os.path.join(LOCAL_DIR, MP3_LOCAL_SUBDIR) local_mp3 = os.path.join(local_mp3_dir, mp3_filename) s3_mp3_path = f"s3://{S3_BUCKET}/{S3_DIR}{MP3_S3_SUBDIR}{mp3_filename}" s3_mp3_url = f"http://{S3_BUCKET}/{S3_DIR}{MP3_S3_SUBDIR}{mp3_filename}" print(f"\n=== 处理MP3文件：{mp3_filename} ===") # 仅当不跳过上传时，执行转码和上传逻辑 if not skip_uploading: safe_exec(f"mkdir -p {local_mp3_dir}") print(f"[步骤1] 检查本地MP3: {local_mp3}") if not os.path.exists(local_mp3) or DRY_RUN: if not DRY_RUN and not os.path.exists(local_mp3): print("[生成] 本地不存在，使用ffmpeg提取音频") safe_exec(f"ffmpeg -y -i '{local_mp4}' -vn -acodec libmp3lame '{local_mp3}'") else: print("[DRY RUN] 模拟生成MP3文件") print(f"[步骤2] 检查S3是否存在: {s3_mp3_path}") s3_exists = s3_file_exists(s3_mp3_path) s3_md5_check = False if s3_exists and s3_md5_check: local_md5 = get_file_md5(local_mp3) s3_md5 = get_s3_md5(s3_mp3_path) print(f"[对比] 本地MD5: {local_md5} | S3 MD5: {s3_md5}") if local_md5 != s3_md5: print("[结论] MD5不一致，需要重新上传") s3_exists = False else: print("[结论] MD5一致，跳过上传") if not s3_exists: print("[步骤3] 开始上传到S3") safe_exec(f"s3cmd put --mime-type=\"audio/mpeg\" --acl-public --continue --check-md5 '{local_mp3}' '{s3_mp3_path}'") else: print("[跳过上传] 仅生成列表链接") # 无论是否跳过上传，均生成列表链接（核心需求） timestamp = get_file_timestamp(local_mp4) link_line = f"
  • {mp3_filename} ({timestamp})
    • " safe_exec(f"echo '{link_line}' >> {MP3_LINK_FILE}") print("[步骤4] 已添加链接到列表") # -------------------------- 主函数（控制skip_uploading参数） -------------------------- def main(): parser = argparse.ArgumentParser( description='上传MP4文件到S3并生成MP3', formatter_class=argparse.RawTextHelpFormatter ) parser.add_argument( '--since', help='只处理在此时间之后创建的文件，格式必须为:\n YYYY-MM-DD HH:MM:SS\n例如:\n --since "2023-10-01 08:30:00"' ) args = parser.parse_args() reference_time = get_reference_time(args.since, parser) print(f"⏱️ 参考时间: {reference_time.strftime('%Y-%m-%d %H:%M:%S')}") print("========================================") if DRY_RUN: print("🔴 当前为DRY RUN测试模式") else: print("🟢 当前为正式模式") # 清空并初始化链接文件（确保所有文件重新生成列表） safe_exec(f"> {LINK_FILE}") safe_exec(f"echo '
      ' > {LINK_FILE}") safe_exec(f"> {MP3_LINK_FILE}") safe_exec(f"echo '
        ' > {MP3_LINK_FILE}") # 1. 收集所有MP4文件（不筛选时间，全部加入列表） mp4_files = [] for file in os.listdir(LOCAL_DIR): if file.lower().endswith(".mp4"): mp4_files.append(os.path.join(LOCAL_DIR, file)) # 存储绝对路径 mp4_files.sort(key=lambda x: natural_sort_key(os.path.basename(x))) # 2. 处理所有文件：根据时间决定是否跳过上传 for mp4_file in mp4_files: filename = os.path.basename(mp4_file) create_time = datetime.fromtimestamp(os.path.getctime(mp4_file)) # 用绝对路径取创建时间 # 核心逻辑：创建时间早于参考时间则跳过上传（但仍生成列表） skip_uploading = create_time <= reference_time print(f"\n===== 处理文件：{filename}（{'跳过上传' if skip_uploading else '正常处理'}） =====") process_mp4(mp4_file, skip_uploading) # 传入跳过参数 process_mp3(mp4_file, skip_uploading) # 传入跳过参数 # 生成最终HTML（所有文件均在列表中） print("\n[生成HTML列表]") safe_exec(f"echo '
      ' >> {LINK_FILE}") safe_exec(f"cat {TOP_FILE} > {VIDEO_FILE}") safe_exec(f"cat {LINK_FILE} >> {VIDEO_FILE}") safe_exec(f"cat {BOTTOM_FILE} >> {VIDEO_FILE}") safe_exec(f"s3cmd put --mime-type=\"text/html\" --acl-public '{VIDEO_FILE}' 's3://{S3_BUCKET}/tools/data/video_list.html'") safe_exec(f"echo '
    ' >> {MP3_LINK_FILE}") safe_exec(f"cat {TOP_FILE} > {MP3_VIDEO_FILE}") safe_exec(f"cat {MP3_LINK_FILE} >> {MP3_VIDEO_FILE}") safe_exec(f"cat {BOTTOM_FILE} >> {MP3_VIDEO_FILE}") safe_exec(f"s3cmd put --mime-type=\"text/html\" --acl-public '{MP3_VIDEO_FILE}' 's3://{S3_BUCKET}/tools/data/audio_list.html'") save_current_run_time() print(f"\n🎉 所有文件处理完成，均已加入列表") if __name__ == "__main__": main()