如果您一直在遵循我们的 DIY AI 程序,那么您现在已经设置了文件系统并准备开始编码。如果这是您第一次访问,最好从头开始,因为我们已经介绍了一些内容,包括设置 Visual Studio Code、安装重要的 Python 库以及设置虚拟环境。在本指南中,我们将设置中央调度程序,这将是我们 DIY AI 开发项目中的第一个脚本。如果将虚拟环境视为大脑,将文件系统视为身体,那么中央调度程序就是心脏。
中央调度程序将确保我们编写的脚本能够自动按时运行。许多人工智能工作将涉及收集数据以在特定时间进行研究。例如,要了解计算机上的文件系统、经常使用的文件以及可能对您有害的文件,您可能希望人工智能每天晚上在您睡觉时扫描您的驱动器。中央调度程序将帮助您做到这一点。
程序运行后,您可能会发现有很多文件要运行。该脚本将通过实现一个简单的系统来帮助您保持领先,该系统可让您快速轻松地添加和删除任务。
选择您的存储方式
关于 DIY,您会发现的一件事是,有很多不同的方法可以完成任务,而创建中央调度程序只是此项目中的一个脚本,您可以根据不同的方式完成它。该脚本中主要的担忧是,作为程序的开发人员和用户,您可能需要经常更改脚本运行的内容和时间。因此,代码需要与外部源通信以查看要运行哪些脚本。否则,每次需求发生变化时,您都需要重写脚本。
流行的选项包括使用 JSON 文件或数据库,例如 Microsoft Access 或 SQLite。当您只运行几个文件时,JSON 很好,并且可以快速添加和删除任务,但是当您的程序开始变大时,文件可能会变得很大并且难以管理。我建议使用 Microsoft Access 数据库,我们将选择它,因为我已经在使用它了。然而,还有很多其他数据库选项,如果您更喜欢使用其他数据库,编码将在很大程度上是相同的。每个数据库都将使用您可以像我们在这里所做的那样设置的表,然后使用 Python 进行访问以了解要运行哪些脚本以及何时运行。我们还可以使用同一数据库中的不同表来存储各种数据供人工智能使用。
设置数据库表
您的表将需要六个字段。一项是自动设置条目的 ID,一项是日期和时间,另外四项是短文本。我建议将表保存为“ai_tasks”,或者许多程序使用的“任务”之外的任何内容。
安装额外的库
我们在过去的指南中安装的库充当我们程序的主要框架。尽管如此,当需要时我们可能需要添加额外的库,而这就是这样的场合。我们需要安装“pyodbc”库以及“schedule”库。
“pyodbc”库适用于所有最常见的数据库,包括 Microsoft Office、Oracle、MySQL 等。 “schedule”库以其简单的语法和强大的功能使调度变得轻而易举。
要安装它们,请在虚拟环境中的提示符中键入以下代码:
pip 安装 pyodbc 时间表
安装库后,我们可以开始编写 Python 脚本,该脚本将检查数据库中是否有我们想要以特定时间间隔运行的脚本。让我们仔细检查一下该脚本的每个部分,然后我将向您提供整个内容以供您复制和粘贴。
导入库
即使我们安装了这些库,我们仍然需要导入它们才能使用它们,这就是这部分代码所做的事情,导入时间表、时间、日期时间和 pyodbc 库。
进口时间表
导入时间
从日期时间导入日期时间
导入pyodbc
数据库连接
脚本的下一部分将处理与数据库的连接。您需要将数据库路径更改为保存我们之前创建的表的数据库的位置。
DB_PATH = r"C:\Path\To\TaskScheduler.accdb" # 使用您的文件路径更新
CONN_STR = f"DRIVER={{Microsoft Access 驱动程序 (*.mdb, *.accdb)}};DBQ={DB_PATH};"
从数据库中获取任务
我们需要 load_tasks 函数从我们之前创建的表中获取信息。您需要将 your_table_name 替换为您为表选择的名称。
def load_tasks():
conn = pyodbc.connect(CONN_STR)
光标 = conn.cursor()
cursor.execute("从 Your_Table_Name 中选择 id、脚本、间隔、时间、last_run")
任务 = 游标.fetchall()
conn.close()
返回任务
更新上次运行时间
update_last_run 函数将更新数据库中的表以反映上次运行该特定脚本的时间。您需要使用您的表名称更新这部分代码。
运行任务
run_task 函数将运行脚本,然后调用 update_last_run 函数。如果出现问题,它会打印一条错误消息。
def run_task(脚本,task_id):
尝试:
exec(打开(脚本).read())
update_last_run(任务id)
除了异常 e:
print(f"运行脚本 {script} 时出错:{e}")
设置时间表
setup_schedule 函数负责在特定时间运行脚本,无论是每天、每小时还是每隔几个小时。
def setup_schedule():
任务 = load_tasks()
对于任务中的任务:
task_id、脚本、间隔、time_str、last_run = 任务
如果间隔==“每天”:
Schedule.every().day.at(time_str).do(run_task, script, task_id)
elif 间隔==“每小时”:
Schedule.every().hour.do(run_task, script, task_id)
elif Interval.startswith("every_"): # 处理自定义间隔
小时 = int(interval.split("_")
。分裂(”_”)
Schedule.every(小时).hours.do(run_task,脚本,task_id)
推动未来发展的顶级现代编程语言
主循环
主循环负责使整个脚本运行。它调用 setup_scedule 函数来加载任务并检查并运行任何到期的执行任务。 time.sleep 函数告诉脚本在自动再次运行之前要等待多长时间。在这种情况下,它将每秒运行一次。增加数量会减慢速度并更轻松地占用系统资源,但会牺牲计时精度。
如果 __name__ == "__main__":
setup_schedule()
print("调度程序正在运行...")
而真实:
而真实:
调度.run_pending()
时间.睡眠(1)
将此脚本粘贴到我们在上一篇指南中创建的 src 文件夹中创建的新文件中。使用要在特定时间运行的文件更新数据库,然后让脚本运行以查看其实际情况。