Python自动化脚本管理工具实现

每天都有大量的脚本需要运行，并且还需要记住每个脚本应该何时执行，真是令人头疼！干脆开发一个工具来管理这些脚本吧，让它自动执行、监控状态并查看运行日志，岂不快哉。

项目架构设计

这个工具主要分为三个部分：脚本注册管理、任务调度器和日志系统。简单来说，就是告诉工具要执行什么脚本、什么时候执行以及执行完毕后的状态。

        
            class ScriptManager:
                def __init__(self):
                    self.scripts = {}
                    self.scheduler = {}
                    self.log_path = "script_logs/"

                def register_script(self, name, path, schedule=""):
                    self.scripts[name] = {
                        "path": path,
                        "schedule": schedule,
                        "status": "idle"
                    }
        
    

脚本注册功能

当然，工具需要知道要管理哪些脚本。因此，我们需要编写一个注册函数，用来记录脚本的名字、路径和执行计划。

        
            def add_script(self, name, script_path):
                if not os.path.exists(script_path):
                    raise FileNotFoundError("脚本文件不存在！")
                
                self.scripts[name] = {
                    "path": script_path,
                    "last_run": None,
                    "status": "ready"
                }
        
    

???? 温馨提示：注册脚本时最好检查文件是否存在，以免运行时找不到文件。

任务调度模块

仅仅注册脚本是不够的，必须按照计划执行。这里我们可以使用schedule库来创建定时任务，既简单又实用。

        
            import schedule
            import time
            from datetime import datetime

            def setup_scheduler(self):
                for name, info in self.scripts.items():
                    if info["schedule"]:
                        schedule.every().day.at(info["schedule"]).do(
                            self.run_script,
                            name
                        )
                
                while True:
                    schedule.run_pending()
                    time.sleep(1)
        
    

脚本执行器

执行脚本时需要使用子进程来运行，以确保主程序不会被阻塞。同时，还需要将输出重定向到日志文件中。

        
            def run_script(self, script_name):
                script_info = self.scripts[script_name]
                log_file = f"{self.log_path}{script_name}_{datetime.now():%Y%m%d_%H%M}.log"

                try:
                    process = subprocess.Popen(
                        ["python", script_info["path"]],
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE
                    )

                    output, error = process.communicate()

                    with open(log_file, "w") as f:
                        f.write(output.decode())
                        if error:
                            f.write("\nErrors:\n" + error.decode())

                    script_info["status"] = "completed" if process.returncode == 0 else "failed"

                except Exception as e:
                    script_info["status"] = "error"
                    with open(log_file, "w") as f:
                        f.write(f"执行出错啦：{str(e)}")
        
    

状态监控

需要了解脚本是否执行成功，如果失败了也应及时发现。

        
            def get_status(self):
                status_info = {}
                for name, info in self.scripts.items():
                    status_info[name] = {
                        "status": info["status"],
                        "last_run": info["last_run"]
                    }
                return status_info
        
    

???? 温馨提示：在监控状态时可以添加一个邮件通知功能，以便在脚本运行失败时能够立即收到提醒。

使用示例

看看这个工具如何使用：

        
            manager = ScriptManager()

            # 注册脚本
            manager.add_script("数据备份", "/scripts/backup.py")
            manager.add_script("日报生成", "/scripts/daily_report.py", "18:00")

            # 启动调度器
            manager.setup_scheduler()

            # 查看状态
            print(manager.get_status())
        
    

至此，代码已经完成。现在再也不需要手动运行脚本，也不会忘记执行时间。它不仅能够自动执行，还能记录日志，一旦出现问题可以迅速定位原因。

未来还可以为这个工具添加更多功能，例如Web界面管理和远程执行脚本，这样会更加便捷。