Matrix 首页推荐
Matrix 是少数派的写作社区,我们主张分享真实的产品体验,有实用价值的经验与思考。我们会不定期挑选 Matrix 最优质的文章,展示来自用户的最真实的体验和观点。
文章代表作者个人观点,少数派仅对标题和排版略作修改。
背景
我使用 Syncthing 在我的多个设备间(安卓手机、笔记本、台式机和 NAS)进行媒体文件的同步。选择 Syncthing 的原因有两点。相比 Nextcloud 这类功能全面的私有云,在除去掉不必要的多用户管理、文件分享和在线协作等功能后,Syncthing 更纯粹和轻量,专注于文件同步。而 Immich 这类自托管的媒体服务虽然提供了安卓客户端的备份功能,但它的文件存储结构由应用管理,不适合我这种希望保留原始文件夹结构的用户。
因此,我的最终方案是,使用 Syncthing 进行同步,并配合 Immich 的外部图库功能来进行管理。
然而,Syncthing 同步方案在安卓设备上存在一个问题:安卓的媒体文件(照片和视频、系统截图以及应用的图片等)散落在 DCIM、Pictures、Movies、Download 多个不同的目录下。固然可以为这些顶层目录各建立一个同步任务,但这样就会同步很多不必要的文件(例如放入回收站的媒体文件和应用临时产生的缩略图或缓存文件)。而如果为每一个需要同步的子文件,都手动建立一个同步任务,管理起来又过于繁琐了。自然而然地,一个想法产生了:创建一个统一的中转文件夹,然后定期将散落的媒体文件移动到这个中转站中,之后只需要让这个中转站保持同步即可。
方案
首先是脚本的执行方案。要在安卓手机上运行这个脚本,我立即想到了通过 Termux 来执行定时任务。折腾过安卓的人多少可能都听说过这个软件。简单的说,这是一个终端模拟器,无需 Root 和额外设置,只需要安装完软件就可以为安卓手机提供一个 Linux 环境。再通过一些设置和扩展,Termux 就可以访问设备的存储目录并且实现和系统的良好集成。用它来执行这个脚本再合适不过了。
然后是脚本的实现方式。固然可以编写一个纯粹的 Shell 脚本,通过后缀名来判断文件类型并执行移动操作,但这种识别方式并不可靠。更准确的方式是,通过读取文件的魔数(Magic Number),即文件开头的一串固定字节序列,来判断文件的类型,类 UNIX 系统中的 file
命令就是通过它来工作的。我选择了 Python 来实现脚本。Python 的 python-magic 库可以轻松地读取文件的魔数判断类型。在复杂的处理逻辑时(如排除特定的目录、忽略隐藏文件),Python 的代码结构和可读性也远优于 Shell 脚本。
实现
准备同步文件夹
首先,在安卓设备上创建一个单独的文件夹,作为 Syncthing 同步媒体文件的中转站,例如 /storage/emulated/0/Syncthing/pictures
(即安卓存储目录下的 Syncthing 目录中的 pictures 文件夹,Syncthing 文件夹与 Download、DCIM 等文件夹处于同一级)。然后在 Syncthing 应用中,把这个文件夹设置为需要监控的媒体目录,也是唯一需要监控的媒体目录。安卓上的 Syncthing 应用可以使用这个。
编写工具脚本
脚本工具的逻辑并不复杂:通过递归扫描指定路径,然后排除忽略的文件和目录并识别出需要处理的媒体文件,最后再执行移动到目标目录的操作即可。
以下是一个完整的 Python 实现示例代码:
```python
import os
import shutil
import datetime
import pathlib # 使用 pathlib 来处理路径
import magic # 使用 python-magic 识别媒体文件
import subprocess
# --- 配置 ---
HOME_DIR = pathlib.Path.home()
# 日志
LOG_DIR = HOME_DIR / "logs"
LOG_FILE = LOG_DIR / "scan_and_move_media.log"
# 安卓存储目录在 Termux 中的映射
SHARED_DIR = HOME_DIR / "storage/shared"
# 目标目录
TARGET_DIR = SHARED_DIR / "Syncthing/pictures"
# 扫描的源目录
INCLUDED_DIRS_CANDIDATES = ["DCIM", "Pictures", "Movies", "Download"]
# 排除的目录
EXCLUDED_SUBDIRS = {".thumbnails", "cache"}
# 媒体文件类型
MEDIA_MIME_PREFIXES = ("image/", "video/")
def log(message: str, level: str = "INFO", indent: int = 0):
"""
结构化的日志记录函数。
- level: 日志级别 (INFO, WARN, ERROR, DEBUG)
- indent: 日志缩进层级,用于美化输出
"""
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
indent_str = " " * indent
log_message = f"{timestamp} [{level.upper():<5}] {indent_str}{message}\n"
try:
# 确保日志目录存在
LOG_DIR.mkdir(parents=True, exist_ok=True)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(log_message)
except Exception as e:
print(f"CRITICAL: Failed to write to log file {LOG_FILE}: {e}")
print(log_message)
def find_media_to_move() -> list:
"""
扫描文件系统,找出所有需要移动的媒体文件。
此函数只读,不修改任何文件。
返回一个包含 (源路径,目标路径) 元组的列表。
"""
log("Scan phase started.", level="INFO")
move_operations = []
for dir_name in INCLUDED_DIRS_CANDIDATES:
scan_dir = SHARED_DIR / dir_name
if not scan_dir.is_dir():
log(f"Skipping non-existent directory: {scan_dir}", level="WARN")
continue
log(f"Scanning directory: {scan_dir}", indent=1)
for root, dirs, files in os.walk(scan_dir, topdown=True):
current_root = pathlib.Path(root)
# 跳过 .nomedia 文件目录处理
if ".nomedia" in files:
log(
f"Skipping due to .nomedia: {current_root}", level="DEBUG", indent=2
)
dirs[:] = []
continue
# 跳过排除目录
dirs[:] = [d for d in dirs if d not in EXCLUDED_SUBDIRS]
for name in files:
# 忽略隐藏文件
if name.startswith("."):
continue
file_path = current_root / name
try:
mime_type = magic.from_file(str(file_path), mime=True)
if mime_type.startswith(MEDIA_MIME_PREFIXES):
# 计算相对路径以保持目录结构
rel_path = file_path.relative_to(scan_dir)
dest_path = TARGET_DIR / dir_name / rel_path
# 将操作添加到列表
move_operations.append((file_path, dest_path))
log(
f"Found media ({mime_type}): {file_path}",
level="DEBUG",
indent=2,
)
except (IOError, OSError) as e:
log(f"Could not access {file_path}: {e}", level="ERROR", indent=2)
except Exception as e:
if "0-byte file" in str(e):
pass
else:
log(
f"Unexpected error with {file_path}: {e}",
level="ERROR",
indent=2,
)
log(
f"Scan phase finished. Found {len(move_operations)} files to move.",
level="INFO",
)
return move_operations
def execute_move_operations(operations: list):
"""
执行文件移动操作
"""
total_ops = len(operations)
if total_ops == 0:
log("No files to move.", level="INFO")
return
log(f"Execution phase started. Moving {total_ops} files...", level="INFO")
success_count = 0
failure_count = 0
for i, (source_path, dest_path) in enumerate(operations, 1):
log(f"Processing [{i}/{total_ops}]: {source_path}", indent=1)
try:
# 确保目标目录存在
dest_path.parent.mkdir(parents=True, exist_ok=True)
# 移动文件
shutil.move(str(source_path), str(dest_path))
log(f"Moved to: {dest_path}", level="DEBUG", indent=2)
# 通知 Android MediaStore 扫描新文件,以便图库更新
# 需要安装 Termux:API 插件
log("Triggering media scan for the new file...", level="DEBUG", indent=2)
# 使用 subprocess.run 调用 termux-media-scan
result = subprocess.run(
["termux-media-scan", str(dest_path)],
capture_output=True,
text=True,
check=False,
)
if result.returncode == 0:
log("Media scan successful.", level="DEBUG", indent=2)
else:
# 记录扫描失败的错误,但继续执行
log(
f"Media scan failed: {result.stderr.strip()}",
level="WARN",
indent=2,
)
# 如无需即时更新图库,可以注释掉 termux-media-scan 相关代码
success_count += 1
except Exception as e:
log(f"Failed to move file: {e}", level="ERROR", indent=2)
failure_count += 1
log(
f"Execution phase finished. Moved: {success_count}, Failed: {failure_count}.",
level="INFO",
)
def main():
log("=========================================")
log("Scan and Move script started.")
try:
operations_to_perform = find_media_to_move()
execute_move_operations(operations_to_perform)
except Exception as e:
log(f"A critical error occurred in main execution: {e}", level="ERROR")
finally:
log("Script finished.")
log("=========================================\n")
if __name__ == "__main__":
main()
```
这个脚本可以直接复制保存为 scan_and_move_media.py
来使用。
设置自动化执行
最后就是脚本的自动化执行了。安装完 Termux 后,直接打开 APP 就可以启动终端进行配置。
Termux 基本配置
# 可选,但建议更换为国内的镜像源
# termux-change-repo
# 设置安卓存储目录在 Termux 环境中的软链接,需要给 Termux 进行授权
# 成功后可以通过 ls ~/storage/shared 查看手机存储目录
termux-setup-storage
# 安装必要的软件包
# file 提供 libmagic,这是 python-magic 必须的。cronie 提供定时任务。
pkg update && pkg upgrade
pkg install python file cronie
# 如果需要使用远程连接
# pkg install sshd
# sshd
创建 Python 环境
python -m venv ~/.venv
source ~/.venv/bin/activate
pip install python-magic pathlib
设置定时任务
crontab -e
编辑定时任务后运行 crond
启动定时任务即可。我的定时任务如下:
~ $ crontab -l
0 * * * * /data/data/com.termux/files/usr/bin/bash /data/data/com.termux/files/home/scripts/scan_and_move_media.sh
* * * * * echo "$(date) - $(whoami)" >> ~/temp.logs
第一行为实际执行的脚本,配置为 1 小时一次。第二行是用来监控 crond 是否在正常执行的。因为使用了虚拟环境的原因,我没有直接执行 Python 脚本,而是通过一个 Shell 脚本来执行的。并且为了统一管理,这个 Shell 脚本和第二步的 Python 脚本我都统一放在 `$HOME/scripts` 路径下。
#!/data/data/com.termux/files/usr/bin/bash
# 设置 trap:无论脚本是正常退出 (EXIT)、被中断 (INT) 还是被终止 (TERM) 都会释放唤醒锁
trap termux-wake-unlock EXIT INT TERM
# 获取 Termux 唤醒锁
termux-wake-lock
# 激活虚拟环境并执行 Python 脚本
source "$HOME/.venv/bin/activate"
python "$HOME/scripts/scan_and_move_media.py"
# 脚本正常退出,执行释放唤醒锁的命令
exit 0
最后别忘了给脚本可执行权限:
# 直接给 scripts 目录可执行权限
# ~ $ chmod +x -R scripts/
# 或者单独设置两个脚本的可执行权限
~ $ chmod +x scripts/scan_and_move_media.py
~ $ chmod +x scripts/scan_and_move_media.sh
完成配置后,让 Termux 在后台保持运行即可。一段时间后,可以通过检查日志文件,确定定时任务的执行情况。
优化
- 开机启动:可以安装 Termux:Boot 应用,配置
crond
命令在安卓设备开机时自动启动。 - 后台保活:让 Termux 获取唤醒锁或者将 Termux 加入电池优化白名单,避免定时任务执行失败。
- 媒体库刷新:需要执行
pkg install termux-api
,并在手机上安装 Termux:API应用。然后配合 Python 脚本中执行的 termux-media-scan 命令即可。
> 关注 少数派小红书,感受精彩数字生活 🍃
> 实用、好用的 正版软件,少数派为你呈现 🚀