RailWise KB已发布
自动化监测脚本模板
RailWise自动化监测脚本标准化代码模板，支持定时数据采集、自动处理、报告生成与预警推送的完整自动化工作流，实现无人值守的监测运维。
template-doc
自动化监测脚本模板
Section titled “自动化监测脚本模板”1. 模板概述
Section titled “1. 模板概述”1.1 用途
Section titled “1.1 用途”本模板提供自动化监测运维的完整脚本框架，支持：
- 定时采集：按设定频率自动采集监测数据
- 自动处理：数据清洗、计算、预警检测自动执行
- 报告生成：自动生成日报、周报、月报
- 预警推送：自动检测预警并推送通知
- 状态监控：系统健康检查、日志记录、异常恢复
- 数据归档：自动归档历史数据、清理过期数据
1.2 适用场景
Section titled “1.2 适用场景”| 场景 | 频率 | 触发条件 | 执行内容 |
|---|---|---|---|
| 自动化全站仪监测 | 1-4小时 | 定时 | 采集→处理→预警→报告 |
| 传感器数据接入 | 分钟级 | 实时/定时 | 采集→入库→检测 |
| 日终数据处理 | 每日 | 凌晨2:00 | 汇总→计算→归档 |
| 周报自动生成 | 每周 | 周一8:00 | 汇总→分析→报告 |
| 数据清理归档 | 每月 | 每月1日 | 归档→清理→备份 |
1.3 输入输出
Section titled “1.3 输入输出”- 输入：定时触发信号、配置文件、数据库连接
- 输出：监测数据、报告文件、通知消息、日志记录
- 中间产物：临时数据、处理状态、执行日志
2. 代码示例
Section titled “2. 代码示例”2.1 核心自动化引擎
Section titled “2.1 核心自动化引擎”"""RailWise 自动化监测脚本引擎支持定时任务、工作流编排、异常恢复与状态监控
作者: RailWise技术部版本: 1.0.0"""
import jsonimport loggingimport signalimport sysimport timeimport tracebackfrom abc import ABC, abstractmethodfrom dataclasses import dataclass, fieldfrom datetime import datetime, timedelta, timezonefrom enum import Enumfrom pathlib import Pathfrom typing import Any, Callable, Dict, List, Optional, Tuple, Union
import schedulefrom apscheduler.schedulers.background import BackgroundSchedulerfrom apscheduler.triggers.cron import CronTriggerfrom apscheduler.triggers.interval import IntervalTrigger
# 配置日志logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[ logging.StreamHandler(sys.stdout), logging.FileHandler("automation.log", encoding="utf-8") ])logger = logging.getLogger("RailWise.AutomationEngine")
class TaskStatus(Enum): """任务状态枚举""" PENDING = "pending" # 待执行 RUNNING = "running" # 执行中 SUCCESS = "success" # 成功 FAILED = "failed" # 失败 SKIPPED = "skipped" # 跳过 TIMEOUT = "timeout" # 超时
class TaskPriority(Enum): """任务优先级""" CRITICAL = 1 # 关键任务（如预警推送） HIGH = 2 # 高优先级（如数据采集） NORMAL = 3 # 普通优先级（如报告生成） LOW = 4 # 低优先级（如数据清理）
@dataclassclass TaskConfig: """任务配置""" task_id: str task_name: str task_type: str schedule: str # cron表达式或间隔描述 priority: TaskPriority = TaskPriority.NORMAL timeout_seconds: int = 3600 max_retries: int = 3 retry_delay_seconds: int = 300 enabled: bool = True dependencies: List[str] = field(default_factory=list) # 依赖任务ID parameters: Dict[str, Any] = field(default_factory=dict)
@dataclassclass TaskResult: """任务执行结果""" task_id: str status: TaskStatus start_time: Optional[datetime] = None end_time: Optional[datetime] = None duration_ms: float = 0.0 output: Any = None error_message: Optional[str] = None retry_count: int = 0
@dataclassclass AutomationConfig: """自动化配置""" project_id: str project_name: str log_dir: Path = field(default_factory=lambda: Path("./logs")) data_dir: Path = field(default_factory=lambda: Path("./data")) report_dir: Path = field(default_factory=lambda: Path("./reports")) archive_dir: Path = field(default_factory=lambda: Path("./archive")) enable_health_check: bool = True health_check_interval_seconds: int = 60 max_concurrent_tasks: int = 5 shutdown_timeout_seconds: int = 30
class AutomationTask(ABC): """自动化任务抽象基类"""
def __init__(self, config: TaskConfig): self.config = config self.logger = logging.getLogger(f"Task.{config.task_id}") self.current_result: Optional[TaskResult] = None
@abstractmethod def execute(self, **kwargs) -> Any: """执行任务""" pass
def pre_execute(self) -> bool: """执行前检查""" self.logger.info(f"任务 {self.config.task_id} 开始执行") return True
def post_execute(self, result: Any) -> Any: """执行后处理""" self.logger.info(f"任务 {self.config.task_id} 执行完成") return result
def on_error(self, error: Exception) -> None: """错误处理""" self.logger.error(f"任务 {self.config.task_id} 执行失败: {error}") self.logger.error(traceback.format_exc())
def run(self, **kwargs) -> TaskResult: """运行任务（包含重试逻辑）""" result = TaskResult(task_id=self.config.task_id, status=TaskStatus.PENDING)
for attempt in range(self.config.max_retries + 1): try: # 执行前检查 if not self.pre_execute(): result.status = TaskStatus.SKIPPED result.error_message = "执行前检查失败" return result
# 记录开始时间 result.start_time = datetime.now(timezone.utc) result.status = TaskStatus.RUNNING result.retry_count = attempt
# 执行任务 output = self.execute(**kwargs)
# 执行后处理 output = self.post_execute(output)
# 记录完成 result.end_time = datetime.now(timezone.utc) result.duration_ms = (result.end_time - result.start_time).total_seconds() * 1000 result.status = TaskStatus.SUCCESS result.output = output
self.logger.info(f"任务 {self.config.task_id} 成功完成，耗时: {result.duration_ms:.2f}ms") break
except Exception as e: self.on_error(e) result.error_message = str(e)
if attempt < self.config.max_retries: self.logger.warning(f"任务 {self.config.task_id} 第 {attempt + 1} 次失败，{self.config.retry_delay_seconds}秒后重试") time.sleep(self.config.retry_delay_seconds) else: result.status = TaskStatus.FAILED result.end_time = datetime.now(timezone.utc) self.logger.error(f"任务 {self.config.task_id} 最终失败")
self.current_result = result return result
class DataCollectionTask(AutomationTask): """数据采集任务"""
def __init__(self, config: TaskConfig, instrument_config: Dict[str, Any]): super().__init__(config) self.instrument_config = instrument_config
def execute(self, **kwargs) -> Any: """执行数据采集""" instrument_type = self.instrument_config.get("type", "total_station") instrument_id = self.instrument_config.get("id", "unknown")
self.logger.info(f"开始从 {instrument_id} ({instrument_type}) 采集数据")
# 模拟数据采集（实际项目中调用仪器SDK或API） collected_data = self._collect_from_instrument(instrument_type, instrument_id)
self.logger.info(f"采集完成: {len(collected_data)} 条记录") return collected_data
def _collect_from_instrument(self, instrument_type: str, instrument_id: str) -> List[Dict]: """从仪器采集数据""" # 实际项目中根据仪器类型调用不同的采集方法 # 这里模拟返回数据 import random from datetime import datetime
data = [] for i in range(5): # 模拟5个测点 data.append({ "record_id": f"{instrument_id}_{datetime.now().strftime('%Y%m%d%H%M%S')}_{i}", "instrument_id": instrument_id, "point_id": f"JC-{i+1:02d}", "observation_time": datetime.now(timezone.utc).isoformat(), "value_x": round(random.uniform(-0.5, 0.5), 2), "value_y": round(random.uniform(-0.5, 0.5), 2), "value_z": round(random.uniform(-0.1, 0.1), 2), })
return data
class DataProcessingTask(AutomationTask): """数据处理任务"""
def __init__(self, config: TaskConfig, processing_config: Dict[str, Any]): super().__init__(config) self.processing_config = processing_config
def execute(self, **kwargs) -> Any: """执行数据处理""" data = kwargs.get("data", [])
if not data: self.logger.warning("无数据需要处理") return []
self.logger.info(f"开始处理 {len(data)} 条记录")
# 1. 数据清洗 cleaned_data = self._clean_data(data)
# 2. 计算指标 calculated_data = self._calculate_metrics(cleaned_data)
# 3. 预警检测 alert_data = self._detect_alerts(calculated_data)
self.logger.info(f"处理完成: 清洗 {len(cleaned_data)} 条, 预警 {len([d for d in alert_data if d.get('alert_level') != '正常'])} 条")
return alert_data
def _clean_data(self, data: List[Dict]) -> List[Dict]: """清洗数据""" cleaned = [] for record in data: # 检查必要字段 if all(k in record for k in ["point_id", "observation_time"]): # 数值精度处理 for key in ["value_x", "value_y", "value_z"]: if key in record and record[key] is not None: record[key] = round(record[key], 2) cleaned.append(record)
return cleaned
def _calculate_metrics(self, data: List[Dict]) -> List[Dict]: """计算监测指标""" for record in data: # 计算累计位移（简化版） values = [record.get("value_x", 0), record.get("value_y", 0), record.get("value_z", 0)] record["value_d"] = round(sum(v**2 for v in values) ** 0.5, 2)
# 计算变化速率（需要历史数据，这里简化） record["change_rate"] = 0.0
return data
def _detect_alerts(self, data: List[Dict]) -> List[Dict]: """检测预警""" thresholds = self.processing_config.get("thresholds", {})
for record in data: value_d = record.get("value_d", 0)
# 判断预警等级 if value_d >= thresholds.get("red", 10): record["alert_level"] = "红色预警" elif value_d >= thresholds.get("orange", 8): record["alert_level"] = "橙色预警" elif value_d >= thresholds.get("yellow", 5): record["alert_level"] = "黄色预警" else: record["alert_level"] = "正常"
return data
class ReportGenerationTask(AutomationTask): """报告生成任务"""
def __init__(self, config: TaskConfig, report_config: Dict[str, Any]): super().__init__(config) self.report_config = report_config
def execute(self, **kwargs) -> Any: """执行报告生成""" report_type = self.report_config.get("type", "daily")
self.logger.info(f"开始生成 {report_type} 报告")
# 模拟报告生成 report_data = { "report_type": report_type, "report_date": datetime.now(timezone.utc).isoformat(), "project_id": self.report_config.get("project_id", "unknown"), "total_points": 45, "active_points": 43, "alert_count": 2, "summary": "监测数据正常，2个测点黄色预警" }
# 保存报告 report_path = self._save_report(report_data)
self.logger.info(f"报告已生成: {report_path}") return report_path
def _save_report(self, report_data: Dict) -> Path: """保存报告""" report_dir = Path(self.report_config.get("output_dir", "./reports")) report_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") report_path = report_dir / f"report_{report_data['report_type']}_{timestamp}.json"
with open(report_path, 'w', encoding='utf-8') as f: json.dump(report_data, f, ensure_ascii=False, indent=2)
return report_path
class AlertNotificationTask(AutomationTask): """预警通知任务"""
def __init__(self, config: TaskConfig, notification_config: Dict[str, Any]): super().__init__(config) self.notification_config = notification_config
def execute(self, **kwargs) -> Any: """执行预警通知""" alerts = kwargs.get("alerts", [])
if not alerts: self.logger.info("无预警需要通知") return []
self.logger.info(f"开始发送 {len(alerts)} 条预警通知")
sent_notifications = [] for alert in alerts: # 模拟发送通知 notification = self._send_notification(alert) sent_notifications.append(notification)
self.logger.info(f"预警通知发送完成: {len(sent_notifications)} 条") return sent_notifications
def _send_notification(self, alert: Dict) -> Dict: """发送单条通知""" # 实际项目中调用通知服务 return { "alert_id": alert.get("record_id"), "alert_level": alert.get("alert_level"), "sent_at": datetime.now(timezone.utc).isoformat(), "channels": ["dingtalk", "email"], "status": "sent" }
class DataArchivingTask(AutomationTask): """数据归档任务"""
def __init__(self, config: TaskConfig, archive_config: Dict[str, Any]): super().__init__(config) self.archive_config = archive_config
def execute(self, **kwargs) -> Any: """执行数据归档""" retention_days = self.archive_config.get("retention_days", 365) archive_dir = Path(self.archive_config.get("archive_dir", "./archive"))
self.logger.info(f"开始归档超过 {retention_days} 天的数据")
# 计算归档截止日期 cutoff_date = datetime.now(timezone.utc) - timedelta(days=retention_days)
# 模拟归档操作 archived_count = 0 # 实际项目中从数据库查询并归档旧数据
self.logger.info(f"归档完成: {archived_count} 条记录") return {"archived_count": archived_count, "cutoff_date": cutoff_date.isoformat()}
class AutomationEngine: """自动化引擎"""
def __init__(self, config: AutomationConfig): self.config = config self.logger = logging.getLogger("AutomationEngine") self.scheduler = BackgroundScheduler() self.tasks: Dict[str, AutomationTask] = {} self.task_results: Dict[str, List[TaskResult]] = {} self.running = False
# 确保目录存在 self.config.log_dir.mkdir(parents=True, exist_ok=True) self.config.data_dir.mkdir(parents=True, exist_ok=True) self.config.report_dir.mkdir(parents=True, exist_ok=True) self.config.archive_dir.mkdir(parents=True, exist_ok=True)
# 注册信号处理 signal.signal(signal.SIGTERM, self._signal_handler) signal.signal(signal.SIGINT, self._signal_handler)
def register_task(self, task: AutomationTask): """注册任务""" self.tasks[task.config.task_id] = task self.task_results[task.config.task_id] = []
self.logger.info(f"任务已注册: {task.config.task_id} ({task.config.task_name})")
def schedule_task(self, task_id: str): """调度任务""" task = self.tasks.get(task_id) if not task: self.logger.error(f"任务未找到: {task_id}") return
if not task.config.enabled: self.logger.info(f"任务已禁用: {task_id}") return
# 解析调度表达式 schedule_str = task.config.schedule
if schedule_str.startswith("cron:"): # cron表达式 cron_expr = schedule_str.replace("cron:", "") parts = cron_expr.split() if len(parts) == 5: trigger = CronTrigger( minute=parts[0], hour=parts[1], day=parts[2], month=parts[3], day_of_week=parts[4] ) else: self.logger.error(f"无效的cron表达式: {cron_expr}") return elif schedule_str.startswith("interval:"): # 间隔表达式 interval_str = schedule_str.replace("interval:", "") # 解析如 "4h", "30m", "1d" value = int(interval_str[:-1]) unit = interval_str[-1]
if unit == 'm': trigger = IntervalTrigger(minutes=value) elif unit == 'h': trigger = IntervalTrigger(hours=value) elif unit == 'd': trigger = IntervalTrigger(days=value) else: self.logger.error(f"无效的间隔单位: {unit}") return else: self.logger.error(f"无效的调度表达式: {schedule_str}") return
# 添加任务到调度器 self.scheduler.add_job( self._execute_task, trigger=trigger, args=[task_id], id=task_id, name=task.config.task_name, max_instances=1, # 同一任务同时只能执行一个实例 misfire_grace_time=300 # 错过执行后5分钟内仍可执行 )
self.logger.info(f"任务已调度: {task_id} ({schedule_str})")
def _execute_task(self, task_id: str): """执行任务""" task = self.tasks.get(task_id) if not task: return
self.logger.info(f"开始执行任务: {task_id}")
# 检查依赖 for dep_id in task.config.dependencies: dep_results = self.task_results.get(dep_id, []) if not dep_results or dep_results[-1].status != TaskStatus.SUCCESS: self.logger.warning(f"任务 {task_id} 依赖 {dep_id} 未满足，跳过执行") return
# 执行任务 result = task.run()
# 记录结果 self.task_results[task_id].append(result)
# 保存结果到日志 self._save_task_result(result)
def _save_task_result(self, result: TaskResult): """保存任务结果""" result_file = self.config.log_dir / f"{result.task_id}_results.json"
results = [] if result_file.exists(): with open(result_file, 'r', encoding='utf-8') as f: results = json.load(f)
results.append({ "task_id": result.task_id, "status": result.status.value, "start_time": result.start_time.isoformat() if result.start_time else None, "end_time": result.end_time.isoformat() if result.end_time else None, "duration_ms": result.duration_ms, "error_message": result.error_message, "retry_count": result.retry_count })
with open(result_file, 'w', encoding='utf-8') as f: json.dump(results, f, ensure_ascii=False, indent=2)
def start(self): """启动自动化引擎""" self.logger.info("启动自动化引擎")
# 调度所有已注册任务 for task_id in self.tasks: self.schedule_task(task_id)
# 启动调度器 self.scheduler.start() self.running = True
self.logger.info("自动化引擎已启动")
def stop(self): """停止自动化引擎""" self.logger.info("停止自动化引擎") self.running = False
# 关闭调度器 self.scheduler.shutdown(wait=True)
self.logger.info("自动化引擎已停止")
def _signal_handler(self, signum, frame): """信号处理""" self.logger.info(f"收到信号 {signum}，准备关闭") self.stop() sys.exit(0)
def get_task_status(self, task_id: str) -> Optional[TaskResult]: """获取任务状态""" results = self.task_results.get(task_id, []) return results[-1] if results else None
def get_all_status(self) -> Dict[str, TaskResult]: """获取所有任务状态""" return {task_id: self.get_task_status(task_id) for task_id in self.tasks}
def run_task_now(self, task_id: str) -> TaskResult: """立即执行任务""" self._execute_task(task_id) return self.get_task_status(task_id)2.2 工作流编排
Section titled “2.2 工作流编排”"""工作流编排模块 - 将多个任务组合为工作流"""
from typing import List, Dict, Any
class Workflow: """工作流"""
def __init__(self, workflow_id: str, workflow_name: str): self.workflow_id = workflow_id self.workflow_name = workflow_name self.steps: List[Dict[str, Any]] = [] self.logger = logging.getLogger(f"Workflow.{workflow_id}")
def add_step( self, task: AutomationTask, input_from: Optional[str] = None, condition: Optional[Callable] = None ): """添加工作流步骤""" self.steps.append({ "task": task, "input_from": input_from, "condition": condition })
def execute(self, initial_data: Any = None) -> Dict[str, Any]: """执行工作流""" self.logger.info(f"开始执行工作流: {self.workflow_name}")
results = {} current_data = initial_data
for i, step in enumerate(self.steps): task = step["task"] input_from = step["input_from"] condition = step["condition"]
# 检查条件 if condition and not condition(current_data): self.logger.info(f"步骤 {i+1} 条件不满足，跳过") continue
# 准备输入数据 if input_from and input_from in results: task_input = {"data": results[input_from].output} else: task_input = {"data": current_data}
# 执行任务 self.logger.info(f"执行步骤 {i+1}/{len(self.steps)}: {task.config.task_name}") result = task.run(**task_input)
results[task.config.task_id] = result current_data = result.output
# 如果任务失败，中断工作流 if result.status == TaskStatus.FAILED: self.logger.error(f"工作流在步骤 {i+1} 失败") break
self.logger.info(f"工作流执行完成: {self.workflow_name}") return results
# 预定义工作流模板def create_monitoring_workflow( engine: AutomationEngine, project_id: str, instrument_config: Dict[str, Any], processing_config: Dict[str, Any], report_config: Dict[str, Any], notification_config: Dict[str, Any]) -> Workflow: """ 创建标准监测工作流
流程: 数据采集 -> 数据处理 -> 预警通知 -> 报告生成 """ workflow = Workflow( workflow_id=f"monitoring_{project_id}", workflow_name=f"{project_id} 自动化监测工作流" )
# 1. 数据采集 collection_task = DataCollectionTask( TaskConfig( task_id="data_collection", task_name="数据采集", task_type="collection", schedule="interval:4h", priority=TaskPriority.HIGH ), instrument_config )
# 2. 数据处理 processing_task = DataProcessingTask( TaskConfig( task_id="data_processing", task_name="数据处理", task_type="processing", schedule="interval:4h", priority=TaskPriority.HIGH, dependencies=["data_collection"] ), processing_config )
# 3. 预警通知（只在有预警时执行） alert_task = AlertNotificationTask( TaskConfig( task_id="alert_notification", task_name="预警通知", task_type="notification", schedule="interval:4h", priority=TaskPriority.CRITICAL, dependencies=["data_processing"] ), notification_config )
# 4. 报告生成 report_task = ReportGenerationTask( TaskConfig( task_id="report_generation", task_name="报告生成", task_type="report", schedule="cron:0 8 * * *", # 每天8:00 priority=TaskPriority.NORMAL, dependencies=["data_processing"] ), report_config )
# 添加步骤 workflow.add_step(collection_task) workflow.add_step(processing_task, input_from="data_collection") workflow.add_step( alert_task, input_from="data_processing", condition=lambda data: any(d.get("alert_level") != "正常" for d in (data or [])) ) workflow.add_step(report_task, input_from="data_processing")
return workflow3. 参数说明
Section titled “3. 参数说明”3.1 TaskConfig 任务配置参数
Section titled “3.1 TaskConfig 任务配置参数”| 参数名 | 类型 | 必填 | 默认值 | 说明 |
|---|---|---|---|---|
task_id |
str | 是 | - | 任务唯一标识 |
task_name |
str | 是 | - | 任务名称 |
task_type |
str | 是 | - | 任务类型 |
schedule |
str | 是 | - | 调度表达式（cron:或interval:） |
priority |
TaskPriority | 否 | NORMAL | 任务优先级 |
timeout_seconds |
int | 否 | 3600 | 任务超时时间(秒) |
max_retries |
int | 否 | 3 | 最大重试次数 |
retry_delay_seconds |
int | 否 | 300 | 重试延迟(秒) |
enabled |
bool | 否 | True | 是否启用 |
dependencies |
List[str] | 否 | [] | 依赖任务ID列表 |
parameters |
Dict | 否 | {} | 任务参数 |
3.2 AutomationConfig 自动化配置参数
Section titled “3.2 AutomationConfig 自动化配置参数”| 参数名 | 类型 | 必填 | 默认值 | 说明 |
|---|---|---|---|---|
project_id |
str | 是 | - | 项目编号 |
project_name |
str | 是 | - | 项目名称 |
log_dir |
Path | 否 | ./logs | 日志目录 |
data_dir |
Path | 否 | ./data | 数据目录 |
report_dir |
Path | 否 | ./reports | 报告目录 |
archive_dir |
Path | 否 | ./archive | 归档目录 |
enable_health_check |
bool | 否 | True | 启用健康检查 |
health_check_interval_seconds |
int | 否 | 60 | 健康检查间隔(秒) |
max_concurrent_tasks |
int | 否 | 5 | 最大并发任务数 |
shutdown_timeout_seconds |
int | 否 | 30 | 关闭超时(秒) |
3.3 调度表达式格式
Section titled “3.3 调度表达式格式”| 格式 | 示例 | 说明 |
|---|---|---|
cron:分 时 日 月 周 |
cron:0 8 * * * |
每天8:00 |
cron:0 */4 * * * |
cron:0 */4 * * * |
每4小时 |
interval:4h |
interval:4h |
每4小时 |
interval:30m |
interval:30m |
每30分钟 |
interval:1d |
interval:1d |
每天 |
4. 使用示例
Section titled “4. 使用示例”4.1 基本自动化工作流
Section titled “4.1 基本自动化工作流”from datetime import datetime, timezonefrom pathlib import Pathfrom template_automation_script import ( AutomationEngine, AutomationConfig, TaskConfig, TaskPriority, DataCollectionTask, DataProcessingTask, AlertNotificationTask, ReportGenerationTask, DataArchivingTask)
# 1. 配置自动化引擎config = AutomationConfig( project_id="RW-2024-001", project_name="绕城高速管廊工程1标段", log_dir=Path("./logs"), data_dir=Path("./data"), report_dir=Path("./reports"), archive_dir=Path("./archive"))
engine = AutomationEngine(config)
# 2. 配置仪器instrument_config = { "type": "total_station", "id": "TS-001", "brand": "Leica", "model": "TS16", "connection": { "type": "tcp", "host": "192.168.1.100", "port": 8080 }}
# 3. 配置处理参数processing_config = { "thresholds": { "yellow": 5.0, "orange": 8.0, "red": 10.0 }, "baseline_date": "2024-01-01T00:00:00Z"}
# 4. 配置报告report_config = { "type": "daily", "project_id": "RW-2024-001", "output_dir": "./reports", "template": "daily_report_template"}
# 5. 配置通知notification_config = { "channels": ["dingtalk", "email"], "recipients": ["monitoring@railwise.com"], "templates": { "yellow": "yellow_alert_template", "orange": "orange_alert_template", "red": "red_alert_template" }}
# 6. 创建并注册任务# 6.1 数据采集任务（每4小时）collection_task = DataCollectionTask( TaskConfig( task_id="data_collection", task_name="全站仪数据采集", task_type="collection", schedule="interval:4h", priority=TaskPriority.HIGH ), instrument_config)engine.register_task(collection_task)
# 6.2 数据处理任务（每4小时，依赖采集）processing_task = DataProcessingTask( TaskConfig( task_id="data_processing", task_name="监测数据处理", task_type="processing", schedule="interval:4h", priority=TaskPriority.HIGH, dependencies=["data_collection"] ), processing_config)engine.register_task(processing_task)
# 6.3 预警通知任务（每4小时，依赖处理）alert_task = AlertNotificationTask( TaskConfig( task_id="alert_notification", task_name="预警通知推送", task_type="notification", schedule="interval:4h", priority=TaskPriority.CRITICAL, dependencies=["data_processing"] ), notification_config)engine.register_task(alert_task)
# 6.4 日报生成任务（每天8:00，依赖处理）report_task = ReportGenerationTask( TaskConfig( task_id="daily_report", task_name="监测日报生成", task_type="report", schedule="cron:0 8 * * *", priority=TaskPriority.NORMAL, dependencies=["data_processing"] ), report_config)engine.register_task(report_task)
# 6.5 数据归档任务（每月1日2:00）archive_task = DataArchivingTask( TaskConfig( task_id="data_archive", task_name="历史数据归档", task_type="archive", schedule="cron:0 2 1 * *", priority=TaskPriority.LOW ), {"retention_days": 365, "archive_dir": "./archive"})engine.register_task(archive_task)
# 7. 启动引擎engine.start()
# 8. 查看任务状态print("任务状态:")for task_id, status in engine.get_all_status().items(): if status: print(f" {task_id}: {status.status.value} (耗时: {status.duration_ms:.2f}ms)")
# 9. 手动触发任务# result = engine.run_task_now("data_collection")
# 10. 保持运行try: while engine.running: time.sleep(1)except KeyboardInterrupt: engine.stop()4.2 使用预定义工作流
Section titled “4.2 使用预定义工作流”from template_automation_script import ( AutomationEngine, AutomationConfig, create_monitoring_workflow)
# 创建引擎config = AutomationConfig( project_id="RW-2024-001", project_name="绕城高速管廊工程1标段")engine = AutomationEngine(config)
# 创建标准监测工作流workflow = create_monitoring_workflow( engine=engine, project_id="RW-2024-001", instrument_config=instrument_config, processing_config=processing_config, report_config=report_config, notification_config=notification_config)
# 执行工作流results = workflow.execute()
# 查看结果for task_id, result in results.items(): print(f"{task_id}: {result.status.value}")4.3 自定义任务
Section titled “4.3 自定义任务”from template_automation_script import AutomationTask, TaskConfig, TaskResult, TaskStatus
class CustomHealthCheckTask(AutomationTask): """自定义健康检查任务"""
def __init__(self, config: TaskConfig, check_items: List[str]): super().__init__(config) self.check_items = check_items
def execute(self, **kwargs) -> Any: """执行健康检查""" health_status = {}
for item in self.check_items: # 模拟检查 health_status[item] = { "status": "healthy", "last_check": datetime.now(timezone.utc).isoformat() }
return health_status
# 注册自定义任务health_task = CustomHealthCheckTask( TaskConfig( task_id="health_check", task_name="系统健康检查", task_type="health", schedule="interval:5m", priority=TaskPriority.HIGH ), check_items=["database", "network", "disk_space", "memory"])
engine.register_task(health_task)5. 输出示例
Section titled “5. 输出示例”5.1 任务执行日志
Section titled “5.1 任务执行日志”2024-07-08 08:00:00,123 - AutomationEngine - INFO - 启动自动化引擎2024-07-08 08:00:00,125 - AutomationEngine - INFO - 任务已调度: data_collection (interval:4h)2024-07-08 08:00:00,126 - AutomationEngine - INFO - 任务已调度: data_processing (interval:4h)2024-07-08 08:00:00,127 - AutomationEngine - INFO - 任务已调度: alert_notification (interval:4h)2024-07-08 08:00:00,128 - AutomationEngine - INFO - 任务已调度: daily_report (cron:0 8 * * *)2024-07-08 08:00:00,129 - AutomationEngine - INFO - 任务已调度: data_archive (cron:0 2 1 * *)2024-07-08 08:00:00,130 - AutomationEngine - INFO - 自动化引擎已启动
2024-07-08 08:00:00,200 - Task.data_collection - INFO - 任务 data_collection 开始执行2024-07-08 08:00:00,201 - Task.data_collection - INFO - 开始从 TS-001 (total_station) 采集数据2024-07-08 08:00:05,456 - Task.data_collection - INFO - 采集完成: 5 条记录2024-07-08 08:00:05,457 - Task.data_collection - INFO - 任务 data_collection 成功完成，耗时: 5257.00ms
2024-07-08 08:00:05,500 - Task.data_processing - INFO - 任务 data_processing 开始执行2024-07-08 08:00:05,501 - Task.data_processing - INFO - 开始处理 5 条记录2024-07-08 08:00:06,123 - Task.data_processing - INFO - 处理完成: 清洗 5 条, 预警 0 条2024-07-08 08:00:06,124 - Task.data_processing - INFO - 任务 data_processing 成功完成，耗时: 624.00ms
2024-07-08 08:00:06,200 - Task.alert_notification - INFO - 无预警需要通知5.2 任务结果文件
Section titled “5.2 任务结果文件”[ { "task_id": "data_collection", "status": "success", "start_time": "2024-07-08T08:00:00.200000+00:00", "end_time": "2024-07-08T08:00:05.456000+00:00", "duration_ms": 5257.0, "error_message": null, "retry_count": 0 }, { "task_id": "data_processing", "status": "success", "start_time": "2024-07-08T08:00:05.500000+00:00", "end_time": "2024-07-08T08:00:06.124000+00:00", "duration_ms": 624.0, "error_message": null, "retry_count": 0 }]5.3 工作流执行结果
Section titled “5.3 工作流执行结果”{ "data_collection": TaskResult( task_id="data_collection", status=TaskStatus.SUCCESS, output=[{...}, {...}, ...] ), "data_processing": TaskResult( task_id="data_processing", status=TaskStatus.SUCCESS, output=[{...}, {...}, ...] ), "alert_notification": TaskResult( task_id="alert_notification", status=TaskStatus.SKIPPED, # 无预警，跳过 output=[] ), "daily_report": TaskResult( task_id="daily_report", status=TaskStatus.SUCCESS, output=Path("reports/report_daily_20240708_080000.json") )}6. 扩展指南
Section titled “6. 扩展指南”6.1 添加新的任务类型
Section titled “6.1 添加新的任务类型”from template_automation_script import AutomationTask, TaskConfig
class DatabaseBackupTask(AutomationTask): """数据库备份任务"""
def __init__(self, config: TaskConfig, db_config: Dict[str, Any]): super().__init__(config) self.db_config = db_config
def execute(self, **kwargs) -> Any: """执行数据库备份""" import subprocess
db_name = self.db_config.get("database", "monitoring") backup_path = self.db_config.get("backup_dir", "./backups")
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_file = f"{backup_path}/{db_name}_{timestamp}.sql"
# 执行备份命令 cmd = [ "mysqldump", "-h", self.db_config.get("host", "localhost"), "-u", self.db_config.get("user", "root"), "-p" + self.db_config.get("password", ""), db_name ]
with open(backup_file, 'w') as f: subprocess.run(cmd, stdout=f, check=True)
return {"backup_file": backup_file, "size": Path(backup_file).stat().st_size}
# 注册backup_task = DatabaseBackupTask( TaskConfig( task_id="db_backup", task_name="数据库备份", task_type="backup", schedule="cron:0 3 * * *", # 每天3:00 priority=TaskPriority.LOW ), db_config={"database": "railwise_monitoring", "host": "localhost", "user": "backup_user"})
engine.register_task(backup_task)6.2 集成到系统服务
Section titled “6.2 集成到系统服务”# 保存为 automation_service.pyimport sysimport timefrom template_automation_script import AutomationEngine, AutomationConfig
def main(): """主函数""" config = AutomationConfig( project_id="RW-2024-001", project_name="绕城高速管廊工程1标段" )
engine = AutomationEngine(config)
# 注册任务... # engine.register_task(...)
# 启动 engine.start()
# 保持运行 try: while True: time.sleep(1) except KeyboardInterrupt: engine.stop()
if __name__ == "__main__": main()使用systemd管理：
[Unit]Description=RailWise Automation ServiceAfter=network.target
[Service]Type=simpleUser=railwiseWorkingDirectory=/opt/railwiseExecStart=/opt/railwise/venv/bin/python automation_service.pyRestart=alwaysRestartSec=10
[Install]WantedBy=multi-user.target6.3 监控任务执行状态
Section titled “6.3 监控任务执行状态”from template_automation_script import AutomationEngineimport json
def monitor_tasks(engine: AutomationEngine): """监控任务状态""" status = engine.get_all_status()
report = { "timestamp": datetime.now(timezone.utc).isoformat(), "tasks": {} }
for task_id, result in status.items(): if result: report["tasks"][task_id] = { "status": result.status.value, "last_run": result.end_time.isoformat() if result.end_time else None, "duration_ms": result.duration_ms, "success_rate": calculate_success_rate(task_id) }
# 保存监控报告 with open("task_monitor.json", "w") as f: json.dump(report, f, indent=2)
def calculate_success_rate(task_id: str) -> float: """计算任务成功率""" # 从结果历史计算 return 0.95 # 示例6.4 异常恢复机制
Section titled “6.4 异常恢复机制”class ResilientTask(AutomationTask): """弹性任务（支持异常恢复）"""
def __init__(self, config: TaskConfig): super().__init__(config) self.state_file = Path(f"./state/{config.task_id}_state.json") self.state_file.parent.mkdir(parents=True, exist_ok=True)
def save_state(self, state: Dict): """保存状态""" with open(self.state_file, 'w') as f: json.dump(state, f)
def load_state(self) -> Optional[Dict]: """加载状态""" if self.state_file.exists(): with open(self.state_file, 'r') as f: return json.load(f) return None
def execute(self, **kwargs) -> Any: """执行（支持恢复）""" state = self.load_state()
if state and state.get("in_progress"): self.logger.info("检测到未完成的任务，尝试恢复") # 从上次中断处恢复 return self._resume_from_state(state)
# 正常执行 self.save_state({"in_progress": True, "start_time": datetime.now().isoformat()})
try: result = self._do_work(**kwargs) self.save_state({"in_progress": False, "completed": True}) return result except Exception as e: self.save_state({"in_progress": True, "error": str(e)}) raise7. 常见问题
Section titled “7. 常见问题”Q1: 任务执行超时怎么办？
Section titled “Q1: 任务执行超时怎么办？”A: 设置合理的超时时间，并在任务中定期检查：
def execute(self, **kwargs): start_time = time.time() timeout = self.config.timeout_seconds
while True: if time.time() - start_time > timeout: raise TimeoutError("任务执行超时")
# 执行任务... time.sleep(1)Q2: 如何处理任务依赖失败？
Section titled “Q2: 如何处理任务依赖失败？”A: 在任务配置中设置依赖，引擎会自动检查：
TaskConfig( task_id="dependent_task", dependencies=["prerequisite_task"], # 依赖任务 # 如果依赖任务失败，本任务会自动跳过)Q3: 如何查看任务执行历史？
Section titled “Q3: 如何查看任务执行历史？”A: 任务结果保存在日志目录：
cat logs/data_collection_results.jsonQ4: 如何手动触发任务？
Section titled “Q4: 如何手动触发任务？”A: 使用run_task_now方法：
result = engine.run_task_now("data_collection")print(f"执行结果: {result.status.value}")Q5: 如何优雅关闭引擎？
Section titled “Q5: 如何优雅关闭引擎？”A: 发送SIGTERM信号或调用stop方法：
# 方法1：信号# kill -TERM <pid>
# 方法2：代码engine.stop()相关文档链接
Section titled “相关文档链接”- 监测数据导入标准化模板 — 自动化数据采集
- 监测报告自动生成模板 — 自动化报告生成
- 监测预警通知模板 — 自动化预警通知
- 批量数据处理模板 — 自动化数据处理
ai_tags: domain: ["工程监测", "自动化运维", "任务调度"] technology: ["Python", "APScheduler", "定时任务", "工作流"] standard: ["GB 50497", "CJJ/T 202", "JGJ 8"] product: ["RAILWISE-TSM", "RAILWISE-CLI", "WorkWise"] difficulty: "advanced" role: ["运维工程师", "系统管理员", "监测工程师"] task_type: ["自动化", "定时任务", "运维监控"]