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 命令即可。

> 关注 少数派小红书,感受精彩数字生活 🍃

> 实用、好用的 正版软件,少数派为你呈现 🚀