Python监控脚本二次提交异常求助：Git Add未触发问题排查

需求背景 #

• 网络交换机配置变更后，自动将当前配置发送至FTP服务器。
• 用Watchdog监控目录，收到新的juniper.conf.gz文件时，解压并执行git add/git commit/git push操作。

问题现象 #

首次提交文件时脚本运行正常，配置可在GitLab查看；同一交换机发送第二个文件时，旧文件被成功覆盖，但脚本卡顿，git add操作未触发。

实现代码 #

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

import argparse
import gzip
import os
import re
import shutil
import time
from pathlib import Path

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


class Watcher:
    def __init__(self, directory, handler=FileSystemEventHandler()):
        self.observer = Observer()
        self.handler = handler
        self.directory = directory
        
    def run(self):
        self.observer.schedule(self.handler, self.directory, recursive=False)
        self.observer.start()
        print("\nWatcher Running in {}/\n".format(self.directory))
        try:
            while True:
                time.sleep(1)
        except:
            self.observer.stop()
        self.observer.join()
        print("\nWatcher Terminated\n")

class FsHandler(FileSystemEventHandler):    
    
    """
    Purpose: React on file creation
    Event: created
    Action: log and unzip
    Condition: if file ends with juniper.conf.gz
    """
    def on_created(self, event):
        print("on_created")
        if event.src_path.endswith("juniper.conf.gz"):
            print(self.__cleanup(event.src_path))
        if event.src_path.endswith("juniper.conf"):
            # Start revision handling
            sh.git.add("-A")
            sh.git.commit("-m", str(datetime.now()))
            sh.git.push().wait()
                       
    
    """
    Purpose: React on file moving/renaming
    Event: moved
    Action: log and commit to git
    Condition: -NONE-
    """
    def on_moved(self, event):
        print("on_moved")
        # decompress the gzip file
        with gzip.open(event.dest_path, 'rb') as f_in:
            with open(re.sub('.gz$', '', event.dest_path),'wb') as f_out:
                shutil.copyfileobj(f_in, f_out)
                f_out.close()
        
        # remove the gzip file version
        os.remove(event.dest_path);
        
    """
    Purpose: React on file deletion
    Event: deleted
    Action:  log
    Condition: -NONE-
    """
    def on_deleted(self, event):
        print("on_deleted")
        print("Filename: " + Path(str(event.src_path)).name)
    
    """
    Purpose: Remove date and time from file
    Reason: Otherwise each config file will have its own commit,
             no matter it comes from the same switch.
    """
    def __cleanup(self, old_filename):
        new_filename = re.sub('_[0-9]+_[0-9]+_juniper.conf.gz', '_juniper.conf.gz', old_filename)
        os.rename(old_filename, new_filename)
    
if __name__ == "__main__":
    # Default path to watch is current dir
    watch_path='.'
    
    # Handle arguments from commandline
    parser = argparse.ArgumentParser(
        prog="autocommit.py",
        description="Autocommit Junos configuration from our switches to git repo",
        epilog="(c)2022 Roger Nordqvist <roger.nordqvist@filipstad.se>"
        )
    parser.add_argument(
        "-p", "--path", help="Specify path to watch"
        )
    
    # Parse commandline options
    args = parser.parse_args()
    if args.path:
        watch_path=args.path
        print(watch_path) 

    # move script execution to the path we are watching
    os.chdir(watch_path)

    w = Watcher(watch_path, FsHandler())
    w.run()

问题根源分析 #

1. 文件写入未完成就触发事件：on_created事件在文件刚创建时就触发，但此时交换机可能还在写入内容，直接处理会导致文件不完整，后续操作异常。
2. 事件处理逻辑混乱：解压操作放在on_moved中，重命名触发的on_moved事件与Git操作的触发条件（on_created检测.conf文件）脱节，二次提交时覆盖文件的逻辑导致事件触发顺序异常。
3. Git操作阻塞主线程：sh.git.push().wait()会阻塞事件循环，若网络或Git服务器响应慢，后续事件无法及时处理。
4. 无文件状态校验：二次提交覆盖文件时，若前一次操作未完成，会出现文件占用冲突，导致步骤失败。

解决方案 #

1. 等待文件写入完成 #

处理前先等待文件大小稳定，确保交换机完成传输。

2. 统一处理流程 #

将重命名、解压、Git提交整合到单一逻辑流，避免事件交叉遗漏。

3. 异步执行Git操作 #

用线程处理Git命令，避免阻塞Watchdog的事件循环。

4. 覆盖场景兼容 #

增加on_modified事件处理，捕获二次提交时的文件覆盖行为。

修改后的完整代码 #

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

import argparse
import gzip
import os
import re
import shutil
import time
from pathlib import Path
from threading import Thread

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


class Watcher:
    def __init__(self, directory, handler=FileSystemEventHandler()):
        self.observer = Observer()
        self.handler = handler
        self.directory = directory
        
    def run(self):
        self.observer.schedule(self.handler, self.directory, recursive=False)
        self.observer.start()
        print("\nWatcher Running in {}/\n".format(self.directory))
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            self.observer.stop()
        self.observer.join()
        print("\nWatcher Terminated\n")

class FsHandler(FileSystemEventHandler):    
    def wait_for_file_stable(self, file_path, wait_interval=0.5, max_retries=10):
        """等待文件大小稳定，确认写入完成"""
        prev_size = -1
        retries = 0
        while retries < max_retries:
            if not os.path.exists(file_path):
                time.sleep(wait_interval)
                retries +=1
                continue
            current_size = os.path.getsize(file_path)
            if current_size == prev_size:
                return True
            prev_size = current_size
            time.sleep(wait_interval)
            retries +=1
        return False
    
    def process_config_file(self, gz_file_path):
        """处理压缩文件：重命名、解压、Git提交"""
        # 等待文件写入完成
        if not self.wait_for_file_stable(gz_file_path):
            print(f"文件 {gz_file_path} 写入超时，跳过处理")
            return
        
        # 重命名文件（移除时间戳）
        new_gz_filename = re.sub('_[0-9]+_[0-9]+_juniper.conf.gz', '_juniper.conf.gz', gz_file_path)
        try:
            os.rename(gz_file_path, new_gz_filename)
            print(f"重命名文件至 {new_gz_filename}")
        except OSError as e:
            print(f"重命名文件失败: {e}")
            return
        
        # 解压文件
        conf_file_path = re.sub('.gz$', '', new_gz_filename)
        try:
            with gzip.open(new_gz_filename, 'rb') as f_in:
                with open(conf_file_path, 'wb') as f_out:
                    shutil.copyfileobj(f_in, f_out)
            print(f"解压完成，生成文件 {conf_file_path}")
        except Exception as e:
            print(f"解压文件失败: {e}")
            return
        
        # 删除压缩包
        try:
            os.remove(new_gz_filename)
            print(f"删除压缩包 {new_gz_filename}")
        except OSError as e:
            print(f"删除压缩包失败: {e}")
        
        # 异步执行Git操作，避免阻塞事件循环
        Thread(target=self.perform_git_commit, args=(conf_file_path,)).start()
    
    def perform_git_commit(self, conf_file_path):
        """执行Git提交操作"""
        try:
            sh.git.add(conf_file_path)
            commit_msg = f"Update config {Path(conf_file_path).name} at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
            sh.git.commit("-m", commit_msg)
            sh.git.push()
            print(f"Git提交成功: {commit_msg}")
        except sh.ErrorReturnCode as e:
            print(f"Git操作失败: {e}")
    
    def on_created(self, event):
        if event.is_directory:
            return
        if event.src_path.endswith("juniper.conf.gz"):
            print(f"检测到新文件: {event.src_path}")
            self.process_config_file(event.src_path)
    
    def on_modified(self, event):
        # 处理文件被覆盖的情况（二次提交时可能触发modified事件）
        if event.is_directory:
            return
        if event.src_path.endswith("juniper.conf.gz"):
            print(f"文件被修改: {event.src_path}")
            self.process_config_file(event.src_path)

if __name__ == "__main__":
    watch_path='.'
    
    parser = argparse.ArgumentParser(
        prog="autocommit.py",
        description="Autocommit Junos configuration from our switches to git repo",
        epilog="(c)2022 Roger Nordqvist <roger.nordqvist@filipstad.se>"
        )
    parser.add_argument("-p", "--path", help="Specify path to watch")
    
    args = parser.parse_args()
    if args.path:
        watch_path=args.path
        print(f"监控路径设置为: {watch_path}") 

    os.chdir(watch_path)

    w = Watcher(watch_path, FsHandler())
    w.run()

关键改进点 #

• 文件稳定校验：新增wait_for_file_stable方法，确保文件写入完成后再处理。
• 统一处理逻辑：将重命名、解压、Git提交整合到单一方法，流程更清晰。
• 异步Git操作：用线程执行Git命令，避免阻塞事件循环。
• 覆盖场景兼容：增加on_modified事件处理，确保二次提交被捕获。
• 完善错误处理：每个步骤添加异常捕获，避免单个环节失败导致脚本卡顿。

内容的提问来源于stack exchange，提问作者Roger Nordqvist