Python脚本如何自定义业务同步触发条件

wen python案例 2026-06-24 260

本文目录导读：

基于时间条件的触发
基于数据变更的触发
基于业务规则的触发
综合触发管理器
使用示例
高级特性：组合条件触发器

我来详细介绍Python中自定义业务同步触发条件的几种常用方法：

基于时间条件的触发

import time
from datetime import datetime, timedelta
import schedule
class BusinessSyncer:
    def __init__(self):
        self.last_sync_time = None
        self.sync_interval = 300  # 默认5分钟
    # 定时触发
    def sync_by_time(self, interval_minutes=5):
        """按固定时间间隔同步"""
        current_time = datetime.now()
        if self.last_sync_time is None:
            self.last_sync_time = current_time
            return True
        # 计算时间差
        time_diff = (current_time - self.last_sync_time).total_seconds()
        if time_diff >= interval_minutes * 60:
            self.last_sync_time = current_time
            return True
        return False
    # 指定时间点触发
    def sync_at_specific_time(self):
        """在特定时间点同步"""
        current_time = datetime.now()
        sync_times = ["09:00", "12:00", "15:00", "18:00", "21:00"]
        current_time_str = current_time.strftime("%H:%M")
        if current_time_str in sync_times:
            return True
        return False

基于数据变更的触发

import hashlib
import json
import os
class DataChangeTrigger:
    def __init__(self):
        self.data_checksum = None
        self.data_file = "data_checksum.txt"
    def calculate_checksum(self, data):
        """计算数据的校验和"""
        data_str = json.dumps(data, sort_keys=True)
        return hashlib.md5(data_str.encode()).hexdigest()
    def check_data_changed(self, new_data):
        """检查数据是否发生变化"""
        new_checksum = self.calculate_checksum(new_data)
        if self.data_checksum is None:
            self.data_checksum = new_checksum
            return True
        if new_checksum != self.data_checksum:
            self.data_checksum = new_checksum
            return True
        return False
    def check_file_changed(self, filepath):
        """检查文件是否被修改"""
        if not os.path.exists(filepath):
            return False
        # 检查文件的最后修改时间
        current_mtime = os.path.getmtime(filepath)
        if not hasattr(self, 'last_mtime'):
            self.last_mtime = current_mtime
            return True
        if current_mtime > self.last_mtime:
            self.last_mtime = current_mtime
            return True
        return False

基于业务规则的触发

class BusinessRuleTrigger:
    def __init__(self):
        self.thresholds = {
            "order_count": 100,
            "revenue": 10000,
            "error_rate": 0.05
        }
    def check_business_rules(self, business_metrics):
        """检查业务规则是否触发同步"""
        triggers = []
        # 规则1：订单量达到阈值
        if business_metrics.get("order_count", 0) >= self.thresholds["order_count"]:
            triggers.append("order_threshold")
        # 规则2：收入达到阈值
        if business_metrics.get("revenue", 0) >= self.thresholds["revenue"]:
            triggers.append("revenue_threshold")
        # 规则3：错误率超过阈值
        if business_metrics.get("error_rate", 0) >= self.thresholds["error_rate"]:
            triggers.append("error_threshold")
        # 规则4：组合条件
        if (business_metrics.get("order_count", 0) >= 50 and 
            business_metrics.get("revenue", 0) >= 5000):
            triggers.append("combined_condition")
        return triggers
    def check_state_machine(self, current_state, event):
        """基于状态机的触发条件"""
        state_machine = {
            "init": {"data_ready": "syncing"},
            "syncing": {"success": "synced", "failed": "error"},
            "synced": {"new_data": "pending", "timeout": "syncing"},
            "error": {"retry": "syncing", "force_sync": "syncing"}
        }
        if current_state in state_machine and event in state_machine[current_state]:
            next_state = state_machine[current_state][event]
            return next_state == "syncing"
        return False

综合触发管理器

from typing import List, Dict, Any, Callable
import threading
import logging
class TriggerManager:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.triggers: List[Dict[str, Any]] = []
        self.sync_function = None
    def register_trigger(self, name: str, condition_func: Callable, 
                        priority: int = 0, cooldown: int = 60):
        """注册触发条件"""
        trigger = {
            "name": name,
            "condition": condition_func,
            "priority": priority,
            "cooldown": cooldown,
            "last_triggered": 0
        }
        self.triggers.append(trigger)
        self.triggers.sort(key=lambda x: x["priority"], reverse=True)
    def evaluate_triggers(self, context: Dict[str, Any]) -> List[str]:
        """评估所有触发条件"""
        triggered = []
        current_time = time.time()
        for trigger in self.triggers:
            try:
                # 检查冷却时间
                if current_time - trigger["last_triggered"] < trigger["cooldown"]:
                    continue
                # 执行条件判断
                if trigger["condition"](context):
                    trigger["last_triggered"] = current_time
                    triggered.append(trigger["name"])
                    self.logger.info(f"Trigger {trigger['name']} activated")
            except Exception as e:
                self.logger.error(f"Error evaluating trigger {trigger['name']}: {e}")
        return triggered
    def set_sync_function(self, func: Callable):
        """设置同步函数"""
        self.sync_function = func
    def run_sync(self, context: Dict[str, Any]):
        """执行同步"""
        if not self.sync_function:
            self.logger.error("No sync function set")
            return
        triggered_triggers = self.evaluate_triggers(context)
        if triggered_triggers:
            try:
                self.logger.info(f"Starting sync due to triggers: {triggered_triggers}")
                self.sync_function(context)
            except Exception as e:
                self.logger.error(f"Sync failed: {e}")

使用示例

# 创建触发管理实例
trigger_manager = TriggerManager()
# 定义同步函数
def sync_data(context):
    print(f"Syncing data with context: {context}")
    # 实际同步逻辑...
# 设置同步函数
trigger_manager.set_sync_function(sync_data)
# 创建各种触发器
class CustomTriggers:
    @staticmethod
    def order_volume_trigger(context):
        return context.get("order_count", 0) > 100
    @staticmethod
    def time_interval_trigger(context):
        # 每30分钟触发一次
        return datetime.now().minute % 30 == 0
    @staticmethod
    def database_change_trigger(context):
        # 检查数据库是否有新记录
        return context.get("db_changed", False)
    @staticmethod
    def error_rate_trigger(context):
        # 错误率超过0.1%时触发
        return context.get("error_rate", 0) > 0.001
# 注册触发器
trigger_manager.register_trigger(
    "order_volume", 
    CustomTriggers.order_volume_trigger, 
    priority=10, 
    cooldown=300
)
trigger_manager.register_trigger(
    "time_interval", 
    CustomTriggers.time_interval_trigger, 
    priority=5, 
    cooldown=1800
)
trigger_manager.register_trigger(
    "db_change", 
    CustomTriggers.database_change_trigger, 
    priority=8, 
    cooldown=60
)
trigger_manager.register_trigger(
    "error_rate", 
    CustomTriggers.error_rate_trigger, 
    priority=9, 
    cooldown=120
)
# 业务上下文
business_context = {
    "order_count": 150,
    "error_rate": 0.002,
    "db_changed": True,
    "timestamp": datetime.now()
}
# 执行同步检查
trigger_manager.run_sync(business_context)

高级特性：组合条件触发器

class CompositeTrigger:
    def __init__(self):
        self.conditions = {}
    def and_condition(self, *conditions):
        """AND条件组合"""
        def check(context):
            return all(cond(context) for cond in conditions)
        return check
    def or_condition(self, *conditions):
        """OR条件组合"""
        def check(context):
            return any(cond(context) for cond in conditions)
        return check
    def not_condition(self, condition):
        """NOT条件"""
        def check(context):
            return not condition(context)
        return check
    def threshold_condition(self, key, threshold, operator="greater"):
        """阈值条件"""
        def check(context):
            value = context.get(key, 0)
            if operator == "greater":
                return value > threshold
            elif operator == "less":
                return value < threshold
            elif operator == "equal":
                return value == threshold
            return False
        return check
# 使用组合条件
composite = CompositeTrigger()
# 创建复杂条件
critical_sync_condition = composite.and_condition(
    composite.threshold_condition("order_count", 50, "greater"),
    composite.or_condition(
        composite.threshold_condition("revenue", 10000, "greater"),
        composite.threshold_condition("error_rate", 0.05, "greater"),
        composite.not_condition(
            composite.threshold_condition("system_load", 90, "less")
        )
    )
)
# 注册组合条件
trigger_manager.register_trigger(
    "critical_sync", 
    critical_sync_condition, 
    priority=20, 
    cooldown=60
)

这样的设计提供了：