你应该检查命令行参数是否合法，如果不是合法时间要打出帮助并退出
#!/usr/bin/env python3
import os
import subprocess
import re
import argparse
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子目录名（末尾加/）
LAST_RUN_TIME_FILE = "/home/nick/diabloforum/tools/data/last_run_time.txt" # 上次运行时间记录文件
# -------------------------- 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):
"""自然排序键生成函数"""
name_without_ext = os.path.splitext(filename)[0]
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
return (main_part, number_part)
def get_reference_time(time_arg):
"""获取参考时间：优先使用命令行参数，其次使用上次运行时间，否则使用纪元时间"""
# 命令行参数优先
if time_arg:
try:
return datetime.strptime(time_arg, "%Y-%m-%d %H:%M:%S")
except ValueError:
print(f"⚠️ 时间格式错误，使用默认格式: YYYY-MM-DD HH:MM:SS")
# 尝试读取上次运行时间
if os.path.exists(LAST_RUN_TIME_FILE):
try:
with open(LAST_RUN_TIME_FILE, 'r') as f:
time_str = f.read().strip()
return datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S")
except (ValueError, IOError):
print(f"⚠️ 读取上次运行时间失败，使用纪元时间")
# 默认返回纪元时间
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}")
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"<li><a href=\"{s3_mp4_url}\">{filename} ({timestamp})</a></li>"
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_mp4)
link_line = f"<li><a href=\"{s3_mp3_url}\">{mp3_filename} ({timestamp})</a></li>"
safe_exec(f"echo '{link_line}' >> {MP3_LINK_FILE}")
print("[步骤4] 已添加链接到列表")
def main():
# 解析命令行参数
parser = argparse.ArgumentParser(description='上传MP4文件到S3并生成MP3')
parser.add_argument('--since', help='只处理在此时间之后创建的文件，格式: "YYYY-MM-DD HH:MM:SS"')
args = parser.parse_args()
# 获取参考时间
reference_time = get_reference_time(args.since)
print(f"⏱️ 参考时间: {reference_time.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"📌 只处理此时间之后创建的文件")
# 初始化链接文件
print("========================================")
if DRY_RUN:
print("🔴 当前为DRY RUN测试模式，不会实际执行操作")
else:
print("🟢 当前为正式模式，将执行实际操作")
# 清空并初始化链接文件
safe_exec(f"> {LINK_FILE}")
safe_exec(f"echo '<ol>' > {LINK_FILE}")
safe_exec(f"> {MP3_LINK_FILE}")
safe_exec(f"echo '<ol>' > {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"):
file_path = os.path.join(LOCAL_DIR, file)
# 获取文件创建时间
create_time = datetime.fromtimestamp(os.path.getctime(file_path))
# 只添加创建时间在参考时间之后的文件
if create_time > reference_time:
mp4_files.append(file_path)
# 使用自然排序
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 '</ol>' >> {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 '</ol>' >> {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'"
)
# 保存当前运行时间
save_current_run_time()
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()