#!/usr/bin/env python3 import os import subprocess import re from datetime import datetime # -------------------------- 1. 配置参数（必改） -------------------------- LOCAL_DIR = "/home/nick/.wine/drive_c/users/nick/Documents/wechat" # 本地视频文件夹 S3_BUCKET = "www.staroceans.org" # S3桶名 S3_DIR = "video/wechat/" # S3目标目录（末尾加/） MP3_LOCAL_SUBDIR = "mp3" # 本地MP3子目录名 MP3_S3_SUBDIR = "mp3/" # S3上MP3子目录名（末尾加/） # -------------------------- 2. 测试模式开关（选填） -------------------------- DRY_RUN = False # 测试时改True（只打印不执行），正式上传改False # -------------------------- 3. 链接文件路径配置 -------------------------- 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): """安全执行命令，根据DRY_RUN决定是否实际执行""" print(f"[{'DRY RUN' if DRY_RUN else '实际执行'}] {cmd}") if not DRY_RUN: result = subprocess.run( cmd, shell=shell, check=check, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) return result return None def get_file_md5(file_path): """获取文件MD5值""" result = safe_exec(f"md5sum {file_path}", check=True) if result: return result.stdout.split()[0] return "模拟MD5" # 测试模式返回占位符 def s3_file_exists(s3_path): """检查S3上文件是否存在""" result = safe_exec(f"s3cmd info {s3_path}", check=False) if DRY_RUN: return False # 测试模式默认不存在 return result.returncode == 0 def get_s3_md5(s3_path): """获取S3文件的MD5值""" 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): """获取文件最后修改时间戳（DRY_RUN模式返回模拟时间）""" 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): """ 自然排序键生成函数： 1. 先按文件名主体（非数字部分）的字母顺序排序 2. 再按文件名末尾的数字后缀（如"10"）的数值大小排序 例如："硅谷七姐妹分析之英伟达9.mp4" 会排在 "硅谷七姐妹分析之英伟达10.mp4" 前面 """ # 移除文件后缀（.mp4） name_without_ext = os.path.splitext(filename)[0] # 正则匹配：分离文件名中的"非数字部分"和"末尾数字部分" # 例如："硅谷七姐妹分析之微软10" → 非数字部分"硅谷七姐妹分析之微软"，数字部分10 match = re.match(r'(.*?)(\d+)$', name_without_ext) if match: main_part = match.group(1) # 文件名主体（非数字部分） number_part = int(match.group(2)) # 末尾数字（转为整数用于排序） else: main_part = name_without_ext # 无数字后缀，主体为完整文件名 number_part = 0 # 无数字时用0 # 返回排序键：(主体部分, 数字部分) # 排序时先按主体字母顺序，再按数字大小 return (main_part, number_part) def process_mp4(local_mp4): """处理MP4文件：检查S3存在性、上传、生成链接""" 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} ===") print(f"[步骤1] 检查S3是否存在: {s3_mp4_path}") # 检查S3存在性 s3_exists = s3_file_exists(s3_mp4_path) s3_md5_check = False # 可改为True启用MD5校验 # MD5校验逻辑 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 " f"'{local_mp4}' '{s3_mp4_path}'" ) # 生成链接 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): """处理MP3文件：生成、检查S3存在性、上传、生成链接""" 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} ===") # 确保本地MP3目录存在 safe_exec(f"mkdir -p {local_mp3_dir}") # 检查并生成本地MP3 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文件") # 检查S3存在性 print(f"[步骤2] 检查S3是否存在: {s3_mp3_path}") s3_exists = s3_file_exists(s3_mp3_path) s3_md5_check = False # 可改为True启用MD5校验 # MD5校验逻辑 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 " f"'{local_mp3}' '{s3_mp3_path}'" ) # 生成链接 timestamp = get_file_timestamp(local_mp3) link_line = f"
  • {mp3_filename} ({timestamp})
    • " safe_exec(f"echo '{link_line}' >> {MP3_LINK_FILE}") print("[步骤4] 已添加链接到列表") def main(): # 初始化链接文件 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}") print(f"本地目录: {LOCAL_DIR}") print(f"本地MP3目录: {os.path.join(LOCAL_DIR, MP3_LOCAL_SUBDIR)}") print(f"S3 MP4目标: s3://{S3_BUCKET}/{S3_DIR}") print(f"S3 MP3目标: s3://{S3_BUCKET}/{S3_DIR}{MP3_S3_SUBDIR}") print("========================================") # 获取并排序本地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))) # DRY_RUN模式下打印排序后的所有文件名（调试用） if DRY_RUN: print("\n📋 按处理顺序排列的MP4文件列表：") for idx, file in enumerate(mp4_files, 1): print(f" {idx}. {os.path.basename(file)}") print(f"\n📊 共发现 {len(mp4_files)} 个MP4文件") # 处理每个MP4文件 processed_count = 0 for mp4_file in mp4_files: processed_count += 1 print(f"\n===== 开始处理第 {processed_count}/{len(mp4_files)} 个文件 =====") process_mp4(mp4_file) process_mp3(mp4_file) print("----------------------------------------") # 生成最终HTML文件 print("\n[生成HTML列表]") # 处理MP4 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 --continue --check-md5 " f"'{VIDEO_FILE}' 's3://{S3_BUCKET}/tools/data/video_list.html'" ) # 处理MP3 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 --continue --check-md5 " f"'{MP3_VIDEO_FILE}' 's3://{S3_BUCKET}/tools/data/audio_list.html'" ) print("\n🎉 所有文件处理完成！") print(f"📊 处理统计：共发现 {len(mp4_files)} 个MP4文件，实际处理 {processed_count} 个") if DRY_RUN: print("🔴 注意：测试模式未执行实际操作，请核对排序是否符合预期") else: print(f"🟢 MP4链接列表: {LINK_FILE}") print(f"🟢 MP3链接列表: {MP3_LINK_FILE}") if __name__ == "__main__": main()