python: Task Scheduler
·
一、引言
本节介绍 DDD 分层架构下任务调度系统的整体设计目标、技术选型（Python 3.11、PyCharm、ttkbootstrap、SQLAlchemy、FastAPI 等）以及文章要解决的核心问题，帮助读者快速建立对系统的整体认知。
二、领域层设计
本节讲解领域层各核心实体的职责划分与数据模型设计，说明如何通过 dataclass 和枚举清晰表达任务、分组、告警、重试策略等业务概念。
- 任务与分组实体：介绍 TaskEntity、TaskGroupEntity 的字段设计，以及任务调度类型（ONCE、INTERVAL、CRON）和状态机（IDLE、RUNNING、PAUSE）的定义。
- 告警与重试策略：介绍 AlertEntity、TaskRetryPolicyEntity 如何支撑任务失败后的通知与自动重试机制。
- 执行历史与自定义异常：介绍 TaskHistoryEntity 记录每次执行结果，以及 DomainBaseException 体系如何统一处理领域层异常。
三、基础设施与配置层
本节讲解全局配置加载、数据库初始化、备份清理、后台调度线程等基础设施组件的实现思路，说明它们如何为上层应用提供稳定支撑。
- 全局配置加载：介绍 app_settings.py 如何读取 app.yaml，并通过 GlobalConfig 统一管理应用、数据库、备份、托盘、UI、调度、告警、API 等配置。
- 数据库初始化与备份清理：介绍 db_init.py 建表流程，以及 db_backup_cleaner.py 如何实现定时备份与过期文件自动清理。
- 后台调度与告警发送：介绍 BackgroundSchedulerThread 轮询调度逻辑，以及 alert_sender.py 如何通过钉钉 webhook 和 SMTP 邮件发送失败告警。
四、实战示例：任务调度系统完整实现
本节以 PyTaskScheduler 项目为例，从主入口 main.py 到任务列表界面 task_view.py，完整演示如何将领域层、基础设施层、应用服务层和界面层串联起来，实现一个可运行的桌面任务调度系统。
- 系统启动流程：讲解 main.py 中加载配置、初始化数据库、启动调度线程、初始化 GUI 与系统托盘的完整步骤。
- 任务管理界面：讲解 task_view.py 中任务列表的增删改查、JSON 导入导出、手动立即执行以及自动刷新等功能的实现。
- 运行效果与验证：展示系统运行截图，说明如何验证任务调度、告警通知和数据库备份等核心功能是否正常。
五、总结
本节回顾 DDD 分层架构在任务调度系统中的落地效果，总结领域驱动设计带来的可维护性、可扩展性优势，并给出后续可继续优化的方向，如分布式调度、任务编排和更丰富的告警渠道。
项目结构：
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 14:48
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : alert_entity.py
"""
领域层：告警配置实体
支持 mail / dingtalk / wecom / wechat
"""
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
@dataclass
class AlertEntity:
alert_id: Optional[int]
alert_name: str
alert_type: str
webhook_url: str
mail_smtp_server: str
mail_smtp_port: int
mail_user: str
mail_password: str
mail_to: str
is_enable: bool
created_at: datetime
updated_at: datetime
encoding: utf-8
版权所有 2026 ©涂聚文有限公司™ ®
许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
描述：
Author : geovindu,Geovin Du 涂聚文.
IDE : PyCharm 2024.3.6 python 3.11
os : windows 10
database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
Datetime : 2026/9/8 12:23
User : geovindu
Product : PyCharm
Project : PyTaskScheduler
File : domain_exception.py
"""
领域层自定义异常
"""
class DomainBaseException(Exception):
"""
领域基础异常
"""
pass
class TaskInvalidException(DomainBaseException):
"""
任务数据校验失败异常
"""
pass
class TaskNotFoundException(DomainBaseException):
"""
任务不存在异常
"""
pass
encoding: utf-8
版权所有 2026 ©涂聚文有限公司™ ®
许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
描述：
Author : geovindu,Geovin Du 涂聚文.
IDE : PyCharm 2024.3.6 python 3.11
os : windows 10
database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
Datetime : 2026/9/8 12:24
User : geovindu
Product : PyCharm
Project : PyTaskScheduler
File : scheduler_enum.py
"""
领域层：调度枚举
"""
from enum import Enum
class TaskScheduleType(Enum):
ONCE = "once" #只执行一次
INTERVAL = "interval" #间隔秒
CRON = "cron" #cron表达式
class TaskStatus(Enum):
IDLE = "idle" #空闲
RUNNING = "running" #执行中
PAUSE = "pause" #暂停
encoding: utf-8
版权所有 2026 ©涂聚文有限公司™ ®
许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
描述：
Author : geovindu,Geovin Du 涂聚文.
IDE : PyCharm 2024.3.6 python 3.11
os : windows 10
database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
Datetime : 2026/9/8 12:24
User : geovindu
Product : PyCharm
Project : PyTaskScheduler
File : task_entity.py
"""
领域层：任务主实体
"""
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
from domain.scheduler_enum import TaskScheduleType, TaskStatus
from domain.task_retry_entity import TaskRetryPolicyEntity
@dataclass
class TaskEntity:
task_id: Optional[int] = None
task_name: str = ""
group_id: Optional[int] = None
schedule_type: TaskScheduleType = TaskScheduleType.CRON
cron_expr: str = ""
interval_seconds: int = 60
command: str = ""
is_enable: bool = True
task_status: TaskStatus = TaskStatus.IDLE
last_execute_time: Optional[datetime] = None
next_execute_time: Optional[datetime] = None
retry_policy: Optional[TaskRetryPolicyEntity] = None
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
timeout_sec: int = 300
max_retry: int = 0
retry_gap_sec: int = 10
remark: str = ""
def is_should_run(self, last_run_time: Optional[datetime], now: datetime) -> bool:
"""依据调度类型判断当前是否应触发执行（调度线程轮询调用）"""
if not self.is_enable:
return False
if self.schedule_type == TaskScheduleType.ONCE:
return last_run_time is None
if self.schedule_type == TaskScheduleType.INTERVAL:
if last_run_time is None:
return True
return (now - last_run_time).total_seconds() >= self.interval_seconds
# CRON：最近一次cron触发时刻是否晚于上次执行时间
try:
from croniter import croniter
itr = croniter(self.cron_expr, now)
prev = itr.get_prev(datetime)
if last_run_time is None:
return True
return prev > last_run_time
except Exception:
return False
encoding: utf-8
版权所有 2026 ©涂聚文有限公司™ ®
许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
描述：
Author : geovindu,Geovin Du 涂聚文.
IDE : PyCharm 2024.3.6 python 3.11
os : windows 10
database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
Datetime : 2026/9/8 14:46
User : geovindu
Product : PyCharm
Project : PyTaskScheduler
File : task_group_entity.py
"""
领域层：任务分组实体
"""
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
@dataclass
class TaskGroupEntity:
group_id: Optional[int]
group_name: str
group_desc: str
created_at: datetime
updated_at: datetime
encoding: utf-8
版权所有 2026 ©涂聚文有限公司™ ®
许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
描述：
Author : geovindu,Geovin Du 涂聚文.
IDE : PyCharm 2024.3.6 python 3.11
os : windows 10
database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
Datetime : 2026/9/8 12:27
User : geovindu
Product : PyCharm
Project : PyTaskScheduler
File : task_history_entity.py
"""
领域层：任务执行历史实体
"""
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
@dataclass
class TaskHistoryEntity:
history_id: Optional[int]
task_id: int
execute_start_time: Optional[datetime] = None
execute_end_time: Optional[datetime] = None
is_success: bool = False
execute_msg: str = ""
created_at: Optional[datetime] = None
encoding: utf-8
版权所有 2026 ©涂聚文有限公司™ ®
许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
描述：
Author : geovindu,Geovin Du 涂聚文.
IDE : PyCharm 2024.3.6 python 3.11
os : windows 10
database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
Datetime : 2026/9/8 14:47
User : geovindu
Product : PyCharm
Project : PyTaskScheduler
File : task_retry_entity.py
"""
领域层：任务失败重试策略实体
"""
from dataclasses import dataclass
from typing import Optional
@dataclass
class TaskRetryPolicyEntity:
policy_id: Optional[int]
enable_retry: bool
max_retry_count: int
retry_interval_sec: int
encoding: utf-8
版权所有 2026 ©涂聚文有限公司™ ®
许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
描述：
Author : geovindu,Geovin Du 涂聚文.
IDE : PyCharm 2024.3.6 python 3.11
os : windows 10
database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
Datetime : 2026/9/8 12:28
User : geovindu
Product : PyCharm
Project : PyTaskScheduler
File : app_settings.py
"""
app_settings.py
全局配置单例，读取app.yaml
"""
import os
import yaml
from dataclasses import dataclass
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(file)))
CONFIG_PATH = os.path.join(BASE_DIR, "config", "app.yaml")
@dataclass
class AppConfig:
app_name: str
log_level: str
log_dir: str
@dataclass
class DatabaseConfig:
db_file: str
echo_sql: bool
@dataclass
class BackupConfig:
cron_expr: str
backup_dir: str
keep_backup_days: int = 30
@dataclass
class TrayConfig:
enable_tray: bool
start_minimize: bool
icon_path: str
@dataclass
class UiConfig:
main_window_width: int
main_window_height: int
dashboard_refresh_ms: int
@dataclass
class SchedulerConfig:
scan_interval_ms: int
concurrent_max: int
@dataclass
class AlertConfig:
enable_alert: bool
notify_type: str
dingtalk_webhook: str
smtp_server: str
smtp_port: int
smtp_user: str
smtp_password: str
email_to: str
@dataclass
class ApiConfig:
api_host: str
api_port: int
api_token: str
@dataclass
class GlobalConfig:
app: AppConfig
database: DatabaseConfig
backup: BackupConfig
tray: TrayConfig
ui: UiConfig
scheduler: SchedulerConfig
alert: AlertConfig
api: ApiConfig
def load_config() -> GlobalConfig:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
raw = yaml.safe_load(f)
return GlobalConfig(
app=AppConfig(**raw["app"]),
database=DatabaseConfig(**raw["database"]),
backup=BackupConfig(**raw["backup"]),
tray=TrayConfig(**raw["tray"]),
ui=UiConfig(**raw["ui"]),
scheduler=SchedulerConfig(**raw["scheduler"]),
alert=AlertConfig(**raw["alert"]),
api=ApiConfig(**raw["api"])
)
g_app_settings = load_config()
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 14:55
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : alert_sender.py
"""
alert_sender.py
告警发送组件，支持钉钉机器人webhook、SMTP邮件
读取g_app_settings告警配置，任务执行失败时调用
"""
import logging
import requests
import smtplib
from email.mime.text import MIMEText
from email.header import Header
from config.app_settings import g_app_settings
from domain.task_entity import TaskEntity
logger = logging.getLogger(__name__)
class AlertSender:
def __init__(self):
self.cfg = g_app_settings.alert
def send_task_fail_alert(self, task: TaskEntity, error_msg: str):
"""任务失败统一告警入口"""
if not self.cfg.enable_alert:
return
notify_type = self.cfg.notify_type
content = self._build_alert_content(task, error_msg)
if notify_type == "dingtalk":
self._send_dingtalk(content)
elif notify_type == "email":
self._send_email(content)
else:
logger.debug("告警类型为none，不发送通知")
def _build_alert_content(self, task: TaskEntity, error_msg: str) -> str:
text = f"""【任务调度告警】
任务ID：{task.task_id}
任务名称：{task.task_name}
Cron表达式：{task.cron_expr}
执行命令：{task.command}
执行失败详情：
{error_msg}
"""
return text
def _send_dingtalk(self, content: str):
"""钉钉机器人webhook"""
webhook = self.cfg.dingtalk_webhook
if not webhook:
logger.warning("钉钉webhook未配置，跳过钉钉告警")
return
payload = {
"msgtype": "text",
"text": {
"content": content
}
}
try:
resp = requests.post(webhook, json=payload, timeout=10)
resp.raise_for_status()
logger.info("钉钉告警发送成功")
except Exception as e:
logger.error(f"钉钉告警发送失败 {str(e)}")
def _send_email(self, content: str):
"""SMTP发送邮件告警"""
smtp_server = self.cfg.smtp_server
smtp_port = self.cfg.smtp_port
smtp_user = self.cfg.smtp_user
smtp_pwd = self.cfg.smtp_password
to_addr = self.cfg.email_to
if not all([smtp_server, smtp_user, smtp_pwd, to_addr]):
logger.warning("邮件告警配置不全，跳过邮件发送")
return
msg = MIMEText(content, "plain", "utf-8")
msg["Subject"] = Header("DDD任务调度系统-任务失败告警", "utf-8")
msg["From"] = smtp_user
msg["To"] = to_addr
try:
if int(smtp_port) ==465:
server = smtplib.SMTP_SSL(smtp_server, int(smtp_port), timeout=15)
else:
server = smtplib.SMTP(smtp_server, int(smtp_port), timeout=15)
server.starttls()
server.login(smtp_user, smtp_pwd)
server.sendmail(smtp_user, [to_addr], msg.as_string())
server.quit()
logger.info("邮件告警发送成功")
except Exception as e:
logger.error(f"邮件告警发送失败 {str(e)}")
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 14:56
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : db_backup_cleaner.py
"""
db_backup_cleaner.py
数据库备份 + 旧备份文件自动清理
读取app.yaml配置：backup_dir、backup_cron、keep_backup_days
"""
import os
import shutil
import glob
import logging
from datetime import datetime, timedelta
from config.app_settings import g_app_settings
logger = logging.getLogger(__name__)
class DbBackupCleaner:
def __init__(self):
self.db_cfg = g_app_settings.database
self.cfg = g_app_settings.backup
self.db_file = self.db_cfg.db_file
self.backup_dir = self.cfg.backup_dir
self.keep_days = self.cfg.keep_backup_days
# 备份目录不存在自动创建
if not os.path.exists(self.backup_dir):
os.makedirs(self.backup_dir, exist_ok=True)
def run_backup(self):
"""执行数据库备份，生成文件名 task_20260908_221000.db"""
if not os.path.exists(self.db_file):
logger.warning(f"数据库文件不存在 {self.db_file}，跳过备份")
return
now = datetime.now()
time_str = now.strftime("%Y%m%d_%H%M%S")
bak_filename = f"task_{time_str}.db"
bak_full_path = os.path.join(self.backup_dir, bak_filename)
try:
shutil.copy2(self.db_file, bak_full_path)
logger.info(f"数据库备份完成：{bak_full_path}")
except Exception as e:
logger.error(f"数据库备份失败 {str(e)}", exc_info=True)
def clean_old_backups(self):
"""删除超过保留天数的备份文件"""
expire_time = datetime.now() - timedelta(days=self.keep_days)
pattern = os.path.join(self.backup_dir, "task_*.db")
file_list = glob.glob(pattern)
del_count = 0
for fpath in file_list:
try:
mtime = datetime.fromtimestamp(os.path.getmtime(fpath))
if mtime < expire_time:
os.remove(fpath)
del_count += 1
logger.debug(f"清理过期备份：{fpath}")
except Exception as e:
logger.warning(f"清理备份文件失败 {fpath} : {str(e)}")
if del_count > 0:
logger.info(f"备份清理完成，删除{del_count}个过期备份")
def backup_and_clean(self):
"""一次完成：备份 + 清理旧文件，供定时任务调用"""
self.run_backup()
self.clean_old_backups()
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 15:33
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : db_init.py
"""
db_init.py
数据库初始化入口
导入全部ORM模型，执行create_all完成建表
"""
import os
from sqlalchemy import create_engine
from config.app_settings import g_app_settings
# 全部ORM模型，必须全部import，否则metadata识别不到表
from infrastructure.orm_task_model import Base as TaskBase
from infrastructure.orm_task_history_model import Base as HistoryBase
from infrastructure.orm_alert_model import Base as AlertBase
from infrastructure.orm_retry_model import Base as RetryBase
def get_sqlite_engine():
db_path = g_app_settings.database.db_file
conn_str = f"sqlite:///{db_path}"
engine = create_engine(
conn_str,
echo=g_app_settings.database.echo_sql,
connect_args={"check_same_thread": False}
)
return engine
def init_database():
"""初始化数据库，不存在则创建全部数据表"""
engine = get_sqlite_engine()
# 全部Base元数据执行建表，存在则跳过
TaskBase.metadata.create_all(engine)
HistoryBase.metadata.create_all(engine)
AlertBase.metadata.create_all(engine)
RetryBase.metadata.create_all(engine)
return engine
if __name__ == "__main__":
init_database()
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 12:29
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : db_session.py
"""
基础设施层：SQLAlchemy会话工厂，SQLite
"""
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from config.app_settings import g_app_settings
SQLITE_DB_PATH = g_app_settings.database.db_file
engine = create_engine(
f"sqlite:///{SQLITE_DB_PATH}",
connect_args={"check_same_thread": False},
echo=False
)
# 启动时自动建表（缺表则创建，已存在则跳过）
from infrastructure.orm_task_model import Base as TaskBase
from infrastructure.orm_task_history_model import Base as HistoryBase
from infrastructure.orm_alert_model import Base as AlertBase
from infrastructure.orm_retry_model import Base as RetryBase
TaskBase.metadata.create_all(engine)
HistoryBase.metadata.create_all(engine)
AlertBase.metadata.create_all(engine)
RetryBase.metadata.create_all(engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_engine():
return engine
def get_session():
db: Session = SessionLocal()
try:
yield db
finally:
db.close()
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 16:08
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : global_exception_catcher.py
"""
global_exception_catcher.py
全局异常捕获
1. tkinter主线程未捕获异常弹窗
2. 后台线程异常捕获装饰器
"""
import sys
import threading
import logging
import tkinter as tk
from tkinter import messagebox
logger = logging.getLogger(__name__)
class GlobalExceptionCatcher:
def __init__(self, root_window: tk.Tk):
self.root = root_window
# 接管tkinter主线程异常钩子
self.old_tk_error = self.root.report_callback_exception
self.root.report_callback_exception = self._tk_callback_exception
# 接管系统全局异常钩子
sys.excepthook = self._sys_excepthook
def _tk_callback_exception(self, exc, val, tb):
"""Tkinter UI主线程异常"""
logger.error("UI主线程发生未捕获异常", exc_info=(exc, val, tb))
self._show_error_dialog(str(val))
if self.old_tk_error:
self.old_tk_error(exc, val, tb)
def _sys_excepthook(self, exc_type, exc_value, exc_traceback):
"""主线程非tk异常"""
logger.error("主线程未捕获异常", exc_info=(exc_type, exc_value, exc_traceback))
self.root.after(0, lambda: self._show_error_dialog(str(exc_value)))
def _show_error_dialog(self, err_msg: str):
"""弹出错误弹窗，after(0)保证在UI线程执行"""
messagebox.showerror("程序异常", f"发生未捕获异常：\n{err_msg}")
@staticmethod
def thread_catch_exc(func):
"""后台线程函数装饰器，捕获子线程异常，抛到UI弹窗"""
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
logger.error(f"后台线程异常 {func.__name__}", exc_info=True)
# 拿到根窗口，调度到UI线程弹窗
root = tk._default_root
if root:
root.after(0, lambda: messagebox.showerror("后台线程异常", f"{func.__name__}：{str(e)}"))
return wrapper
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 14:32
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : gui_log_handler.py
"""
gui_log_handler.py
GUI日志处理器：logging的Handler，把日志输出到tk/ttk文本控件
线程安全：后台线程仅写入queue，由主线程(root.after轮询)取队列更新UI
"""
import logging
import queue
import tkinter as tk
class GuiLogHandler(logging.Handler):
POLL_MS = 100
def __init__(self, text_widget):
super().__init__()
self.text_widget = text_widget
self.root = text_widget.winfo_toplevel()
self.msg_queue = queue.Queue()
self.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S"))
# 在主线程启动轮询链（跨线程更新UI的唯一入口）
self.root.after(self.POLL_MS, self.poll)
def emit(self, record):
"""后台线程可安全调用：仅写入线程安全队列，不做任何Tk操作"""
try:
self.msg_queue.put(self.format(record))
except Exception:
self.handleError(record)
def poll(self):
"""主线程轮询：取出队列日志写入文本控件，再调度下一次"""
try:
while True:
msg = self.msg_queue.get_nowait()
self._append_log(msg)
except queue.Empty:
pass
self.root.after(self.POLL_MS, self.poll)
def _append_log(self, msg: str):
self.text_widget.config(state=tk.NORMAL)
self.text_widget.insert(tk.END, msg + "\n")
self.text_widget.see(tk.END)
self.text_widget.config(state=tk.DISABLED)
def clear(self):
"""清空日志面板"""
self._drain()
self.text_widget.config(state=tk.NORMAL)
self.text_widget.delete("1.0", tk.END)
self.text_widget.config(state=tk.DISABLED)
def _drain(self):
try:
while True:
self.msg_queue.get_nowait()
except queue.Empty:
pass
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 14:55
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : json_import_export.py
"""
json_import_export.py
任务JSON导入导出功能
导出：全部任务导出为json文件
导入：读取json批量新增/覆盖任务，跳过历史记录
"""
import json
import os
import logging
from typing import List, Dict
from domain.task_entity import TaskEntity
from domain.scheduler_enum import TaskStatus
from infrastructure.db_session import get_session
from infrastructure.task_repository import TaskRepository
logger = logging.getLogger(__name__)
def _parse_status(raw: str) -> TaskStatus:
"""兼容大小写解析任务状态，如 IDLE/idle/Running"""
if not raw:
return TaskStatus.IDLE
upper = str(raw).upper()
for member in TaskStatus:
if member.name.upper() == upper or str(member.value).upper() == upper:
return member
return TaskStatus.IDLE
class TaskJsonImportExport:
def __init__(self):
self.repo = TaskRepository()
def export_all_to_json(self, file_path: str):
"""导出全部任务到JSON文件"""
gen = get_session()
db = next(gen)
entity_list = self.repo.get_all(db)
db.close()
export_list = []
for e in entity_list:
item = {
"task_id": e.task_id,
"task_name": e.task_name,
"cron_expr": e.cron_expr,
"command": e.command,
"timeout_sec": e.timeout_sec,
"max_retry": e.max_retry,
"retry_gap_sec": e.retry_gap_sec,
"is_enable": e.is_enable,
"remark": e.remark,
"task_status": e.task_status.name
}
export_list.append(item)
with open(file_path, "w", encoding="utf-8") as f:
json.dump(export_list, f, ensure_ascii=False, indent=2)
def import_from_json(self, file_path: str) -> Dict[str, int]:
"""
从JSON导入任务
返回统计：{"total":总条数,"insert":新增,"update":更新,"skip":跳过}
"""
if not os.path.exists(file_path):
raise FileNotFoundError(f"文件不存在 {file_path}")
with open(file_path, "r", encoding="utf-8") as f:
data = json.load(f)
# 兼容：整个文件是单个对象{} 或 对象数组[]
if isinstance(data, dict):
raw_list = [data]
else:
raw_list = data
stat = {"total": len(raw_list), "insert": 0, "update": 0, "skip": 0}
gen = get_session()
db = next(gen)
for raw in raw_list:
try:
if not isinstance(raw, dict):
logger.warning(f"导入跳过：JSON元素非对象 {raw!r}")
stat["skip"] += 1
continue
task_id = raw.get("task_id", 0)
entity = TaskEntity(
task_id=task_id,
task_name=raw["task_name"],
cron_expr=raw["cron_expr"],
command=raw["command"],
timeout_sec=int(raw["timeout_sec"]),
max_retry=int(raw["max_retry"]),
retry_gap_sec=int(raw["retry_gap_sec"]),
is_enable=bool(raw["is_enable"]),
remark=raw.get("remark", ""),
task_status=_parse_status(raw.get("task_status", "IDLE"))
)
exist = self.repo.get_by_id(db, task_id)
if task_id <=0 or exist is None:
entity.task_id = 0
self.repo.insert(db, entity)
stat["insert"] += 1
else:
self.repo.update(db, entity)
stat["update"] +=1
except Exception as e:
logger.warning(f"导入一条任务跳过:{str(e)}")
stat["skip"] +=1
db.commit()
db.close()
return stat
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 14:52
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : orm_alert_model.py
"""
orm_alert_model.py ORM模型：告警发送记录表
记录每一次告警发送日志，方便排查告警是否发出
"""
from sqlalchemy import Column, Integer, String, Text, DateTime, Boolean
from sqlalchemy.orm import declarative_base
from datetime import datetime
Base = declarative_base()
class AlertLogOrm(Base):
__tablename__ = "t_alert_log"
id = Column(Integer, primary_key=True, autoincrement=True, comment="主键ID")
task_id = Column(Integer, nullable=True, comment="关联任务ID")
alert_type = Column(String(32), nullable=False, comment="告警类型 dingtalk / email")
send_success = Column(Boolean, default=False, comment="是否发送成功")
content = Column(Text, nullable=True, comment="告警内容")
error_msg = Column(Text, nullable=True, comment="发送失败异常信息")
create_time = Column(DateTime, default=datetime.now, comment="记录时间")
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 14:51
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : orm_group_model.py
"""
ORM：任务分组表
"""
from sqlalchemy import Column, Integer, String, DateTime, Text
from infrastructure.orm_model import Base
import datetime
class TaskGroupOrm(Base):
__tablename__ = "task_group"
group_id = Column(Integer, primary_key=True, autoincrement=True)
group_name = Column(String(128), nullable=False)
group_desc = Column(Text, nullable=True)
created_at = Column(DateTime, default=datetime.datetime.now)
updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 14:52
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : orm_retry_model.py
"""
orm_retry_model.py ORM模型：任务重试记录表
记录每次任务重试明细，便于排查多次失败原因
"""
from sqlalchemy import Column, Integer, String, Text, DateTime, Boolean
from sqlalchemy.orm import declarative_base
from datetime import datetime
Base = declarative_base()
class TaskRetryLogOrm(Base):
__tablename__ = "t_task_retry_log"
id = Column(Integer, primary_key=True, autoincrement=True, comment="主键")
task_id = Column(Integer, nullable=False, comment="任务ID")
retry_index = Column(Integer, nullable=False, comment="第几次重试，0=首次执行")
is_success = Column(Boolean, default=False, comment="本次是否成功")
output_msg = Column(Text, nullable=True, comment="执行输出")
create_time = Column(DateTime, default=datetime.now, comment="发生时间")
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 19:57
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : orm_task_history_model.py
"""
orm_task_history_model.py 任务执行历史ORM模型
与 domain.task_history_entity.TaskHistoryEntity 字段严格一一对应
"""
from sqlalchemy import Column, Integer, DateTime, Text, Boolean
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class TaskHistoryOrm(Base):
__tablename__ = "t_task_history"
id = Column(Integer, primary_key=True, autoincrement=True, comment="历史主键ID")
task_id = Column(Integer, nullable=False, index=True, comment="关联任务ID")
execute_start_time = Column(DateTime, nullable=False, comment="执行开始时间")
execute_end_time = Column(DateTime, nullable=False, comment="执行结束时间")
is_success = Column(Boolean, nullable=False, default=False, comment="是否执行成功")
execute_msg = Column(Text, nullable=True, comment="命令输出日志")
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 15:52
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : quartz_backup_scheduler.py
"""
quartz_backup_scheduler.py
Quartz风格定时包装，使用APScheduler执行数据库备份任务
读取app.yaml database.backup_cron配置，定时执行DbBackupCleaner.backup_and_clean()
"""
import logging
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from config.app_settings import g_app_settings
from infrastructure.db_backup_cleaner import DbBackupCleaner
logger = logging.getLogger(__name__)
class BackupQuartzScheduler:
def __init__(self):
self.scheduler = BackgroundScheduler()
self.cleaner = DbBackupCleaner()
self.cron_expr = g_app_settings.backup.cron_expr
def start(self):
"""启动备份定时任务"""
try:
trigger = CronTrigger.from_crontab(self.cron_expr)
self.scheduler.add_job(
self.cleaner.backup_and_clean,
trigger=trigger,
id="db_backup_job",
replace_existing=True
)
self.scheduler.start()
logger.info(f"数据库备份定时任务已启动，cron={self.cron_expr}")
except Exception as e:
logger.error(f"备份定时任务启动失败:{str(e)}", exc_info=True)
def stop(self):
"""停止定时任务，程序退出调用"""
if self.scheduler.running:
self.scheduler.shutdown()
logger.info("备份定时调度已停止")
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 12:55
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : scheduler_background_thread.py
"""
scheduler_background_thread.py
后台调度线程主循环
使用 task_entity.is_should_run() 做cron时间判断
负责轮询数据库任务、触发执行、控制运行锁、休眠间隔；支持线程stop停止信号
"""
import threading
import time
import logging
from datetime import datetime
from typing import Optional
from infrastructure.db_session import get_session
from infrastructure.task_repository import TaskRepository
from infrastructure.task_history_repository import TaskHistoryRepository
from infrastructure.task_run_lock import TaskRunLock
from application.task_scheduler_app_service import TaskSchedulerAppService
logger = logging.getLogger(__name__)
class BackgroundSchedulerThread(threading.Thread):
def __init__(self):
super().__init__(name="BackgroundSchedulerThread", daemon=True)
self._stop_event = threading.Event()
self.scan_interval_sec = 5 # 调度扫描周期，每5秒扫描一次任务表
self.task_repo = TaskRepository()
self.history_repo = TaskHistoryRepository()
self.run_lock = TaskRunLock()
self.app_service = TaskSchedulerAppService()
# 记录每个任务上一次执行时间 {task_id: last_run_datetime}
self.last_run_map: dict[int, datetime] = {}
def stop(self):
"""安全停止调度线程"""
logger.info("收到停止调度线程信号")
self._stop_event.set()
def is_stopped(self) -> bool:
return self._stop_event.is_set()
def run(self):
"""线程入口主循环"""
logger.info("任务调度后台线程启动，扫描间隔：%d秒", self.scan_interval_sec)
# 初始化：启动时刻作为所有任务初始last_run时间
boot_dt = datetime.now()
while not self.is_stopped():
try:
self.schedule_loop_once(boot_dt)
except Exception as e:
logger.error("调度循环发生异常", exc_info=e)
# 休眠，可被stop信号打断
self._stop_event.wait(timeout=self.scan_interval_sec)
logger.info("任务调度后台线程已退出")
def schedule_loop_once(self, boot_dt: datetime):
"""单次调度扫描逻辑"""
gen = get_session()
db = next(gen)
task_list = self.task_repo.get_all(db)
db.close()
now_dt = datetime.now()
for task in task_list:
if self.is_stopped():
break
# 获取该任务上一次执行时间，没有则使用程序启动时间
last_run_dt: Optional[datetime] = self.last_run_map.get(task.task_id, boot_dt)
# cron调度判断核心：调用领域实体方法 is_should_run
need_run = task.is_should_run(last_run_time=last_run_dt, now=now_dt)
if not need_run:
continue
# 判断任务是否正在运行，避免并发重入
if self.run_lock.is_running(task.task_id):
logger.debug(f"task_id={task.task_id} 正在运行，跳过本次调度触发")
continue
logger.info(f"cron调度触发 task_id={task.task_id} 【{task.task_name}】")
# 提交任务执行（非阻塞，内部新开线程执行）
self.app_service.execute_task_background(task)
# 更新本任务最后执行时间为当前时刻
self.last_run_map[task.task_id] = now_dt
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 12:19
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : system_tray_helper.py
"""
system_tray_helper.py
系统托盘封装 pystray
最小化到托盘、显示窗口、退出程序
"""
import pystray
from PIL import Image
import logging
from typing import Callable
logger = logging.getLogger(__name__)
class SystemTrayHelper:
def __init__(self, window, icon_image: Image):
"""
:param window: tk主窗口对象
:param icon_image:Pillow Image托盘图标
"""
self.window = window
self.icon_image = icon_image
self.tray_icon: pystray.Icon = None
self.on_exit_callback: Optional[Callable] = None
def setup_tray(self):
menu = pystray.Menu(
pystray.MenuItem("显示窗口", self._show_window),
pystray.MenuItem("退出程序", self._on_exit)
)
self.tray_icon = pystray.Icon("task_scheduler", self.icon_image, "DDD任务调度系统", menu)
def run_tray(self):
if self.tray_icon:
self.tray_icon.run_detached()
def stop_tray(self):
if self.tray_icon:
self.tray_icon.stop()
def minimize_to_tray(self):
"""隐藏主窗口，保留托盘"""
self.window.withdraw()
def _show_window(self):
self.window.deiconify()
self.window.lift()
def _on_exit(self):
self.stop_tray()
if self.on_exit_callback:
self.on_exit_callback()
self.window.destroy()
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 14:32
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : task_subprocess_runner.py
"""
task_subprocess_runner.py
子进程执行器：调用操作系统shell执行任务命令，支持超时强制杀死进程
返回 (success:bool, output_msg:str)
"""
import subprocess
import threading
import logging
from typing import Tuple, Optional
logger = logging.getLogger(__name__)
class TaskSubprocessRunner:
def __init__(self, timeout_sec: int):
self.timeout_sec = timeout_sec
def run(self, task_id: int, command: str) -> Tuple[bool, str]:
"""
执行shell命令，超时自动kill
:param task_id: 任务ID
:param command: 要执行的命令字符串
:return: (是否成功, 输出/错误信息)
"""
proc: Optional[subprocess.Popen] = None
output_lines = []
def read_stdout(process: subprocess.Popen):
"""子线程读取stdout"""
try:
for line in iter(process.stdout.readline, b""):
if line:
output_lines.append(line.decode("utf‑8", errors="replace"))
except Exception:
pass
def read_stderr(process: subprocess.Popen):
"""子线程读取stderr"""
try:
for line in iter(process.stderr.readline, b""):
if line:
output_lines.append(line.decode("utf‑8", errors="replace"))
except Exception:
pass
try:
proc = subprocess.Popen(
command,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
creationflags=subprocess.CREATE_NO_WINDOW # Windows不弹出黑窗口
)
# 启动读取线程
t_out = threading.Thread(target=read_stdout, args=(proc,), daemon=True)
t_err = threading.Thread(target=read_stderr, args=(proc,), daemon=True)
t_out.start()
t_err.start()
# 等待进程结束，带超时
ret_code = proc.wait(timeout=self.timeout_sec)
t_out.join(timeout=2)
t_err.join(timeout=2)
full_output = "".join(output_lines)
logger.debug(f"task_id={task_id} 命令返回码:{ret_code}")
if ret_code == 0:
return True, full_output
else:
return False, f"返回码:{ret_code}\n{full_output}"
except subprocess.TimeoutExpired:
if proc is not None:
proc.kill()
msg = f"task_id={task_id} 执行超时({self.timeout_sec}秒)，进程已强制终止"
logger.warning(msg)
return False, msg
except Exception as e:
err_msg = f"task_id={task_id} 启动子进程异常:{str(e)}"
logger.error(err_msg, exc_info=True)
return False, err_msg
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 12:40
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : task_app_service.py
"""
应用服务层：任务CRUD与分页查询业务
不直接操作ORM，依赖仓储接口
"""
from typing import Tuple, Optional, List
from datetime import datetime
from sqlalchemy.orm import Session
from domain.task_entity import TaskEntity
from domain.domain_exception import TaskNotFoundException, TaskInvalidException
from infrastructure.task_repository import TaskRepository
class TaskAppService:
"""
任务应用服务，封装任务增删改查业务
"""
def __init__(self):
self._repo = TaskRepository()
def create_task(self, db: Session, task_entity: TaskEntity) -> TaskEntity:
"""
创建新任务
:param db:数据库会话
:param task_entity:任务领域实体
:return:保存完成实体（带主键ID）
"""
task_entity.task_id = None
task_entity.created_at = datetime.now()
task_entity.updated_at = datetime.now()
task_entity.validate()
saved = self._repo.insert(db, task_entity)
return saved
def modify_task(self, db: Session, task_entity: TaskEntity) -> TaskEntity:
"""
修改任务
:param db:会话
:param task_entity:带task_id的领域实体
:return:更新后实体
"""
exist = self._repo.get_by_id(db, task_entity.task_id)
if exist is None:
raise TaskNotFoundException(f"任务ID {task_entity.task_id}不存在")
task_entity.updated_at = datetime.now()
task_entity.validate()
return self._repo.update(db, task_entity)
def remove_task(self, db: Session, task_id: int):
"""删除任务"""
exist = self._repo.get_by_id(db, task_id)
if exist is None:
raise TaskNotFoundException(f"任务ID {task_id}不存在")
self._repo.delete_by_id(db, task_id)
def get_one(self, db: Session, task_id:int) -> Optional[TaskEntity]:
"""获取单个任务"""
return self._repo.get_by_id(db, task_id)
def query_page(self, db: Session, keyword: str, page_index:int, page_size:int) -> Tuple[List[TaskEntity], int]:
"""
任务模糊分页查询
:param db:会话
:param keyword:模糊关键词
:param page_index:页码从1
:param page_size:每页记录数，支持10/20/50自定义
:return: (实体列表,总记录数)
"""
if page_index < 1:
page_index =1
if page_size <1:
page_size =10
return self._repo.page_query(db, keyword, page_index, page_size)
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 14:13
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : task_notify_service.py
"""
UI通知服务，跨线程消息队列
"""
from queue import Queue
from typing import Callable, Optional
class TaskNotifyService:
def __init__(self):
self._msg_queue = Queue()
self._ui_callback: Optional[Callable[[str], None]] = None
def set_ui_callback(self, cb: Callable[[str], None]):
self._ui_callback = cb
def push_msg(self, msg: str):
self._msg_queue.put(msg)
def process(self):
while not self._msg_queue.empty():
m = self._msg_queue.get()
if self._ui_callback:
self._ui_callback(m)
g_task_notify_service = TaskNotifyService()
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 12:40
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : task_scheduler_app_service.py
"""
task_scheduler_app_service.py
应用服务层：任务业务核心
职责：手动执行、后台调度执行、重试逻辑、保存执行历史、调用子进程运行命令、告警触发
依赖：TaskSubprocessRunner、TaskRunLock、Repository、AlertSender
"""
import threading
import time
import logging
from datetime import datetime
from typing import Optional
from domain.task_entity import TaskEntity
from domain.task_history_entity import TaskHistoryEntity
from domain.scheduler_enum import TaskStatus
from infrastructure.db_session import get_session
from infrastructure.task_repository import TaskRepository
from infrastructure.task_history_repository import TaskHistoryRepository
from infrastructure.task_subprocess_runner import TaskSubprocessRunner
from infrastructure.task_run_lock import TaskRunLock
from infrastructure.alert_sender import AlertSender
logger = logging.getLogger(__name__)
class TaskSchedulerAppService:
def __init__(self):
self.task_repo = TaskRepository()
self.history_repo = TaskHistoryRepository()
self.run_lock = TaskRunLock()
self.alert_sender = AlertSender()
def manual_execute_task(self, task_id: int):
"""API/UI手动触发任务执行，新建后台线程执行"""
gen = get_session()
db = next(gen)
task_entity = self.task_repo.get_by_id(db, task_id)
db.close()
if not task_entity:
raise Exception(f"task_id={task_id} 任务不存在")
if not task_entity.is_enable:
raise Exception(f"task_id={task_id}任务已禁用，无法执行")
# 后台线程运行，不阻塞UI/API
t = threading.Thread(target=self.execute_task_background, args=(task_entity,), daemon=True)
t.start()
def execute_task_background(self, task: TaskEntity):
"""
后台执行任务入口（调度线程、手动执行均调用此方法）
包含重试循环、子进程调用、写入执行历史、告警
"""
task_id = task.task_id
# 获取运行锁
if not self.run_lock.try_acquire(task_id):
logger.warning(f"task_id={task_id} 任务正在运行，拒绝重复执行")
return
# 更新任务状态为运行中
self._update_task_status(task_id, TaskStatus.RUNNING)
execute_start = datetime.now()
final_success = False
final_msg = ""
current_retry = 0
runner = TaskSubprocessRunner(timeout_sec=task.timeout_sec)
try:
while current_retry <= task.max_retry:
success, output_msg = runner.run(task_id, task.command)
if success:
final_success = True
final_msg = output_msg
break
current_retry += 1
final_msg = output_msg
if current_retry > task.max_retry:
break
logger.warning(f"task_id={task_id}执行失败，第{current_retry}次重试，间隔{task.retry_gap_sec}秒")
time.sleep(task.retry_gap_sec)
except Exception as e:
logger.error(f"task_id={task_id}执行发生异常", exc_info=e)
final_success = False
final_msg = f"运行异常：{str(e)}"
finally:
# 释放锁
self.run_lock.release(task_id)
# 写执行历史记录
self._save_task_history(
task_id=task_id,
execute_start=execute_start,
is_success=final_success,
execute_msg=final_msg
)
# 一次性任务执行成功后自动禁用，不再被调度执行
if final_success:
self._auto_disable_if_one_time(task, task_id)
# 更新任务状态回到空闲
self._update_task_status(task_id, TaskStatus.IDLE)
# 失败触发告警
if not final_success:
self.alert_sender.send_task_fail_alert(task, final_msg)
def _update_task_status(self, task_id: int, status: TaskStatus):
"""更新数据库任务状态"""
try:
gen = get_session()
db = next(gen)
entity = self.task_repo.get_by_id(db, task_id)
if entity:
entity.task_status = status
self.task_repo.update_status(db, entity.task_id, status)
db.commit()
db.close()
except Exception as ex:
logger.error(f"更新task_id={task_id}状态失败 {ex}")
@staticmethod
def _is_one_time_task(task: TaskEntity) -> bool:
"""一次性任务识别：备注包含“一次性”且当前已启用"""
return bool(task.remark) and "一次性" in task.remark
def _update_task_enable(self, task_id: int, is_enable: bool):
"""更新数据库任务启用开关"""
try:
gen = get_session()
db = next(gen)
self.task_repo.update_enable(db, task_id, is_enable)
db.commit()
db.close()
except Exception as ex:
logger.error(f"更新task_id={task_id}启用状态失败 {ex}")
def _auto_disable_if_one_time(self, task: TaskEntity, task_id: int):
"""一次性任务执行成功后自动禁用，使其不再被调度"""
if self._is_one_time_task(task):
logger.info(f"task_id={task_id} 为一次性任务且执行成功，自动禁用")
self._update_task_enable(task_id, False)
def _save_task_history(self, task_id: int, execute_start: datetime, is_success: bool, execute_msg: str):
"""保存任务执行历史记录"""
try:
hist_entity = TaskHistoryEntity(
history_id=0,
task_id=task_id,
execute_start_time=execute_start,
execute_end_time=datetime.now(),
is_success=is_success,
execute_msg=execute_msg
)
gen = get_session()
db = next(gen)
self.history_repo.save(db, hist_entity)
db.commit()
db.close()
except Exception as ex:
logger.error(f"保存任务历史失败 task_id={task_id}", exc_info=ex)
def execute_single_task(self, task_entity: TaskEntity):
"""
手动/调度触发执行单个任务
:param task_entity: 任务领域实体
:return: 本次执行历史实体
"""
logger.info(f"开始执行任务 task_id={task_entity.task_id} 任务名称={task_entity.task_name}")
start_time = datetime.now()
runner = TaskSubprocessRunner(timeout_sec=task_entity.timeout_sec) # timeout_sec
success, output_msg = runner.run(task_entity.task_id, task_entity.command)
end_time = datetime.now()
# 组装执行历史领域实体
history_entity = TaskHistoryEntity(
history_id=0,
task_id=task_entity.task_id,
execute_start_time=start_time,
execute_end_time=end_time,
is_success=success,
execute_msg=output_msg
)
# 写入数据库
gen = get_session()
db = next(gen)
try:
saved_history = self.history_repo.save(db, history_entity)
finally:
db.close()
if success:
logger.info(f"任务执行成功 task_id={task_entity.task_id}")
else:
logger.error(f"任务执行失败 task_id={task_entity.task_id}, output={output_msg}")
return saved_history
def execute_single_task(self, task_entity: TaskEntity):
"""
手动/调度触发执行单个任务
:param task_entity: 任务领域实体
:return: 本次执行历史实体
"""
logger.info(f"开始执行任务 task_id={task_entity.task_id} 任务名称={task_entity.task_name}")
start_time = datetime.now()
runner = TaskSubprocessRunner(timeout_sec=task_entity.timeout_sec)
success, output_msg = runner.run(task_entity.task_id, task_entity.command)
end_time = datetime.now()
# 组装执行历史领域实体 history_id=0 代表新增
history_entity = TaskHistoryEntity(
history_id=0,
task_id=task_entity.task_id,
execute_start_time=start_time,
execute_end_time=end_time,
is_success=success,
execute_msg=output_msg
)
# 写入数据库
gen = get_session()
db = next(gen)
try:
saved_history = self.history_repo.save(db, history_entity)
finally:
db.close()
if success:
logger.info(f"任务执行成功 task_id={task_entity.task_id}")
else:
logger.error(f"任务执行失败 task_id={task_entity.task_id}, output={output_msg}")
return saved_history
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 14:59
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : dashboard_view.py
"""
dashboard_view.py
仪表盘统计面板
展示：总任务数、启用任务数、历史总执行、成功、失败，最近执行记录
卡片可点击展开明细，带操作按钮（禁用/启用/立即执行）
"""
import tkinter as tk
from tkinter import ttk, messagebox
import ttkbootstrap as bs
from presentation.task_controller import TaskController
from domain.scheduler_enum import TaskStatus
from config.app_settings import g_app_settings
from application.task_scheduler_app_service import TaskSchedulerAppService
from infrastructure.global_exception_catcher import GlobalExceptionCatcher
class DashboardView:
def __init__(self, parent):
self.parent = parent
self.controller = TaskController()
self.app_service = TaskSchedulerAppService()
self._expanded_key = None
self._task_cache = []
self._history_cache = []
self.build_ui(parent)
self.refresh_dashboard()
self.start_auto_refresh()
def build_ui(self, parent):
main_frame = ttk.Frame(parent)
self._frame = main_frame
main_frame.pack(fill="both", expand=True, padx=10, pady=10)
# 顶部统计卡片容器（两行）
self.card_frame = ttk.Frame(main_frame)
self.card_frame.pack(fill="x", pady=8)
self.var_total_task = tk.StringVar(value="0")
self.var_periodic_task = tk.StringVar(value="0")
self.var_onetime_done = tk.StringVar(value="0")
self.var_onetime_undone = tk.StringVar(value="0")
self.var_enable_task = tk.StringVar(value="0")
self.var_running_task = tk.StringVar(value="0")
self.var_total_exec = tk.StringVar(value="0")
self.var_success_exec = tk.StringVar(value="0")
self.var_fail_exec = tk.StringVar(value="0")
self._cards = {}
self._build_card("total", "全部任务", self.var_total_task, "", 0, 0)
self._build_card("periodic", "周期任务(重复)", self.var_periodic_task, "", 0, 1)
self._build_card("onetime_done", "一次性任务(已完成)", self.var_onetime_done, "success", 0, 2)
self._build_card("onetime_undone", "一次性任务(未完成)", self.var_onetime_undone, "warning", 0, 3)
self._build_card("enabled", "已启用任务", self.var_enable_task, "", 0, 4)
self._build_card("running", "正在执行", self.var_running_task, "info", 0, 5)
self._build_card("total_exec", "历史总执行", self.var_total_exec, "", 1, 0)
self._build_card("success", "成功次数", self.var_success_exec, "success", 1, 1)
self._build_card("fail", "失败次数", self.var_fail_exec, "danger", 1, 2)
# 刷新按钮
btn_refresh = bs.Button(main_frame, text="刷新仪表盘", command=self.refresh_dashboard)
btn_refresh.pack(anchor="e", pady=4)
# 分类明细面板（点击卡片展开）
self.detail_frame = ttk.Frame(main_frame)
# 标题 + 关闭按钮
self.detail_title_var = tk.StringVar(value="")
self.detail_title_frame = ttk.Frame(self.detail_frame)
self.detail_title_frame.pack(fill="x", pady=(6, 2))
bs.Label(self.detail_title_frame, textvariable=self.detail_title_var,
font=("微软雅黑", 11, "bold")).pack(side="left")
ttk.Button(self.detail_title_frame, text="收起", command=self._collapse_detail).pack(side="right")
# 操作按钮行
btn_frame = ttk.Frame(self.detail_frame)
btn_frame.pack(fill="x", pady=2)
bs.Button(btn_frame, text="禁用选中", command=self._toggle_disable,
bootstyle="danger-outline").pack(side="left", padx=4)
bs.Button(btn_frame, text="启用选中", command=self._toggle_enable,
bootstyle="success-outline").pack(side="left", padx=4)
bs.Button(btn_frame, text="立即执行", command=self._execute_selected,
bootstyle="info-outline").pack(side="left", padx=4)
# 明细 Treeview
detail_cols = ("task_id", "task_name", "status", "is_enable", "last_run", "remark")
self.detail_tree = ttk.Treeview(self.detail_frame, columns=detail_cols,
show="headings", height=6)
self.detail_tree.heading("task_id", text="ID")
self.detail_tree.heading("task_name", text="任务名称")
self.detail_tree.heading("status", text="状态")
self.detail_tree.heading("is_enable", text="启用")
self.detail_tree.heading("last_run", text="上次执行")
self.detail_tree.heading("remark", text="备注")
self.detail_tree.column("task_id", width=50, anchor="center")
self.detail_tree.column("task_name", width=180, anchor="w")
self.detail_tree.column("status", width=70, anchor="center")
self.detail_tree.column("is_enable", width=60, anchor="center")
self.detail_tree.column("last_run", width=140, anchor="center")
self.detail_tree.column("remark", width=220, anchor="w")
vsb = ttk.Scrollbar(self.detail_frame, orient="vertical",
command=self.detail_tree.yview)
self.detail_tree.configure(yscrollcommand=vsb.set)
self.detail_tree.pack(side="left", fill="both", expand=True)
vsb.pack(side="right", fill="y")
# 分隔线
ttk.Separator(main_frame, orient="horizontal").pack(fill="x", pady=4)
# 最近执行历史表格
ttk.Label(main_frame, text="最近执行记录", font=("微软雅黑", 11)).pack(anchor="w", pady=4)
cols = ("task_id", "task_name", "time", "result")
self.tree = ttk.Treeview(main_frame, columns=cols, show="headings")
self.tree.heading("task_id", text="任务ID")
self.tree.heading("task_name", text="任务名称")
self.tree.heading("time", text="执行时间")
self.tree.heading("result", text="结果")
self.tree.column("task_id", width=80, anchor="center")
self.tree.column("task_name", width=240, anchor="w")
self.tree.column("time", width=200, anchor="center")
self.tree.column("result", width=90, anchor="center")
vsb2 = ttk.Scrollbar(main_frame, orient="vertical", command=self.tree.yview)
self.tree.configure(yscrollcommand=vsb2.set)
self.tree.pack(fill="both", expand=True)
vsb2.pack(side="right", fill="y")
def _build_card(self, key, text, var, bootstyle, row, col):
frame = bs.LabelFrame(self.card_frame, text=text, cursor="hand2")
frame.grid(row=row, column=col, padx=5, pady=3, sticky="nsew")
lbl = bs.Label(frame, textvariable=var, font=("微软雅黑", 14),
bootstyle=bootstyle if bootstyle else "primary")
lbl.grid(row=0, column=0, padx=22, pady=15)
self._cards[key] = (frame, var, text)
# 点击卡片展开明细
for widget in [frame, lbl]:
widget.bind("<Button-1>", lambda e, k=key, t=text: self._toggle_detail(k, t))
def _toggle_detail(self, key, title):
if self._expanded_key == key:
self._collapse_detail()
return
self._expanded_key = key
self.detail_title_var.set(f"{title} — 明细")
self.detail_frame.pack(fill="both", expand=True, padx=4, pady=(4, 8))
self._populate_detail(key)
def _collapse_detail(self):
self._expanded_key = None
self.detail_frame.pack_forget()
def _populate_detail(self, key):
for item in self.detail_tree.get_children():
self.detail_tree.delete(item)
for t in self._task_cache:
if not self._match_category(key, t):
continue
status_text = t.task_status.name if t.task_status else ""
enable_text = "是" if t.is_enable else "否"
last_run = ""
if t.last_execute_time:
last_run = t.last_execute_time.strftime("%Y-%m-%d %H:%M")
self.detail_tree.insert("", tk.END, iid=str(t.task_id),
values=(t.task_id, t.task_name, status_text,
enable_text, last_run, t.remark or ""))
def _match_category(self, key, t):
is_onetime = (t.remark or "") and "一次性" in t.remark
has_success = False
if is_onetime:
try:
hlist = self.controller.get_history_by_task_id(t.task_id, limit=100)
has_success = any(h.is_success for h in hlist)
except Exception:
pass
filters = {
"total": lambda: True,
"periodic": lambda: not is_onetime,
"onetime_done": lambda: is_onetime and has_success,
"onetime_undone": lambda: is_onetime and not has_success,
"enabled": lambda: t.is_enable,
"running": lambda: t.task_status == TaskStatus.RUNNING,
}
if key in filters:
return filters[key]()
return True
# ========= 操作按钮 =========
def _get_selected_task_ids(self):
sel = self.detail_tree.selection()
if not sel:
messagebox.showwarning("提示", "请先选中任务")
return []
return [int(iid) for iid in sel]
def _toggle_disable(self):
ids = self._get_selected_task_ids()
if not ids:
return
for tid in ids:
try:
self.controller.update_task_enable(tid, False)
except Exception:
pass
self.refresh_dashboard()
if self._expanded_key:
self._populate_detail(self._expanded_key)
def _toggle_enable(self):
ids = self._get_selected_task_ids()
if not ids:
return
for tid in ids:
try:
self.controller.update_task_enable(tid, True)
except Exception:
pass
self.refresh_dashboard()
if self._expanded_key:
self._populate_detail(self._expanded_key)
def _execute_selected(self):
ids = self._get_selected_task_ids()
if not ids:
return
import threading
def do_exec(tid):
@GlobalExceptionCatcher.thread_catch_exc
def _run():
task_entity = self.controller.get_task_by_id(tid)
if task_entity:
self.app_service.execute_task_background(task_entity.task_id, task_entity)
t = threading.Thread(target=_run, daemon=True)
t.start()
for tid in ids:
do_exec(tid)
messagebox.showinfo("提示", f"已提交 {len(ids)} 个任务后台执行")
self.refresh_dashboard()
def refresh_dashboard(self):
"""刷新仪表盘全部数据"""
task_list = self.controller.get_all_task()
stat = self.controller.get_statistics()
recent_history = self.controller.get_global_history(limit=50)
self._task_cache = task_list
self._history_cache = recent_history
total_task = len(task_list)
enable_count = sum(1 for t in task_list if t.is_enable)
running_count = sum(1 for t in task_list if t.task_status == TaskStatus.RUNNING)
onetime_tasks = [t for t in task_list if (t.remark or "") and "一次性" in t.remark]
periodic_count = total_task - len(onetime_tasks)
onetime_done = 0
onetime_undone = 0
for t in onetime_tasks:
try:
hlist = self.controller.get_history_by_task_id(t.task_id, limit=100)
if any(h.is_success for h in hlist):
onetime_done += 1
else:
onetime_undone += 1
except Exception:
onetime_undone += 1
self.var_total_task.set(str(total_task))
self.var_periodic_task.set(str(periodic_count))
self.var_onetime_done.set(str(onetime_done))
self.var_onetime_undone.set(str(onetime_undone))
self.var_enable_task.set(str(enable_count))
self.var_running_task.set(str(running_count))
self.var_total_exec.set(str(stat["total_history"]))
self.var_success_exec.set(str(stat["success"]))
self.var_fail_exec.set(str(stat["fail"]))
# 刷新最近执行表格
for item in self.tree.get_children():
self.tree.delete(item)
for h in recent_history:
task_obj = next((t for t in task_list if t.task_id == h.task_id), None)
task_name = task_obj.task_name if task_obj else "未知任务"
res_text = "成功" if h.is_success else "失败"
time_str = h.execute_start_time.strftime("%Y-%m-%d %H:%M:%S") if h.execute_start_time else ""
self.tree.insert("", tk.END, values=(h.task_id, task_name, time_str, res_text))
# 如果有展开的分类，刷新明细
if self._expanded_key:
self._populate_detail(self._expanded_key)
def start_auto_refresh(self):
"""启动定时自动刷新仪表盘"""
self._auto_refresh_running = True
try:
self._refresh_ms = int(g_app_settings.ui.dashboard_refresh_ms)
except Exception:
self._refresh_ms = 5000
self._frame.after(self._refresh_ms, self._auto_refresh)
def _auto_refresh(self):
"""周期刷新：仅在窗口还在且未停止时继续"""
if not getattr(self, "_auto_refresh_running", False):
return
try:
self.refresh_dashboard()
finally:
try:
if self._frame.winfo_exists():
self._frame.after(self._refresh_ms, self._auto_refresh)
except Exception:
pass
def stop_auto_refresh(self):
"""停止自动刷新"""
self._auto_refresh_running = False
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 12:41
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : task_controller.py
"""
task_controller.py
表现层控制器，MVC的Controller
隔离UI界面与Application应用服务、Repository
UI所有操作全部调用Controller，不直接操作db、service
"""
from typing import Optional, List
from infrastructure.db_session import get_session
from infrastructure.task_repository import TaskRepository
from infrastructure.task_history_repository import TaskHistoryRepository
from application.task_scheduler_app_service import TaskSchedulerAppService
from domain.task_entity import TaskEntity
from domain.task_history_entity import TaskHistoryEntity
class TaskController:
def __init__(self):
self.task_repo = TaskRepository()
self.history_repo = TaskHistoryRepository()
self.app_service = TaskSchedulerAppService()
def get_all_task(self) -> List[TaskEntity]:
"""获取全部任务列表"""
gen = get_session()
db = next(gen)
res = self.task_repo.get_all(db)
db.close()
return res
def get_task_by_id(self, task_id: int) -> Optional[TaskEntity]:
"""按id获取单个任务"""
gen = get_session()
db = next(gen)
entity = self.task_repo.get_by_id(db, task_id)
db.close()
return entity
def save_task(self, entity: TaskEntity):
"""新增或者保存编辑任务"""
gen = get_session()
db = next(gen)
if entity.task_id == 0:
self.task_repo.save(db, entity)
else:
self.task_repo.save(db, entity)
db.commit()
db.close()
def delete_task(self, task_id: int):
"""删除任务"""
gen = get_session()
db = next(gen)
self.task_repo.delete_by_id(db, task_id)
db.commit()
db.close()
def update_task_enable(self, task_id: int, is_enable: bool):
"""启用/禁用任务"""
gen = get_session()
db = next(gen)
self.task_repo.update_enable(db, task_id, is_enable)
db.commit()
db.close()
def manual_run_task(self, task_id: int):
"""手动触发任务执行"""
self.app_service.manual_execute_task(task_id)
def get_task_history(self, task_id: int, limit: int = 200) -> List[TaskHistoryEntity]:
"""获取指定任务执行历史"""
gen = get_session()
db = next(gen)
lst = self.history_repo.list_by_taskid(db, task_id, limit)
db.close()
return lst
def get_global_history(self, limit: int = 100):
"""全局历史记录"""
gen = get_session()
db = next(gen)
lst = self.history_repo.list_limit(db, limit)
db.close()
return lst
def get_statistics(self):
"""获取统计数据：总任务、成功数、失败数"""
gen = get_session()
db = next(gen)
total = db.query(self.history_repo.model).count()
success = db.query(self.history_repo.model).filter(self.history_repo.model.is_success == True).count()
fail = db.query(self.history_repo.model).filter(self.history_repo.model.is_success == False).count()
db.close()
return {
"total_history": total,
"success": success,
"fail": fail
}
def get_recent_history(self, limit: int):
gen = get_session()
db = next(gen)
rows = self.history_repo.query_recent(db, limit)
db.close()
return rows
def get_history_by_id(self, history_id: int):
gen = get_session()
db = next(gen)
ent = self.history_repo.get_by_id(db, history_id)
db.close()
return ent
# -----历史查询方法，供仪表盘/UI调用-----
def get_history_by_task_id(self, task_id:int, limit:int=100):
gen = get_session()
db = next(gen)
try:
return self.history_repo.query_by_task_id(db, task_id, limit)
finally:
db.close()
def get_all_task(self):
gen = get_session()
db = next(gen)
try:
return self.task_repo.get_all(db)
finally:
db.close()
def get_task_by_id(self, tid:int):
gen = get_session()
db = next(gen)
try:
return self.task_repo.get_by_id(db, tid)
finally:
db.close()
def save_task(self, entity:TaskEntity):
gen = get_session()
db = next(gen)
try:
return self.task_repo.save(db, entity)
finally:
db.close()
def delete_task(self, tid:int):
gen = get_session()
db = next(gen)
try:
self.task_repo.delete_by_id(db, tid)
finally:
db.close()
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 16:24
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : task_edit_dialog.py
"""
task_edit_dialog.py
任务新增/编辑弹窗窗体
新建：传入task_entity=None；编辑：传入已有TaskEntity实例
"""
import tkinter as tk
from tkinter import ttk, messagebox
from domain.task_entity import TaskEntity
from domain.scheduler_enum import TaskStatus
class TaskEditDialog(tk.Toplevel):
def __init__(self, parent, task_entity: TaskEntity | None = None):
super().__init__(parent)
self.parent = parent
self.task_entity = task_entity
self.result_entity: TaskEntity | None = None
self.title("编辑任务" if task_entity else "新增任务")
self.geometry("520x420")
self.resizable(False, False)
# 模态弹窗
self.transient(parent)
self.grab_set()
self._build_ui()
# 编辑模式回显数据
if self.task_entity is not None:
self._fill_data()
# 窗口居中
self.center_window()
def _build_ui(self):
main_frame = ttk.Frame(self, padding=12)
main_frame.pack(fill=tk.BOTH, expand=True)
row_idx = 0
# 任务名称
ttk.Label(main_frame, text="任务名称：").grid(row=row_idx, column=0, sticky="w", pady=4)
self.var_task_name = tk.StringVar()
ttk.Entry(main_frame, textvariable=self.var_task_name, width=45).grid(row=row_idx, column=1, padx=8, pady=4)
row_idx +=1
# Cron表达式
ttk.Label(main_frame, text="Cron表达式：").grid(row=row_idx, column=0, sticky="w", pady=4)
self.var_cron = tk.StringVar()
ttk.Entry(main_frame, textvariable=self.var_cron, width=45).grid(row=row_idx, column=1, padx=8, pady=4)
row_idx +=1
# 执行命令
ttk.Label(main_frame, text="执行命令：").grid(row=row_idx, column=0, sticky="nw", pady=4)
self.txt_command = tk.Text(main_frame, width=42, height=6)
self.txt_command.grid(row=row_idx, column=1, padx=8, pady=4)
row_idx +=1
# 超时秒
ttk.Label(main_frame, text="超时(秒)：").grid(row=row_idx, column=0, sticky="w", pady=4)
self.var_timeout = tk.StringVar(value="300")
ttk.Entry(main_frame, textvariable=self.var_timeout, width=12).grid(row=row_idx, column=1, sticky="w", padx=8, pady=4)
row_idx +=1
# 最大重试
ttk.Label(main_frame, text="最大重试次数：").grid(row=row_idx, column=0, sticky="w", pady=4)
self.var_max_retry = tk.StringVar(value="0")
ttk.Entry(main_frame, textvariable=self.var_max_retry, width=12).grid(row=row_idx, column=1, sticky="w", padx=8, pady=4)
row_idx +=1
# 重试间隔
ttk.Label(main_frame, text="重试间隔(秒)：").grid(row=row_idx, column=0, sticky="w", pady=4)
self.var_retry_gap = tk.StringVar(value="10")
ttk.Entry(main_frame, textvariable=self.var_retry_gap, width=12).grid(row=row_idx, column=1, sticky="w", padx=8, pady=4)
row_idx +=1
# 是否启用
self.var_enable = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="启用任务", variable=self.var_enable).grid(row=row_idx, column=0, columnspan=2, sticky="w", pady=4)
row_idx +=1
# 备注
ttk.Label(main_frame, text="备注：").grid(row=row_idx, column=0, sticky="nw", pady=4)
self.txt_remark = tk.Text(main_frame, width=42, height=3)
self.txt_remark.grid(row=row_idx, column=1, padx=8, pady=4)
row_idx +=1
# 按钮区
btn_frame = ttk.Frame(main_frame)
btn_frame.grid(row=row_idx, column=0, columnspan=2, pady=12)
ttk.Button(btn_frame, text="确定", command=self.on_ok).pack(side=tk.LEFT, padx=8)
ttk.Button(btn_frame, text="取消", command=self.on_cancel).pack(side=tk.LEFT, padx=8)
def _fill_data(self):
"""编辑模式，回显已有任务实体数据"""
e = self.task_entity
self.var_task_name.set(e.task_name)
self.var_cron.set(e.cron_expr)
self.txt_command.insert(tk.END, e.command)
self.var_timeout.set(str(e.timeout_sec))
self.var_max_retry.set(str(e.max_retry))
self.var_retry_gap.set(str(e.retry_gap_sec))
self.var_enable.set(e.is_enable)
self.txt_remark.insert(tk.END, e.remark)
def on_ok(self):
"""保存校验，组装TaskEntity"""
task_name = self.var_task_name.get().strip()
cron_expr = self.var_cron.get().strip()
command = self.txt_command.get("1.0", tk.END).strip()
remark = self.txt_remark.get("1.0", tk.END).strip()
if not task_name:
messagebox.showwarning("校验", "任务名称不能为空")
return
if not cron_expr:
messagebox.showwarning("校验", "Cron表达式不能为空")
return
if not command:
messagebox.showwarning("校验", "执行命令不能为空")
return
try:
timeout_sec = int(self.var_timeout.get())
max_retry = int(self.var_max_retry.get())
retry_gap_sec = int(self.var_retry_gap.get())
except ValueError:
messagebox.showerror("校验", "超时、重试次数、重试间隔必须是数字")
return
task_id = 0
status = TaskStatus.IDLE
if self.task_entity is not None:
task_id = self.task_entity.task_id
status = self.task_entity.task_status
self.result_entity = TaskEntity(
task_id=task_id,
task_name=task_name,
cron_expr=cron_expr,
command=command,
timeout_sec=timeout_sec,
max_retry=max_retry,
retry_gap_sec=retry_gap_sec,
is_enable=self.var_enable.get(),
remark=remark,
task_status=status
)
self.destroy()
def on_cancel(self):
self.result_entity = None
self.destroy()
def center_window(self):
self.update_idletasks()
w = self.winfo_width()
h = self.winfo_height()
x = self.parent.winfo_rootx() + (self.parent.winfo_width() - w) // 2
y = self.parent.winfo_rooty() + (self.parent.winfo_height() - h) // 2
self.geometry(f"{w}x{h}+{x}+{y}")
@staticmethod
def show_dialog(parent, task_entity=None) -> TaskEntity | None:
"""静态调用入口，弹窗并返回结果，None代表取消"""
dlg = TaskEditDialog(parent, task_entity)
parent.wait_window(dlg)
return dlg.result_entity
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 12:54
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : task_form_dialog.py
"""
task_form_dialog.py
新增、编辑任务弹窗对话框
复用同一个Dialog，传入task实体为编辑，None为新增
字段：任务名称、cron表达式、执行命令、超时秒、最大重试次数、重试间隔、是否启用、备注
做cron表达式合法性校验
"""
import tkinter as tk
from tkinter import ttk, messagebox
import ttkbootstrap as bs
from ttkbootstrap.dialogs import Dialog
from croniter import croniter, CroniterBadCronError
from typing import Optional
from domain.task_entity import TaskEntity
from domain.scheduler_enum import TaskStatus
class TaskFormDialog(Dialog):
def __init__(self, parent, task_entity: Optional[TaskEntity] = None):
"""
:param parent: 父窗口
:param task_entity: 编辑传入TaskEntity，新增传None
"""
self.parent = parent
self.edit_entity: Optional[TaskEntity] = task_entity
self.result_entity: Optional[TaskEntity] = None
# 表单变量
self.var_task_name = tk.StringVar()
self.var_cron_expr = tk.StringVar()
self.var_command = tk.StringVar()
self.var_timeout = tk.StringVar(value="300")
self.var_max_retry = tk.StringVar(value="0")
self.var_retry_gap = tk.StringVar(value="10")
self.var_is_enable = tk.BooleanVar(value=True)
self.var_remark = tk.StringVar()
super().__init__(
title="编辑任务" if self.edit_entity else "新增任务",
parent=parent,
alert=False
)
def create_body(self, master):
"""构建对话框表单UI"""
master.columnconfigure(1, weight=1)
row_idx = 0
# 任务名称
ttk.Label(master, text="任务名称：").grid(row=row_idx, column=0, sticky="w", padx=5, pady=4)
e_name = ttk.Entry(master, textvariable=self.var_task_name)
e_name.grid(row=row_idx, column=1, sticky="ew", padx=5, pady=4)
row_idx +=1
# Cron表达式
ttk.Label(master, text="Cron表达式：").grid(row=row_idx, column=0, sticky="w", padx=5, pady=4)
e_cron = ttk.Entry(master, textvariable=self.var_cron_expr)
e_cron.grid(row=row_idx, column=1, sticky="ew", padx=5, pady=4)
tip_cron = ttk.Label(master, text="示例：*/5 * * * * 每5分钟", foreground="#666666")
tip_cron.grid(row=row_idx, column=2, sticky="w", padx=3)
row_idx +=1
# 执行命令
ttk.Label(master, text="执行命令：").grid(row=row_idx, column=0, sticky="w", padx=5, pady=4)
e_cmd = ttk.Entry(master, textvariable=self.var_command)
e_cmd.grid(row=row_idx, column=1, columnspan=2, sticky="ew", padx=5, pady=4)
row_idx +=1
# 超时秒
ttk.Label(master, text="超时(秒)：").grid(row=row_idx, column=0, sticky="w", padx=5, pady=4)
e_timeout = ttk.Entry(master, textvariable=self.var_timeout, width=12)
e_timeout.grid(row=row_idx, column=1, sticky="w", padx=5, pady=4)
row_idx +=1
# 最大重试次数
ttk.Label(master, text="最大重试次数：").grid(row=row_idx, column=0, sticky="w", padx=5, pady=4)
e_retry = ttk.Entry(master, textvariable=self.var_max_retry, width=12)
e_retry.grid(row=row_idx, column=1, sticky="w", padx=5, pady=4)
row_idx +=1
# 重试间隔
ttk.Label(master, text="重试间隔(秒)：").grid(row=row_idx, column=0, sticky="w", padx=5, pady=4)
e_gap = ttk.Entry(master, textvariable=self.var_retry_gap, width=12)
e_gap.grid(row=row_idx, column=1, sticky="w", padx=5, pady=4)
row_idx +=1
# 是否启用
chk_enable = ttk.Checkbutton(master, text="启用任务", variable=self.var_is_enable)
chk_enable.grid(row=row_idx, column=0, columnspan=2, sticky="w", padx=5, pady=6)
row_idx +=1
# 备注
ttk.Label(master, text="备注：").grid(row=row_idx, column=0, sticky="nw", padx=5, pady=4)
txt_remark = bs.Text(master, height=4)
txt_remark.grid(row=row_idx, column=1, columnspan=2, sticky="ew", padx=5, pady=4)
self.txt_remark_widget = txt_remark
# 如果是编辑模式，回填原有数据
if self.edit_entity is not None:
self.var_task_name.set(self.edit_entity.task_name)
self.var_cron_expr.set(self.edit_entity.cron_expr)
self.var_command.set(self.edit_entity.command)
self.var_timeout.set(str(self.edit_entity.timeout_sec))
self.var_max_retry.set(str(self.edit_entity.max_retry))
self.var_retry_gap.set(str(self.edit_entity.retry_gap_sec))
self.var_is_enable.set(self.edit_entity.is_enable)
txt_remark.insert(tk.END, self.edit_entity.remark or "")
def create_buttonbox(self, master):
"""底部按钮：确定、取消"""
super().create_buttonbox(master)
def validate(self):
"""表单校验：cron合法性、数字字段、非空校验"""
task_name = self.var_task_name.get().strip()
cron_expr = self.var_cron_expr.get().strip()
cmd = self.var_command.get().strip()
if not task_name:
messagebox.showwarning("校验提示", "任务名称不能为空")
return False
if not cron_expr:
messagebox.showwarning("校验提示", "Cron表达式不能为空")
return False
# croniter校验表达式
try:
croniter(cron_expr)
except CroniterBadCronError:
messagebox.showerror("Cron错误", f"Cron表达式格式非法：{cron_expr}")
return False
if not cmd:
messagebox.showwarning("校验提示", "执行命令不能为空")
return False
# 数字校验
try:
int(self.var_timeout.get())
int(self.var_max_retry.get())
int(self.var_retry_gap.get())
except ValueError:
messagebox.showerror("输入错误", "超时、重试次数、重试间隔必须为整数")
return False
return True
def apply(self):
"""点击确定，组装结果实体"""
if not self.validate():
return
remark = self.txt_remark_widget.get("1.0", tk.END).strip()
if self.edit_entity:
# 编辑，复制原有ID
entity = TaskEntity(
task_id=self.edit_entity.task_id,
task_name=self.var_task_name.get().strip(),
cron_expr=self.var_cron_expr.get().strip(),
command=self.var_command.get().strip(),
timeout_sec=int(self.var_timeout.get()),
max_retry=int(self.var_max_retry.get()),
retry_gap_sec=int(self.var_retry_gap.get()),
is_enable=self.var_is_enable.get(),
remark=remark,
task_status=self.edit_entity.task_status
)
else:
# 新增，task_id=0数据库自增
entity = TaskEntity(
task_id=0,
task_name=self.var_task_name.get().strip(),
cron_expr=self.var_cron_expr.get().strip(),
command=self.var_command.get().strip(),
timeout_sec=int(self.var_timeout.get()),
max_retry=int(self.var_max_retry.get()),
retry_gap_sec=int(self.var_retry_gap.get()),
is_enable=self.var_is_enable.get(),
remark=remark,
task_status=TaskStatus.IDLE
)
self.result_entity = entity
def show(self):
"""弹窗调用入口，返回TaskEntity，取消返回None"""
super().show()
return self.result_entity
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 16:06
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : task_history_detail_dialog.py
"""
task_history_detail_dialog.py
任务执行历史详情弹窗：查看某次执行完整输出日志
"""
import tkinter as tk
import ttkbootstrap as ttk
from domain.task_history_entity import TaskHistoryEntity
class TaskHistoryDetailDialog:
def __init__(self, parent, history_entity: TaskHistoryEntity):
"""
:param parent: 父窗口
:param history_entity: 任务历史领域实体
"""
self.parent = parent
self.his = history_entity
dlg = tk.Toplevel(parent)
dlg.title(f"执行历史详情｜ID:{self.his.history_id} 任务ID:{self.his.task_id}")
dlg.geometry("800x550")
dlg.transient(parent)
dlg.grab_set()
main_frm = ttk.Frame(dlg, padding=10)
main_frm.pack(fill="both", expand=True)
info_frm = ttk.Frame(main_frm)
info_frm.pack(fill="x", pady=(0,8))
ttk.Label(info_frm, text=f"执行时间：{self.his.execute_start_time}").pack(anchor="w")
ttk.Label(info_frm, text=f"执行结果：{'✅成功' if self.his.is_success else '❌失败'}").pack(anchor="w")
ttk.Label(main_frm, text="完整输出日志：").pack(anchor="w")
txt_frame = ttk.Frame(main_frm)
txt_frame.pack(fill="both", expand=True)
self.txt_log = tk.Text(txt_frame, wrap=tk.CHAR)
vsb = ttk.Scrollbar(txt_frame, orient="vertical", command=self.txt_log.yview)
hsb = ttk.Scrollbar(txt_frame, orient="horizontal", command=self.txt_log.xview)
self.txt_log.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set)
self.txt_log.grid(row=0, column=0, sticky="nsew")
vsb.grid(row=0, column=1, sticky="ns")
hsb.grid(row=1, column=0, sticky="ew")
txt_frame.grid_rowconfigure(0, weight=1)
txt_frame.grid_columnconfigure(0, weight=1)
self.txt_log.insert(tk.END, self.his.execute_msg if self.his.execute_msg else "(无输出)")
self.txt_log.config(state=tk.DISABLED)
btn_frm = ttk.Frame(main_frm)
btn_frm.pack(pady=8)
ttk.Button(btn_frm, text="关闭", command=dlg.destroy).pack()
dlg.wait_window()
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 12:55
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : task_history_dialog.py
"""
task_history_dialog.py
查看单个任务执行历史弹窗
展示历史列表，点击行查看执行详情
"""
import tkinter as tk
from tkinter import ttk
import ttkbootstrap as bs
from ttkbootstrap.dialogs import Dialog
from typing import List
from domain.task_history_entity import TaskHistoryEntity
class TaskHistoryDialog(Dialog):
def __init__(self, parent, history_list: List[TaskHistoryEntity], task_name: str):
self.parent = parent
self.history_list = history_list
self.task_name = task_name
super().__init__(
title=f"任务历史 - {task_name}",
parent=parent,
alert=False
)
def create_body(self, master):
master.rowconfigure(0, weight=1)
master.columnconfigure(0, weight=1)
# 构建Treeview表格
cols = ("history_id", "start_time", "is_success")
self.tree = ttk.Treeview(master, columns=cols, show="headings")
self.tree.heading("history_id", text="历史ID")
self.tree.heading("start_time", text="执行开始时间")
self.tree.heading("is_success", text="执行结果")
self.tree.column("history_id", width=80, anchor="center")
self.tree.column("start_time", width=220, anchor="center")
self.tree.column("is_success", width=100, anchor="center")
vsb = ttk.Scrollbar(master, orient="vertical", command=self.tree.yview)
self.tree.configure(yscrollcommand=vsb.set)
self.tree.grid(row=0, column=0, sticky="nsew")
vsb.grid(row=0, column=1, sticky="ns")
# 详情文本框
self.detail_text = bs.Text(master, height=8)
self.detail_text.grid(row=1, column=0, columnspan=2, sticky="ew", pady=(6,0))
# 填充数据
for h in self.history_list:
success_text = "成功" if h.is_success else "失败"
time_str = h.execute_start_time.strftime("%Y-%m-%d %H:%M:%S") if h.execute_start_time else ""
self.tree.insert("", tk.END, iid=str(h.history_id), values=(h.history_id, time_str, success_text))
# 绑定选中事件，显示详情
self.tree.bind("<<TreeviewSelect>>", self.on_select_row)
def on_select_row(self, event):
self.detail_text.delete("1.0", tk.END)
sel = self.tree.selection()
if not sel:
return
hid = int(sel[0])
find_hist = next((x for x in self.history_list if x.history_id == hid), None)
if find_hist:
content = f"执行结果：{'成功' if find_hist.is_success else '失败'}\n输出信息：\n{find_hist.execute_msg}"
self.detail_text.insert(tk.END, content)
def create_buttonbox(self, master):
btn_close = bs.Button(master, text="关闭", command=self._close, bootstyle="secondary")
btn_close.pack(pady=6)
def apply(self):
pass
def show(self):
super().show()
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 12:41
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : task_view.py
"""
presentation/task_view.py 修改片段
工具栏增加：立即执行按钮
"""
import tkinter as tk
import ttkbootstrap as ttk
from tkinter import filedialog, messagebox
from presentation.task_controller import TaskController
from presentation.task_edit_dialog import TaskEditDialog
from infrastructure.json_import_export import TaskJsonImportExport
from application.task_scheduler_app_service import TaskSchedulerAppService
from infrastructure.global_exception_catcher import GlobalExceptionCatcher
from config.app_settings import g_app_settings
class TaskMainView:
def __init__(self, parent_frame):
self.parent = parent_frame
self.controller = TaskController()
self.imp_exp = TaskJsonImportExport()
self.app_service = TaskSchedulerAppService()
# 工具栏
tool_frame = ttk.Frame(self.parent)
tool_frame.pack(fill="x", padx=4, pady=4)
ttk.Button(tool_frame, text="新增任务", command=self.on_add).pack(side="left", padx=2)
ttk.Button(tool_frame, text="编辑任务", command=self.on_edit).pack(side="left", padx=2)
ttk.Button(tool_frame, text="删除任务", command=self.on_delete).pack(side="left", padx=2)
# =========新增手动立即执行按钮==========
ttk.Button(tool_frame, text="立即执行", command=self.on_run_now).pack(side="left", padx=2)
ttk.Separator(tool_frame, orient="vertical").pack(side="left", padx=6, fill="y")
ttk.Button(tool_frame, text="导出JSON", command=self.on_export_json).pack(side="left", padx=2)
ttk.Button(tool_frame, text="导入JSON", command=self.on_import_json).pack(side="left", padx=2)
tree_frame = ttk.Frame(self.parent)
tree_frame.pack(fill="both", expand=True, padx=4, pady=4)
columns = ("task_id", "task_name", "cron_expr", "command", "is_enable")
self.tree = ttk.Treeview(tree_frame, columns=columns, show="headings")
self.tree.heading("task_id", text="ID")
self.tree.heading("task_name", text="任务名称")
self.tree.heading("cron_expr", text="Cron")
self.tree.heading("command", text="命令")
self.tree.heading("is_enable", text="启用")
self.tree.column("task_id", width=60)
self.tree.column("task_name", width=160)
self.tree.column("cron_expr", width=140)
self.tree.column("command", width=260)
self.tree.column("is_enable", width=70)
vsb = ttk.Scrollbar(tree_frame, orient="vertical", command=self.tree.yview)
self.tree.configure(yscrollcommand=vsb.set)
self.tree.pack(side="left", fill="both", expand=True)
vsb.pack(side="right", fill="y")
self.tree.bind("<Double-1>", self.on_tree_double_click)
self.refresh_data()
self.start_auto_refresh()
def refresh_data(self):
for item in self.tree.get_children():
self.tree.delete(item)
task_list = self.controller.get_all_task()
for t in task_list:
self.tree.insert("", tk.END, values=(
t.task_id,
t.task_name,
t.cron_expr,
t.command,
"是" if t.is_enable else "否"
))
def on_add(self):
dlg = TaskEditDialog(self.parent.winfo_toplevel(), None)
entity = dlg.result_entity
if entity is not None:
self.controller.save_task(entity)
self.refresh_data()
def on_edit(self):
sel = self.tree.selection()
if not sel:
messagebox.showwarning("提示", "请先选中一行任务")
return
item_id = sel[0]
vals = self.tree.item(item_id, "values")
tid = int(vals[0])
task = self.controller.get_task_by_id(tid)
if not task:
return
dlg = TaskEditDialog(self.parent.winfo_toplevel(), task)
entity = dlg.result_entity
if entity is not None:
self.controller.save_task(entity)
self.refresh_data()
def on_delete(self):
sel = self.tree.selection()
if not sel:
messagebox.showwarning("提示", "请先选中一行任务")
return
item_id = sel[0]
vals = self.tree.item(item_id, "values")
tid = int(vals[0])
ok = messagebox.askyesno("确认删除", f"确定删除任务 id={tid}？")
if ok:
self.controller.delete_task(tid)
self.refresh_data()
def on_tree_double_click(self, event):
sel = self.tree.selection()
if not sel:
return
item_id = sel[0]
vals = self.tree.item(item_id, "values")
tid = int(vals[0])
task = self.controller.get_task_by_id(tid)
if not task:
return
win = tk.Toplevel(self.parent.winfo_toplevel())
win.title(f"任务详情 ID:{tid} {task.task_name}")
win.geometry("760x460")
win.transient(self.parent.winfo_toplevel())
frm = ttk.Frame(win, padding=8)
frm.pack(fill="both", expand=True)
info = [
("任务名称", task.task_name),
("Cron表达式", task.cron_expr),
("是否启用", "是" if task.is_enable else "否"),
("任务状态", task.task_status.name),
("超时(秒)", str(task.timeout_sec)),
("最大重试次数", str(task.max_retry)),
("重试间隔(秒)", str(task.retry_gap_sec)),
("任务ID", str(task.task_id)),
]
for label, value in info:
row = ttk.Frame(frm)
row.pack(fill="x", pady=1)
ttk.Label(row, text=f"{label}：", width=14, anchor="e").pack(side="left")
ttk.Label(row, text=value, anchor="w").pack(side="left")
ttk.Label(frm, text="执行命令：").pack(anchor="w", pady=(8, 2))
txt = ttk.Text(frm)
txt.pack(fill="both", expand=True)
txt.insert(tk.END, task.command)
txt.config(state=tk.DISABLED)
ttk.Label(frm, text="备注：").pack(anchor="w", pady=(8, 2))
txt_remark = ttk.Text(frm)
txt_remark.pack(fill="both", expand=True)
txt_remark.insert(tk.END, task.remark or "")
txt_remark.config(state=tk.DISABLED)
ttk.Button(frm, text="关闭", command=win.destroy).pack(pady=6)
# =========手动立即执行任务==========
def on_run_now(self):
import threading
sel = self.tree.selection()
if not sel:
messagebox.showwarning("提示", "请先选中任务")
return
item_id = sel[0]
vals = self.tree.item(item_id, "values")
tid = int(vals[0])
task_entity = self.controller.get_task_by_id(tid)
if not task_entity:
messagebox.showerror("错误", "找不到该任务")
return
@GlobalExceptionCatcher.thread_catch_exc
def run_task_bg():
"""后台线程执行任务，不阻塞UI"""
self.app_service.execute_single_task(task_entity)
t = threading.Thread(target=run_task_bg, daemon=True)
t.start()
messagebox.showinfo("提示", f"任务ID:{tid}已提交后台执行，请查看日志面板")
def on_export_json(self):
from tkinter import filedialog
path = filedialog.asksaveasfilename(
defaultextension=".json",
filetypes=[("JSON文件", "*.json"), ("全部文件", "*.*")],
title="导出任务到JSON"
)
if not path:
return
try:
self.imp_exp.export_all_to_json(path)
messagebox.showinfo("成功", f"导出完成：{path}")
except Exception as e:
messagebox.showerror("导出失败", str(e))
def on_import_json(self):
from tkinter import filedialog
path = filedialog.askopenfilename(
filetypes=[("JSON文件", "*.json"), ("全部文件", "*.*")],
title="从JSON导入任务"
)
if not path:
return
try:
stat = self.imp_exp.import_from_json(path)
msg = f"导入完成\n总条数:{stat['total']}\n新增:{stat['insert']}\n更新:{stat['update']}\n跳过:{stat['skip']}"
messagebox.showinfo("导入结果", msg)
self.refresh_data()
except Exception as e:
messagebox.showerror("导入失败", str(e))
def start_auto_refresh(self):
"""启动定时自动刷新任务列表"""
self._auto_refresh_running = True
try:
self._refresh_ms = int(g_app_settings.ui.dashboard_refresh_ms)
except Exception:
self._refresh_ms = 5000
self.parent.after(self._refresh_ms, self._auto_refresh)
def _auto_refresh(self):
"""周期刷新：仅在窗口还在且未停止时继续"""
if not getattr(self, "_auto_refresh_running", False):
return
try:
self.refresh_data()
finally:
try:
if self.parent.winfo_exists():
self.parent.after(self._refresh_ms, self._auto_refresh)
except Exception:
pass
def stop_auto_refresh(self):
"""停止自动刷新"""
self._auto_refresh_running = False
调用：
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# Author : geovindu,Geovin Du 涂聚文.
# IDE : PyCharm 2024.3.6 python 3.11
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/9/8 12:19
# User : geovindu
# Product : PyCharm
# Project : PyTaskScheduler
# File : main.py
"""
main.py 任务调度系统主入口
DDD分层架构：domain领域 / infrastructure基础设施 / application应用服务 / presentation界面 / api接口
启动流程：
1.加载yaml配置
2.初始化数据库，自动创建全部数据表
3.初始化日志系统，绑定GUI日志处理器
4.启动后台调度线程
5.启动数据库备份清理线程
6.可选：启动FastAPI后台接口服务
7.初始化tkbootstrap主GUI窗口
8.初始化系统托盘，设置Windows开机自启
"""
"""
main.py Notebook日志面板改造，增加清空日志按钮
"""
import tkinter as tk
import ttkbootstrap as bs
import ttkbootstrap as ttk
from PIL import Image
import logging
from logging.handlers import RotatingFileHandler
from config.app_settings import g_app_settings
from presentation.task_view import TaskMainView
from presentation.dashboard_view import DashboardView
from infrastructure.scheduler_background_thread import BackgroundSchedulerThread
from infrastructure.gui_log_handler import GuiLogHandler
from infrastructure.system_tray_helper import SystemTrayHelper
from infrastructure.quartz_backup_scheduler import BackupQuartzScheduler
from infrastructure.global_exception_catcher import GlobalExceptionCatcher
import os
def setup_logging(gui_handler: GuiLogHandler):
root_logger = logging.getLogger()
root_logger.setLevel(g_app_settings.app.log_level)
root_logger.handlers.clear()
fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
console_handler = logging.StreamHandler()
console_handler.setFormatter(fmt)
root_logger.addHandler(console_handler)
log_dir = g_app_settings.app.log_dir
if not os.path.exists(log_dir):
os.makedirs(log_dir, exist_ok=True)
log_file = os.path.join(log_dir, "app.log")
file_handler = RotatingFileHandler(
log_file,
maxBytes=5*1024*1024,
backupCount=10,
encoding="utf-8"
)
file_handler.setFormatter(fmt)
root_logger.addHandler(file_handler)
root_logger.addHandler(gui_handler)
def main():
root = bs.Window(themename="litera")
root.title(g_app_settings.app.app_name)
root.geometry("1100x720")
exc_catcher = GlobalExceptionCatcher(root)
notebook = ttk.Notebook(root)
notebook.pack(fill="both", expand=True, padx=4, pady=4)
frame_task = ttk.Frame(notebook)
notebook.add(frame_task, text="任务列表")
frame_dashboard = ttk.Frame(notebook)
notebook.add(frame_dashboard, text="仪表盘")
frame_log = ttk.Frame(notebook)
notebook.add(frame_log, text="运行日志")
# =========日志面板：按钮 + 文本框==========
log_tool_frame = ttk.Frame(frame_log)
log_tool_frame.pack(fill="x", padx=4, pady=2)
log_text = bs.Text(frame_log)
log_text.pack(fill="both", expand=True, padx=4, pady=(0,4))
gui_log_handler = GuiLogHandler(log_text)
setup_logging(gui_log_handler)
# 清空日志按钮
def on_clear_log():
gui_log_handler.clear()
ttk.Button(log_tool_frame, text="清空日志面板", command=on_clear_log).pack(side="left")
task_main_view = TaskMainView(frame_task)
dash_view = DashboardView(frame_dashboard)
scheduler_thread = BackgroundSchedulerThread()
scheduler_thread.start()
backup_scheduler = BackupQuartzScheduler()
backup_scheduler.start()
tray_helper = None
if g_app_settings.tray.enable_tray:
ico_img = Image.new("RGB", (64, 64), color="#348ede")
tray_helper = SystemTrayHelper(root, ico_img)
tray_helper.setup_tray()
tray_helper.run_tray()
def on_close_window():
if g_app_settings.tray.start_minimize or g_app_settings.tray.enable_tray:
tray_helper.minimize_to_tray()
root.protocol("WM_DELETE_WINDOW", on_close_window)
def app_exit():
task_main_view.stop_auto_refresh()
dash_view.stop_auto_refresh()
scheduler_thread.stop()
backup_scheduler.stop()
root.destroy()
tray_helper.on_exit_callback = app_exit
else:
def on_quit():
task_main_view.stop_auto_refresh()
dash_view.stop_auto_refresh()
scheduler_thread.stop()
backup_scheduler.stop()
root.destroy()
root.protocol("WM_DELETE_WINDOW", on_quit)
root.mainloop()
task_main_view.stop_auto_refresh()
dash_view.stop_auto_refresh()
scheduler_thread.stop()
backup_scheduler.stop()
if name == "main":
main()
输出：
更多推荐
1
0- 0
已为社区贡献3条内容
所有评论(0)