基于Python文件夹监控的轻量级服务自动化部署脚本实践

前几天有件挺有意思的事。周五晚上快九点，我正准备收工，组里同事在群里问：“东哥，你刚改的那个服务我要发测试，怎么包的版本还是上个月的？”

一问才知道，他每次发版还在手动操作：打包 → 用 scp 传文件 → 登录测试机 → 停止旧服务 → 解压新包 → 启动服务。中途要是来个电话或者被叫去开会，回来就忘了进行到哪一步，难怪经常会发错版本。

我当时就想，这都什么年代了，这种重复劳动还不能让机器自己完成吗？于是决定写个 Python 小脚本，让它盯着构建产物的文件夹，一有变化就自动执行部署流程。这样谁改完代码把包往那一放，服务就自己更新了，多省事。

整个思路很简单，就三步：监控目录变化 → 判定是否为有效部署触发点 → 执行部署脚本。

用Python实现文件夹变化监控

首先解决“盯着文件夹”的问题。我们肯定不想用 while True 加 sleep 轮询目录，那种方式既耗CPU又不实时。

Python 里有个很好用的库叫 watchdog，它跨平台，能通过事件回调的方式精准通知文件变化。

先安装一下（建议在你的虚拟环境里操作）：

pip install watchdog

写个最简单的示例，先把监控到的事件打印出来感受一下：

# file: simple_watch.py
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class PrintEventHandler(FileSystemEventHandler):
    def on_any_event(self, event):
        # event.event_type: modified / created / deleted / moved
        # event.src_path: 发生变化的文件路径
        print(f"[EVENT] type={event.event_type}, path={event.src_path}")

if __name__ == "__main__":
    watch_dir = "/opt/app/build"  # 你要监控的目录，根据实际情况修改

    event_handler = PrintEventHandler()
    observer = Observer()
    observer.schedule(event_handler, watch_dir, recursive=True)
    observer.start()

    print(f"开始盯着 {watch_dir} 了... 按 Ctrl+C 退出")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

运行后，在 watch_dir 目录里新建或修改文件，看看终端是否会实时输出日志。这一步通了，我们就能在事件回调里加入自己的业务逻辑了。

事件过滤：并非所有变化都要触发部署

想想实际场景：前端构建会生成一堆 .js、.map、.css 文件，后端构建可能产出 .tar.gz 包或新的 Docker 镜像标签。我们通常只关心：某个特定的关键文件（构建产物）是否已准备就绪。

例如，我们可以约定：后端构建完成后，将打包好的文件放到 /opt/app/deploy/app.tar.gz。那么监控脚本就只关注这个文件的变化即可。

基于这个思路，我们来改造事件处理器：

# file: deploy_watcher.py
import os
import time
import subprocess
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

TARGET_FILE = "/opt/app/deploy/app.tar.gz"

class DeployEventHandler(FileSystemEventHandler):
    def __init__(self):
        super().__init__()
        self._last_deploy_ts = 0

    def _should_trigger(self, event_path: str) -> bool:
        # 1. 只关心目标文件
        if os.path.abspath(event_path) != os.path.abspath(TARGET_FILE):
            return False

        # 2. 简单防抖，10秒内只触发一次部署
        now = time.time()
        if now - self._last_deploy_ts < 10:
            print(“10 秒内已经部署过一次了，稍等片刻...”)
            return False

        self._last_deploy_ts = now
        return True

    def on_modified(self, event):
        if event.is_directory:
            return
        if not self._should_trigger(event.src_path):
            return
        print(f”[WATCH] 检测到构建产物变化：{event.src_path}“)
        self.deploy()

    def on_created(self, event):
        # 有些构建工具会先写临时文件，再重命名为目标文件，这种情况会触发 created 事件
        if event.is_directory:
            return
        if not self._should_trigger(event.src_path):
            return
        print(f”[WATCH] 检测到构建产物创建：{event.src_path}“)
        self.deploy()

    def deploy(self):
        print(”[DEPLOY] 开始执行部署脚本...“)
        try:
            # 调用你实际的部署脚本，例如 /opt/app/deploy/deploy.sh
            result = subprocess.run(
                [“/bin/bash”, “/opt/app/deploy/deploy.sh”],
                check=True,
                capture_output=True,
                text=True,
            )
            print(”[DEPLOY] 部署成功，输出如下：“)
            print(result.stdout)
        except subprocess.CalledProcessError as e:
            print(”[DEPLOY] 部署失败，错误信息：“)
            print(e.stderr)

主函数部分：

if __name__ == ”__main__“:
    watch_dir = os.path.dirname(TARGET_FILE)

    event_handler = DeployEventHandler()
    observer = Observer()
    observer.schedule(event_handler, watch_dir, recursive=False)
    observer.start()

    print(f”开始监控 {TARGET_FILE}，构建产物放过来就会自动部署...“)
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

这样一来，你只需要确保 CI 流程或本地打包脚本的最后一步，是将构建好的 app.tar.gz 放到指定路径，监控脚本就会自动触发部署。

部署脚本里应该做什么？

前面的 Python 脚本只是一个“触发器”，真正的部署动作最好封装在一个独立的 deploy.sh 脚本中，这样更清晰，也便于单独测试。

下面是一个典型的非 Docker 后端服务部署脚本示例：

#!/bin/bash
# file: /opt/app/deploy/deploy.sh
set -e  # 任何一步执行失败就退出脚本

APP_NAME=“demo-service”
DEPLOY_DIR=“/opt/app/runtime”
PACKAGE=“/opt/app/deploy/app.tar.gz”

echo ”[deploy.sh] 开始部署 ${APP_NAME} ...“

# 1. 停止旧服务（生产环境可换成 systemctl 或 supervisorctl）
if pgrep -f “${APP_NAME}” > /dev/null; then
    echo “[deploy.sh] 停止旧进程...”
    pkill -f “${APP_NAME}”
    # 等待进程完全退出
    sleep 2
fi

# 2. 清理旧部署文件
echo “[deploy.sh] 清理旧文件...”
mkdir -p “${DEPLOY_DIR}”
rm -rf “${DEPLOY_DIR:?}/”*

# 3. 解压新版本包
echo “[deploy.sh] 解压新包...”
tar -xzf “${PACKAGE}” -C “${DEPLOY_DIR}”

# 4. 安装依赖（以Python项目为例）
if [ -f “${DEPLOY_DIR}/requirements.txt” ]; then
    echo “[deploy.sh] 安装Python依赖...”
    pip install -r “${DEPLOY_DIR}/requirements.txt” --no-cache-dir
fi

# 5. 启动新服务（示例使用 nohup 后台启动）
echo “[deploy.sh] 启动新进程...”
cd “${DEPLOY_DIR}”
nohup python main.py > app.log 2>&1 &

echo “[deploy.sh] 部署完成。”

如果你使用 Docker 或 Kubernetes，部署脚本的逻辑就变成了构建镜像、执行 docker compose up -d 或 kubectl rollout restart 等命令。核心思路不变，只是调用的命令不同。

如何监控多个服务？

实际项目通常不止一个服务。一个脚本只监控一个文件显得不够用。我们可以引入配置文件，来管理“监控目标”与“部署脚本”的映射关系。

例如，创建一个 config.yaml：

# file: /opt/app/deploy/config.yaml
items:
  - name: “user-service”
    target: “/opt/app/deploy/user/app.tar.gz”
    script: “/opt/app/deploy/user/deploy.sh”

  - name: “order-service”
    target: “/opt/app/deploy/order/app.tar.gz”
    script: “/opt/app/deploy/order/deploy.sh”

然后，改造 Python 脚本以支持多规则配置：

# file: multi_deploy_watcher.py
import os
import time
import subprocess
from dataclasses import dataclass
from typing import Dict

import yaml
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

@dataclass
class DeployItem:
    name: str
    target: str
    script: str
    last_ts: float = 0.0

class MultiDeployHandler(FileSystemEventHandler):
    def __init__(self, items: Dict[str, DeployItem]):
        super().__init__()
        # key: 目标文件的绝对路径
        self.items = items

    def _trigger(self, path: str):
        abspath = os.path.abspath(path)
        item = self.items.get(abspath)
        if not item:
            return

        now = time.time()
        # 防抖：10秒内不重复部署同一服务
        if now - item.last_ts < 10:
            print(f”[{item.name}] 10 秒内已部署过，忽略此次事件“)
            return
        item.last_ts = now

        print(f”[{item.name}] 检测到 {abspath} 变化，开始部署...“)
        try:
            result = subprocess.run(
                [“/bin/bash”, item.script],
                check=True,
                capture_output=True,
                text=True,
            )
            print(f”[{item.name}] 部署成功：“)
            print(result.stdout)
        except subprocess.CalledProcessError as e:
            print(f”[{item.name}] 部署失败：“)
            print(e.stderr)

    def on_modified(self, event):
        if event.is_directory:
            return
        self._trigger(event.src_path)

    def on_created(self, event):
        if event.is_directory:
            return
        self._trigger(event.src_path)

def load_config(config_path: str) -> Dict[str, DeployItem]:
    with open(config_path, “r”, encoding=“utf-8”) as f:
        data = yaml.safe_load(f)

    items: Dict[str, DeployItem] = {}
    for raw in data.get(“items”, []):
        target = os.path.abspath(raw[“target”])
        items[target] = DeployItem(
            name=raw[“name”],
            target=target,
            script=os.path.abspath(raw[“script”]),
        )
    return items

if __name__ == ”__main__“:
    CONFIG = “/opt/app/deploy/config.yaml”
    items = load_config(CONFIG)
    if not items:
        print(”配置为空，无法启动监控。“)
        exit(1)

    # 收集所有需要监控的目录（去重）
    watch_dirs = set(os.path.dirname(it.target) for it in items.values())

    handler = MultiDeployHandler(items)
    observer = Observer()
    for d in watch_dirs:
        observer.schedule(handler, d, recursive=False)
        print(f”正在监控目录：{d}“)

    observer.start()
    print(”所有监控已启动，构建完丢包即可自动触发部署。“)
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

现在，只要你的构建流程将 user-service 的包输出到 /opt/app/deploy/user/app.tar.gz，监控脚本就会自动调用对应的 user/deploy.sh 完成部署，实现真正的 运维 自动化。

需要注意的几个关键点

这个方案看似简单，但如果想用在准生产环境，有几个坑需要注意，否则可能会在半夜被报警叫起来“救火”。

1. 构建和文件拷贝是原子的吗？
   有些打包工具是先写临时文件，再通过重命名（rename）操作移到目标位置，这种方式是原子的，没问题。但有些工具会直接往目标文件流式写入，这可能导致脚本在文件还没写完（“半包”状态）时就触发部署。一个简单的应对策略是：在部署脚本开始时增加文件完整性校验，例如检查压缩包是否能正常解压，或文件大小是否达到预期阈值，校验不通过则直接失败退出。

2. 部署过程耗时较长怎么办？
   如果一次部署需要1分钟，而在这1分钟内又来了新的构建包，上面的代码仅做了防抖，并没有处理任务排队。一个简单的改进方法是：部署开始时在 /tmp 目录创建一个锁文件（lock file），部署完成后删除。在 _should_trigger 或 _trigger 方法中检查锁文件是否存在，如果存在则跳过本次触发。如果需要更精细的控制，可以实现一个任务队列，由单独的 worker 进程顺序处理。

3. 权限问题
   本地测试时一切顺利，放到服务器上可能就会出现各种 Permission denied。这个监控脚本本质上是“自动帮你执行 shell 命令”，它的权限完全取决于运行它的用户。建议创建一个专用用户（如 deploy），并提前配置好该用户对监控目录、部署脚本以及目标运行目录的读写执行权限。

4. 认清定位：这只是轻量级辅助工具
   这个方案更适合作为测试环境、个人项目或内网工具的“轻量级热部署助手”。对于正式的生产环境，还是应该搭建具备审批流程、回滚机制、灰度发布等能力的完整 CI/CD 流水线。不过，在合适的场景下，这个脚本用起来是真的香。

如何以守护进程方式可靠运行？

通常，我会在服务器上这样部署它：

1. 创建一个专用系统用户，例如 deploy。
2. 将 multi_deploy_watcher.py 和 config.yaml 等文件放到 /opt/app/deploy 目录下。
3. 编写一个 systemd 服务单元文件，将这个监控脚本作为守护进程长期运行。

一个示例的 systemd 服务文件 (/etc/systemd/system/deploy-watcher.service) 如下：

[Unit]
Description=Simple Python Deploy Watcher
After=network.target

[Service]
User=deploy
WorkingDirectory=/opt/app/deploy
ExecStart=/usr/bin/python /opt/app/deploy/multi_deploy_watcher.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

然后执行以下命令启用并启动服务：

systemctl daemon-reload
systemctl enable deploy-watcher
systemctl start deploy-watcher
systemctl status deploy-watcher  # 查看运行状态

这样一来，监控脚本就在后台稳定运行了。你或你的 CI 系统只需要将构建好的包放到约定目录，它就会默默地把服务更新上线。作为开发者，你只需要查看一下日志确认部署结果即可。

通过这样一个简单的自动化脚本，就能把同事从繁琐的手动操作中解放出来，省下来的时间干点啥不好呢？如果你对这类提升效率的自动化脚本感兴趣，欢迎来 云栈社区 交流更多实战技巧。