如果您一直在遵循我們的 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 資料夾中建立的新檔案中。使用要在特定時間運行的文件更新資料庫，然後讓腳本運行以查看其實際情況。
