如何用 Python 自动化监控文件夹完成服务部署！

那天是周五晚上十一点多，我刚把外卖盒子往垃圾桶一扔，手机叮的一声，运维小哥在群里喊：“谁在改配置？线上服务怎么又没起来了？！”

一问才知道，又是有人半夜发版，远程桌面上去，拷了一堆文件到服务器那个 release 文件夹里，拷完手动点了一堆脚本，还顺手删了个旧文件，结果少删了一个、多覆盖了一个，服务直接起不来了，人还已经打车回家路上了，整个人都不好。

我当时就想：这玩意儿完全可以让 Python 帮我们盯着文件夹，有包就自动部署嘛，干嘛每次人肉点来点去。

先说人话版需求：到底要干啥

你可以脑补这么个画面：

• CI（比如 Jenkins、GitLab CI）打完包，扔到某个共享目录，比如 /data/deploy/packages

• 服务器上有个 Python 小脚本，一直悄悄在那儿看着这个目录

• 一旦发现有个新的 xxx-2024-01-01.tar.gz 丢进来，就自动干这几件事：

• 校验一下包是不是传完了、没残缺
• 做个当前版本的备份（比如打个 backup_时间戳 的目录）
• 停服务
• 解压新包，更新软链接或者覆盖目录
• 起服务，跑个健康检查
• 有问题就回滚到刚才那个备份

人只负责看钉钉 / 微信里那句“部署成功 ✅”，爽的就是这么简单。

这个需求翻译成程序语言就是：监控文件夹事件 + 部署流程编排。

怎么让 Python 盯着一个文件夹？

偷懒的办法有两个：

1. 最原始：每隔 N 秒 os.listdir() 扫一遍，对比有没有新文件（像人隔几分钟瞄一眼）
2. 稍微专业点：用系统提供的文件事件，谁动了文件系统，系统主动通知你

Python 里已经有人帮我们把第二种封装好了，就是 watchdog 这个库，装一下就行：

pip install watchdog

来个最小可用 demo，先把“看到新包就打个招呼”跑起来：

import time
from pathlib import Path

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


WATCH_DIR = "/data/deploy/packages"
PACKAGE_SUFFIX = (".tar.gz", ".zip")  # 你们爱打啥包就自己改


classPackageHandler(FileSystemEventHandler):
defon_created(self, event):
# event.src_path 有可能是文件也可能是目录
        path = Path(event.src_path)
if path.is_file() and path.suffix in PACKAGE_SUFFIX:
            print(f"[watcher] 新包来了：{path.name}")
# 这里后面就可以挂真正的部署逻辑
# deploy_service(path)


defstart_watch():
    event_handler = PackageHandler()
    observer = Observer()
    observer.schedule(event_handler, WATCH_DIR, recursive=False)
    observer.start()
    print(f"[watcher] 开始盯着 {WATCH_DIR} 了...")

try:
whileTrue:
            time.sleep(1)
except KeyboardInterrupt:
        observer.stop()
    observer.join()


if __name__ == "__main__":
    start_watch()

这个东西跑起来之后，你往 WATCH_DIR 里丢一个 xxx.tar.gz，终端就会输出一句“新包来了”。说明基本链路通了。

包刚传到一半就去部署？这是要出人命的

现实情况里，CI 上传文件不是原子操作，可能是：

• 先在本地生成 xxx.tar.gz
• 再通过 SCP / SMB 一点点传上去
• 传一半的时候，文件已经出现在目录里了，但大小还在涨

如果我们看到文件一出现就部署，很容易拿着一个半成品的包去解压，直接爆炸。

这个地方我一般会搞两层保险：

1. 约定 CI 侧先上传为临时后缀，比如 .tmp，传完再改名
2. 保险起见再判断文件大小在一段时间内不再变化

第二种很容易写个小函数：

import time
import os
from pathlib import Path


defwait_file_stable(path: Path, check_interval=1, stable_seconds=5, timeout=300):
"""等文件大小稳定 stable_seconds 秒，超时就算了"""
    start = time.time()
    last_size = -1
    last_change_time = time.time()

whileTrue:
ifnot path.exists():
raise FileNotFoundError(f"{path} 不存在了，可能被删了")

        size = os.path.getsize(path)

if size != last_size:
            last_size = size
            last_change_time = time.time()

if time.time() - last_change_time >= stable_seconds:
# 大小至少 stable_seconds 秒没变了
return

if time.time() - start > timeout:
raise TimeoutError(f"等文件稳定超时: {path}")

        time.sleep(check_interval)

然后在 on_created 里加一句等待：

classPackageHandler(FileSystemEventHandler):
defon_created(self, event):
        path = Path(event.src_path)
if path.is_file() and path.suffix in PACKAGE_SUFFIX:
            print(f"[watcher] 发现新包：{path.name}，先等它传完...")
try:
                wait_file_stable(path)
except Exception as e:
                print(f"[watcher] 等包稳定失败：{e}")
return

            print(f"[watcher] 包看起来稳定了，准备部署：{path.name}")
            deploy_service(path)

部署到底干啥：别写成一坨脚本地狱

我比较习惯的目录结构，大概这样：

/app/my_service/
  ├── current -> /app/my_service/releases/2024-01-01_120000
  ├── releases/
  │     ├── 2023-12-01_100000/
  │     ├── 2023-12-15_093000/
  │     └── 2024-01-01_120000/
  └── logs/

每次发版就是：

• 把新包解压到一个新目录：releases/时间戳
• 把 current 这个软链接指向新的目录
• 用 systemctl 重启服务

这样回滚特别爽：软链接往回一指就行。

用 Python 写一个简化版的部署流程：

import shutil
import subprocess
import tarfile
import time
from pathlib import Path

APP_ROOT = Path("/app/my_service")
RELEASES_DIR = APP_ROOT / "releases"
CURRENT_LINK = APP_ROOT / "current"
SERVICE_NAME = "my_service"# systemd 里的 service 名


defrun_cmd(cmd: list[str]):
    print(f"[deploy] 执行命令：{' '.join(cmd)}")
    result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
        print(result.stdout)
        print(result.stderr)
raise RuntimeError(f"命令失败：{' '.join(cmd)}")
return result.stdout


defstop_service():
# 这里可以根据你们自己的方式来，systemd、docker 都行
    run_cmd(["systemctl", "stop", SERVICE_NAME])


defstart_service():
    run_cmd(["systemctl", "start", SERVICE_NAME])


defhealth_check(url="http://127.0.0.1:8000/health", timeout=30):
import requests

    deadline = time.time() + timeout
while time.time() < deadline:
try:
            r = requests.get(url, timeout=2)
if r.status_code == 200:
                print("[deploy] 健康检查通过")
returnTrue
except Exception:
pass
        time.sleep(1)
raise RuntimeError("健康检查超时")


defextract_package(package_path: Path, target_dir: Path):
if target_dir.exists():
raise FileExistsError(f"目标目录已存在：{target_dir}")
    target_dir.mkdir(parents=True)

if package_path.suffixes[-2:] == [".tar", ".gz"] or package_path.suffix == ".tgz":
with tarfile.open(package_path, "r:gz") as tar:
            tar.extractall(target_dir)
elif package_path.suffix == ".zip":
        shutil.unpack_archive(str(package_path), str(target_dir))
else:
raise ValueError(f"不支持的包格式：{package_path}")

    print(f"[deploy] 包已解压到 {target_dir}")

重点的 deploy_service：

defdeploy_service(package_path: Path):
    ts = time.strftime("%Y%m%d_%H%M%S")
    new_release_dir = RELEASES_DIR / ts

# 记录当前指向，方便回滚
    old_target = None
if CURRENT_LINK.is_symlink():
try:
            old_target = CURRENT_LINK.resolve()
except Exception:
            old_target = None

try:
        print(f"[deploy] 开始部署 {package_path.name}")
        extract_package(package_path, new_release_dir)

        print("[deploy] 停服务...")
        stop_service()

# 切软链
if CURRENT_LINK.exists() or CURRENT_LINK.is_symlink():
            CURRENT_LINK.unlink()
        CURRENT_LINK.symlink_to(new_release_dir)

        print("[deploy] 起服务...")
        start_service()

# 健康检查
        health_check()

        print(f"[deploy] 部署成功，新版本目录：{new_release_dir}")
except Exception as e:
        print(f"[deploy] 部署失败：{e}")
# 尝试回滚
if old_target and old_target.exists():
try:
                print("[deploy] 尝试回滚到旧版本...")
if CURRENT_LINK.exists() or CURRENT_LINK.is_symlink():
                    CURRENT_LINK.unlink()
                CURRENT_LINK.symlink_to(old_target)
                start_service()
                health_check()
                print("[deploy] 回滚成功")
except Exception as e2:
                print(f"[deploy] 回滚失败：{e2}")
else:
            print("[deploy] 没找到旧版本可回滚，小心处理")
raise

整一套串起来，基本就是：文件夹有新包 → 等它传完 → 停服务 → 切版本 → 起服务 → 健康检查 → 回滚兜底。

一些现场踩过的坑，顺手说一下

这个东西我在别的项目上搞过几次，除了业务不同，坑都差不多。之前写数据库压测脚本的时候，也搞过一套自动部署，自动刷版本，不然一遍遍手动太折磨人。

有几个点你要是现在注意一下，后面能少掉很多“为什么线上又挂了”的灵魂拷问：

• 权限问题： 你这个 watcher 脚本要有权限停服务、改软链、访问目录。一般会：

• 专门搞一个系统用户，比如 deploy
• 让它能 sudo systemctl restart xxx，但限制好 sudo 规则
• 目录权限也配清楚

• 不要在部署脚本里写死各种绝对路径散得到处都是全放在一个配置里，比如 config.yaml，路径、服务名、健康检查 URL 都写里面，Python 读一下，就完事。

• 日志一定要有： 别全靠 print，随便用个 logging 写到固定日志文件里，出了问题，运维至少能找到“到底跑到哪一步挂了”。

  简单一点这样就够用：

  import logging
  
  logging.basicConfig(
      filename="/var/log/deploy_watcher.log",
      level=logging.INFO,
      format="%(asctime)s [%(levelname)s] %(message)s",
  )
  
  logger = logging.getLogger(__name__)
  

  然后把上面的 print 换成 logger.info / logger.error，世界清静很多。

• 多环境要分开： 测试、预发、生产最好是三个完全不一样的目录和服务名，甚至三台不同机器，不然手一抖，把测试包扔到生产目录里，那真的是连续加班套餐。

怎么跟 CI 串起来，才算一套完整方案

上面这些其实只是“服务器本地的自动部署脚本”。真正省心的是把它跟 CI 整起来，比如：

• 开发 push 到 main
• CI 打包 & 跑测试
• 测过了就把成品包 scp / rsync 到部署机器的 packages 目录
• CI 也不用管怎么重启服务、怎么备份，它只负责“扔包”
• 剩下的都是 Python watcher 在服务器上搞定

好处很明显：

• CI 脚本更简单，跨项目都能复用
• 你的部署逻辑完全掌握在自己这边一小段 Python 里，想加校验、加钉钉通知都好做
• 出了故障直接看 watcher 日志，大概就知道是“包没传完”“健康检查过不了”还是“停服务失败”这种问题

整合一下完整最小版本（能跑的那种）

如果你想先搞个最小可用的玩玩，可以把上面这些拼成一个 deploy_watcher.py，稍微收拾一下：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import logging
import os
import time
from pathlib import Path

import requests
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import tarfile
import shutil
import subprocess


WATCH_DIR = "/data/deploy/packages"
PACKAGE_SUFFIX = (".tar.gz", ".tgz", ".zip")

APP_ROOT = Path("/app/my_service")
RELEASES_DIR = APP_ROOT / "releases"
CURRENT_LINK = APP_ROOT / "current"
SERVICE_NAME = "my_service"
HEALTH_URL = "http://127.0.0.1:8000/health"


logging.basicConfig(
    filename="/var/log/deploy_watcher.log",
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger(__name__)


defrun_cmd(cmd: list[str]):
    logger.info("执行命令：%s", " ".join(cmd))
    result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
        logger.error("命令失败 stdout=%s stderr=%s", result.stdout, result.stderr)
raise RuntimeError(f"命令失败：{' '.join(cmd)}")
return result.stdout


defwait_file_stable(path: Path, check_interval=1, stable_seconds=5, timeout=300):
    start = time.time()
    last_size = -1
    last_change_time = time.time()

whileTrue:
ifnot path.exists():
raise FileNotFoundError(f"{path} 不存在了")

        size = os.path.getsize(path)
if size != last_size:
            last_size = size
            last_change_time = time.time()

if time.time() - last_change_time >= stable_seconds:
            logger.info("文件大小稳定：%s size=%d", path, size)
return

if time.time() - start > timeout:
raise TimeoutError(f"等文件稳定超时: {path}")

        time.sleep(check_interval)


defextract_package(package_path: Path, target_dir: Path):
if target_dir.exists():
raise FileExistsError(f"目标目录已存在：{target_dir}")
    target_dir.mkdir(parents=True, exist_ok=False)

if package_path.suffixes[-2:] == [".tar", ".gz"] or package_path.suffix == ".tgz":
with tarfile.open(package_path, "r:gz") as tar:
            tar.extractall(target_dir)
elif package_path.suffix == ".zip":
        shutil.unpack_archive(str(package_path), str(target_dir))
else:
raise ValueError(f"不支持的包格式：{package_path}")

    logger.info("包已解压到 %s", target_dir)


defstop_service():
    run_cmd(["systemctl", "stop", SERVICE_NAME])


defstart_service():
    run_cmd(["systemctl", "start", SERVICE_NAME])


defhealth_check(timeout=30):
    deadline = time.time() + timeout
while time.time() < deadline:
try:
            r = requests.get(HEALTH_URL, timeout=2)
if r.status_code == 200:
                logger.info("健康检查通过")
return
except Exception:
pass
        time.sleep(1)
raise RuntimeError("健康检查超时")


defdeploy_service(package_path: Path):
    ts = time.strftime("%Y%m%d_%H%M%S")
    new_release_dir = RELEASES_DIR / ts

    old_target = None
if CURRENT_LINK.is_symlink():
try:
            old_target = CURRENT_LINK.resolve()
except Exception:
            old_target = None

try:
        logger.info("开始部署 %s", package_path)
        extract_package(package_path, new_release_dir)

        logger.info("停服务")
        stop_service()

if CURRENT_LINK.exists() or CURRENT_LINK.is_symlink():
            CURRENT_LINK.unlink()
        CURRENT_LINK.symlink_to(new_release_dir)

        logger.info("起服务")
        start_service()

        health_check()
        logger.info("部署成功，新版本目录：%s", new_release_dir)
except Exception as e:
        logger.error("部署失败：%s", e)
if old_target and old_target.exists():
try:
                logger.info("尝试回滚到旧版本：%s", old_target)
if CURRENT_LINK.exists() or CURRENT_LINK.is_symlink():
                    CURRENT_LINK.unlink()
                CURRENT_LINK.symlink_to(old_target)
                start_service()
                health_check()
                logger.info("回滚成功")
except Exception as e2:
                logger.error("回滚失败：%s", e2)
else:
            logger.error("没有旧版本可回滚，人工介入吧")
raise


classPackageHandler(FileSystemEventHandler):
defon_created(self, event):
        path = Path(event.src_path)
ifnot path.is_file():
return

ifnot any(str(path).endswith(s) for s in PACKAGE_SUFFIX):
return

        logger.info("发现新包：%s", path)
try:
            wait_file_stable(path)
            deploy_service(path)
except Exception as e:
            logger.error("处理包失败 %s: %s", path, e)


defstart_watch():
    observer = Observer()
    handler = PackageHandler()
    observer.schedule(handler, WATCH_DIR, recursive=False)
    observer.start()
    logger.info("开始监控目录：%s", WATCH_DIR)

try:
whileTrue:
            time.sleep(1)
except KeyboardInterrupt:
        observer.stop()
    observer.join()


if __name__ == "__main__":
    start_watch()

这个甩到服务器上，配个 systemd 服务让它自启，你们再把 CI 的产物路径对上，基本就能跑起来了。

行，先到这儿吧，我这会儿得去看看我们组小李是不是又在手动拷贝包了，自己写了自动化工具不用的那种人，真的让人头秃…