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 实现示例代码:
pythonimport osimport shutilimport datetimeimport 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_operationsdef 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 upgradepkg install python file cronie# 如果需要使用远程连接# pkg install sshd# sshd创建 Python 环境
python -m venv ~/.venvsource ~/.venv/bin/activatepip 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 命令即可。