临时有个批量图片处理需求，拉到本地处理太慢，试试
![图片[1]-测试下 Hetzner Cloud CAX41 16H32G320G (ARM)-THsInk](https://www.thsink.com/wp-content/uploads/2024/08/image-1024x340.png)
小时计费，价格24.49o/m(含ipv4 0.5o)，临时下载+处理任务很划算，主要是节省很多下载时间。
yabs
root@debian-32gb-nbg1-1:~# curl -sL yabs.sh | bash -s -- -r56
# ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## #
# Yet-Another-Bench-Script #
# v2024-06-09 #
# https://github.com/masonr/yet-another-bench-script #
# ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## #
Fri Aug 2 09:57:48 PM UTC 2024
ARM compatibility is considered *experimental*
Basic System Information:
---------------------------------
Uptime : 0 days, 8 hours, 8 minutes
Processor : Neoverse-N1
BIOS NotSpecified CPU @ 2.0GHz
CPU cores : 16 @ ??? MHz
AES-NI : ✔ Enabled
VM-x/AMD-V : ❌ Disabled
RAM : 30.6 GiB
Swap : 0.0 KiB
Disk : 300.4 GiB
Distro : Debian GNU/Linux 12 (bookworm)
Kernel : 6.1.0-23-arm64
VM Type : KVM
IPv4/IPv6 : ✔ Online / ✔ Online
IPv6 Network Information:
---------------------------------
ISP : Hetzner Online GmbH
ASN : AS24940 Hetzner Online GmbH
Host : Hetzner Online GmbH
Location : Nuremberg, Bavaria (BY)
Country : Germany
fio Disk Speed Tests (Mixed R/W 50/50) (Partition /dev/sda1):
---------------------------------
Block Size | 4k (IOPS) | 64k (IOPS)
------ | --- ---- | ---- ----
Read | 164.92 MB/s (41.2k) | 1.25 GB/s (19.6k)
Write | 164.81 MB/s (41.2k) | 1.29 GB/s (20.2k)
Total | 329.74 MB/s (82.4k) | 2.55 GB/s (39.8k)
| |
Block Size | 512k (IOPS) | 1m (IOPS)
------ | --- ---- | ---- ----
Read | 1.40 GB/s (2.7k) | 1.34 GB/s (1.3k)
Write | 1.52 GB/s (2.9k) | 1.49 GB/s (1.4k)
Total | 2.93 GB/s (5.7k) | 2.84 GB/s (2.7k)
iperf3 Network Speed Tests (IPv4):
---------------------------------
Provider | Location (Link) | Send Speed | Recv Speed | Ping
----- | ----- | ---- | ---- | ----
Clouvider | London, UK (10G) | 2.77 Gbits/sec | 4.67 Gbits/sec | 20.5 ms
Leaseweb | Singapore, SG (10G) | 2.36 Gbits/sec | 2.58 Gbits/sec | 163 ms
Leaseweb | NYC, NY, US (10G) | 2.58 Gbits/sec | 3.27 Gbits/sec | 92.9 ms
iperf3 Network Speed Tests (IPv6):
---------------------------------
Provider | Location (Link) | Send Speed | Recv Speed | Ping
----- | ----- | ---- | ---- | ----
Clouvider | London, UK (10G) | 2.75 Gbits/sec | 8.77 Gbits/sec | 20.7 ms
Leaseweb | Singapore, SG (10G) | 2.22 Gbits/sec | 7.43 Gbits/sec | 163 ms
Leaseweb | NYC, NY, US (10G) | 2.56 Gbits/sec | 8.58 Gbits/sec | 92.3 ms
Geekbench 5 Benchmark Test:
---------------------------------
Test | Value
|
Single Core | 858
Multi Core | 11796
Full Test | https://browser.geekbench.com/v5/cpu/22738360
Geekbench 6 Benchmark Test:
---------------------------------
Test | Value
|
Single Core | 1159
Multi Core | 9267
Full Test | https://browser.geekbench.com/v6/cpu/7168896
YABS completed in 11 min 40 sec脚本备份
功能：将src_dir下图片压缩为75%质量的jpg图片，保存到img_out_dir；将所有视频文件切片为640宽度的m3u8视频，保存到hls_out_dir；将其他文件复制到non_img_out_dir。可选是否清除元数据，自动多线程运行，任务完成后tg通知。
修改8、9行tg信息，113-116行文件夹信息
import os
import shutil
import subprocess
import concurrent.futures
import requests
# Telegram Bot Token和Chat ID
TELEGRAM_TOKEN = 'your_token'
CHAT_ID = 'your_id'
# 错误日志文件
LOG_FILE = './log.txt'
# 视频文件转换为 HLS 的函数
def convert_video_to_hls(video_path, src_dir, output_dir):
try:
# 确保输出目录存在
os.makedirs(output_dir, exist_ok=True)
# 设置相对路径
relative_path = os.path.relpath(video_path, src_dir)
base_filename = os.path.splitext(os.path.basename(video_path))[0]
hls_output_dir = os.path.join(output_dir, os.path.dirname(relative_path))
os.makedirs(hls_output_dir, exist_ok=True) # 创建嵌套目录
m3u8_file_path = os.path.join(hls_output_dir, f'{base_filename}.m3u8')
# 使用 FFmpeg 将视频转换为 HLS，最长边设置为 640，保持长宽比
command = [
'ffmpeg',
'-err_detect', 'ignore_err', # 忽略一些错误
'-max_interleave_delta', '0', # 处理某些文件的结构问题
'-i', video_path,
'-vf', 'scale=640:-1:force_original_aspect_ratio=increase', # 设置最大宽度为 640，保持长宽比
'-profile:v', 'baseline', # 设置视频编码为 baseline
'-level', '3.0', # 设置级别
'-start_number', '0', # 从第一个分片开始
'-hls_time', '10', # 每个分片的持续时间（秒）
'-hls_list_size', '0', # 包含所有分片
'-f', 'hls', # 使用 HLS 格式
m3u8_file_path
]
subprocess.run(command, check=True)
print(f'Converted video to HLS: {m3u8_file_path}')
except subprocess.CalledProcessError as e:
log_error(video_path, str(e))
except Exception as e:
log_error(video_path, f'Unexpected error: {e}')
def log_error(video_path, error_message):
with open(LOG_FILE, 'a') as log_file:
log_file.write(f'Error processing {video_path}: {error_message}\n')
# 压缩图片的函数
def compress_image(file_path, output_path, quality=75, remove_metadata=False):
try:
# 确保输出目录存在
os.makedirs(os.path.dirname(output_path), exist_ok=True)
# 使用 ImageMagick 进行图像压缩并可选地去除元数据
command = ['magick', file_path, '-quality', str(quality)]
if remove_metadata:
command.append('-strip')
command.append(output_path)
subprocess.run(command, check=True)
print(f'Compressed and saved: {output_path}')
except subprocess.CalledProcessError as e:
log_error(file_path, str(e))
except Exception as e:
log_error(file_path, f'Unexpected error: {e}')
# 复制非图片文件的函数
def copy_non_image(src_path, dst_path):
if os.path.isfile(src_path):
os.makedirs(os.path.dirname(dst_path), exist_ok=True)
shutil.copy2(src_path, dst_path)
print(f'Copied: {src_path} to {dst_path}')
# 遍历目录的函数
def process_directory(src_dir, img_out_dir, non_img_out_dir, hls_out_dir, quality, remove_metadata):
with concurrent.futures.ThreadPoolExecutor() as executor:
future_to_file = {}
for root, dirs, files in os.walk(src_dir):
for file in files:
file_path = os.path.join(root, file)
if file.lower().endswith(('.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff')):
# 为图片文件提交任务
relative_path = os.path.relpath(root, src_dir)
output_path = os.path.join(img_out_dir, relative_path, file)
future = executor.submit(compress_image, file_path, output_path, quality, remove_metadata)
future_to_file[future] = file_path
elif file.lower().endswith(('.mp4', '.avi', '.mov', '.mkv')):
# 为视频文件提交任务，转换为 HLS
future = executor.submit(convert_video_to_hls, file_path, src_dir, hls_out_dir)
future_to_file[future] = file_path
else:
# 为非图片和视频文件提交任务
relative_path = os.path.relpath(root, src_dir)
output_path = os.path.join(non_img_out_dir, relative_path, file)
future = executor.submit(copy_non_image, file_path, output_path)
future_to_file[future] = file_path
# 等待所有任务完成
for future in concurrent.futures.as_completed(future_to_file):
try:
future.result() # 获取结果，如果有异常会抛出
except Exception as exc:
log_error(future_to_file[future], str(exc))
# 删除空目录的函数
def remove_empty_directories(directory):
for root, dirs, files in os.walk(directory, topdown=False):
for dir in dirs:
dir_path = os.path.join(root, dir)
try:
os.rmdir(dir_path)
print(f'Removed empty directory: {dir_path}')
except OSError:
pass
# 发送 Telegram 消息
def send_telegram_message(message):
url = f'https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage'
payload = {'chat_id': CHAT_ID, 'text': message}
response = requests.post(url, data=payload)
if response.status_code == 200:
print("Message sent successfully!")
else:
print("Failed to send message.")
def main():
src_dir = '/home/l1angth6/qbittorrent/Downloads' # 输入目录，使用 Unix 风格路径
img_out_dir = '/root/sc/pic' # 输出图片目录
non_img_out_dir = '/root/sc/none' # 输出非图片目录
hls_out_dir = '/root/sc/video' # 输出 HLS 目录
# 询问用户是否删除文件属性和个人信息
remove_metadata_input = input("是否删除所有输出文件的元数据？(y/n): ").strip().lower()
remove_metadata = remove_metadata_input == 'y'
# 创建输出目录
os.makedirs(img_out_dir, exist_ok=True)
os.makedirs(non_img_out_dir, exist_ok=True)
os.makedirs(hls_out_dir, exist_ok=True)
# 使用 ThreadPoolExecutor 进行多线程处理
process_directory(src_dir, img_out_dir, non_img_out_dir, hls_out_dir, 75, remove_metadata)
# 删除不包含文件的目录
remove_empty_directories(non_img_out_dir)
remove_empty_directories(hls_out_dir)
# 发送完成消息
send_telegram_message("任务完成！所有图片已压缩，视频已转换为 HLS，非图片文件已复制。")
if __name__ == "__main__":
main()© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
暂无评论内容