epic

news

task/task_list.py

@@ -0,0 +1,68 @@
+import sqlite3
+import datetime
+import schedule
+import time
+
+# SQLite数据库文件名
+DB_FILE = 'tasks.db'
+
+
+# 初始化数据库
+def init_db():
+    conn = sqlite3.connect(DB_FILE)
+    c = conn.cursor()
+    c.execute('''CREATE TABLE IF NOT EXISTS tasks
+                 (id INTEGER PRIMARY KEY AUTOINCREMENT,
+                  description TEXT NOT NULL,
+                  reminder_time TEXT NOT NULL)''')
+    conn.commit()
+    conn.close()
+
+
+# 添加任务
+def add_task(description, time_str):
+    conn = sqlite3.connect(DB_FILE)
+    c = conn.cursor()
+    time_obj = datetime.datetime.strptime(time_str, '%Y-%m-%d %H:%M:%S')
+    c.execute("INSERT INTO tasks (description, reminder_time) VALUES (?, ?)",
+              (description, time_obj.isoformat()))
+    conn.commit()
+    conn.close()
+    print(f"Task '{description}' added for {time_str}")
+
+
+# 检查并执行任务
+def check_tasks():
+    now = datetime.datetime.now().isoformat()
+    conn = sqlite3.connect(DB_FILE)
+    c = conn.cursor()
+    c.execute("SELECT * FROM tasks WHERE reminder_time <= ?", (now,))
+    tasks = c.fetchall()
+    for task in tasks:
+        print(f"Executing task: {task[1]}")
+        # 如果只想执行一次，则删除任务
+        c.execute("DELETE FROM tasks WHERE id = ?", (task[0],))
+    conn.commit()
+    conn.close()
+
+
+# 定时检查任务
+schedule.every(1).minute.do(check_tasks)  # 每分钟检查一次任务
+
+# 示例使用
+if __name__ == "__main__":
+    init_db()  # 初始化数据库（如果尚未初始化）
+
+    # 添加示例任务（注意：时间应该设置为未来的时间以测试）
+    add_task("Buy groceries", "2024-12-20 14:02:50")  # 修改为未来时间进行测试
+    add_task("Attend meeting", "2024-12-20 14:03:50")  # 修改为未来时间进行测试
+
+    # 由于我们设置了未来的时间，因此需要使用一个足够长的时间循环来等待任务执行
+    # 在实际应用中，你可能会有一个更优雅的方式来停止循环（例如监听特定事件）
+    print("Starting task scheduler...")
+    try:
+        while True:
+            schedule.run_pending()
+            time.sleep(1)  # 等待下一分钟
+    except KeyboardInterrupt:
+        print("Scheduler stopped manually.")