Python脚本如何适配自动化触发同步场景

wen python案例 2026-06-24 28

本文目录导读：

文件系统监控触发
定时调度触发
HTTP/WebHook 触发
消息队列触发
数据库变更捕获
综合适配方案
配置驱动触发
关键设计原则

我来详细说明Python脚本适配自动化触发同步场景的多种方案：

文件系统监控触发

使用 watchdog 库

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class SyncHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory:
            print(f"文件变化: {event.src_path}")
            # 执行同步逻辑
            self.sync_file(event.src_path)
    def sync_file(self, file_path):
        # 同步逻辑实现
        print(f"同步文件: {file_path}")
# 启动监控
observer = Observer()
handler = SyncHandler()
observer.schedule(handler, '/path/to/watch', recursive=True)
observer.start()

定时调度触发

使用 schedule 库

import schedule
import time
def sync_task():
    print("执行定时同步...")
    # 同步逻辑
    pass
# 配置定时任务
schedule.every(10).minutes.do(sync_task)
schedule.every().hour.do(sync_task)
schedule.every().day.at("10:30").do(sync_task)
schedule.every().monday.do(sync_task)
# 持续运行
while True:
    schedule.run_pending()
    time.sleep(1)

HTTP/WebHook 触发

Flask Web服务

from flask import Flask, request, jsonify
import hmac
import hashlib
app = Flask(__name__)
WEBHOOK_SECRET = "your-secret-key"
@app.route('/webhook/sync', methods=['POST'])
def handle_webhook():
    # 验证签名
    signature = request.headers.get('X-Signature')
    if not verify_signature(request.data, signature):
        return jsonify({"error": "Invalid signature"}), 401
    # 触发同步
    sync_data = request.json
    result = execute_sync(sync_data)
    return jsonify({"status": "success", "result": result})
def verify_signature(payload, signature):
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)
def execute_sync(data):
    # 同步逻辑实现
    pass
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

消息队列触发

使用 RabbitMQ

import pika
import json
def callback(ch, method, properties, body):
    message = json.loads(body)
    print(f"收到同步请求: {message}")
    # 执行同步
    execute_sync(message)
    # 确认消息
    ch.basic_ack(delivery_tag=method.delivery_tag)
def start_consumer():
    connection = pika.BlockingConnection(
        pika.ConnectionParameters('localhost')
    )
    channel = connection.channel()
    channel.queue_declare(queue='sync_tasks', durable=True)
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(
        queue='sync_tasks',
        on_message_callback=callback
    )
    print('等待同步任务...')
    channel.start_consuming()
if __name__ == '__main__':
    start_consumer()

数据库变更捕获

使用 Redis 发布/订阅

import redis
import json
r = redis.Redis(host='localhost', port=6379, db=0)
def redis_sync_listener():
    pubsub = r.pubsub()
    pubsub.subscribe('sync_channel')
    for message in pubsub.listen():
        if message['type'] == 'message':
            data = json.loads(message['data'])
            print(f"收到同步通知: {data}")
            execute_sync(data)
def execute_sync(data):
    # 同步逻辑
    pass
if __name__ == '__main__':
    redis_sync_listener()

综合适配方案

import threading
import queue
import time
import signal
import sys
class SyncAdapter:
    def __init__(self):
        self.task_queue = queue.Queue()
        self.running = True
        # 注册信号处理
        signal.signal(signal.SIGTERM, self.shutdown)
        signal.signal(signal.SIGINT, self.shutdown)
    def file_watcher(self, path):
        """文件监控线程"""
        from watchdog.observers import Observer
        from watchdog.events import FileSystemEventHandler
        class Handler(FileSystemEventHandler):
            def on_modified(self, event):
                if not event.is_directory:
                    self.task_queue.put({
                        'type': 'file_change',
                        'path': event.src_path,
                        'timestamp': time.time()
                    })
        observer = Observer()
        observer.schedule(Handler(), path, recursive=True)
        observer.start()
        while self.running:
            time.sleep(1)
        observer.stop()
        observer.join()
    def scheduler(self):
        """定时调度线程"""
        import schedule
        def sync_job():
            self.task_queue.put({
                'type': 'scheduled',
                'timestamp': time.time()
            })
        schedule.every(1).hours.do(sync_job)
        while self.running:
            schedule.run_pending()
            time.sleep(1)
    def webhook_server(self):
        """Webhook服务线程"""
        from flask import Flask, request
        app = Flask(__name__)
        @app.route('/sync', methods=['POST'])
        def webhook():
            data = request.json
            self.task_queue.put({
                'type': 'webhook',
                'data': data,
                'timestamp': time.time()
            })
            return {"status": "accepted"}, 202
        app.run(host='0.0.0.0', port=5000, threaded=True)
    def task_worker(self):
        """任务执行线程"""
        while self.running:
            try:
                task = self.task_queue.get(timeout=1)
                self.execute_sync(task)
                self.task_queue.task_done()
            except queue.Empty:
                continue
    def execute_sync(self, task):
        """核心同步逻辑"""
        task_type = task.get('type')
        print(f"执行同步任务: {task_type}")
        # 同步逻辑实现
        # ...
    def shutdown(self, signum, frame):
        print("收到终止信号，优雅关闭...")
        self.running = False
    def start(self):
        """启动所有组件"""
        threads = [
            threading.Thread(target=self.file_watcher, args=('/path/to/watch',), daemon=True),
            threading.Thread(target=self.scheduler, daemon=True),
            threading.Thread(target=self.webhook_server, daemon=True),
            threading.Thread(target=self.task_worker, daemon=True)
        ]
        for t in threads:
            t.start()
        try:
            while self.running:
                time.sleep(1)
        except KeyboardInterrupt:
            self.shutdown(None, None)
if __name__ == '__main__':
    adapter = SyncAdapter()
    adapter.start()

配置驱动触发

import yaml
import importlib
class ConfigDrivenSync:
    def __init__(self, config_file='sync_config.yaml'):
        with open(config_file) as f:
            self.config = yaml.safe_load(f)
        self.triggers = []
    def load_triggers(self):
        for trigger_config in self.config.get('triggers', []):
            trigger_type = trigger_config['type']
            if trigger_type == 'file_watch':
                self.setup_file_watch(trigger_config)
            elif trigger_type == 'schedule':
                self.setup_schedule(trigger_config)
            elif trigger_type == 'webhook':
                self.setup_webhook(trigger_config)
            elif trigger_type == 'custom':
                self.setup_custom_trigger(trigger_config)
    def setup_file_watch(self, config):
        # 设置文件监控
        pass
    def setup_schedule(self, config):
        # 设置定时任务
        import schedule
        if config.get('interval'):
            getattr(schedule.every(config['interval']), config.get('unit', 'seconds')).do(self.sync_job)
        elif config.get('cron'):
            # 使用cron表达式
            pass
    def setup_webhook(self, config):
        # 设置webhook端点
        from flask import Flask
        app = Flask(__name__)
        @app.route(config.get('endpoint', '/webhook'), methods=['POST'])
        def handle_webhook():
            self.sync_job()
            return {"status": "ok"}
    def setup_custom_trigger(self, config):
        # 加载自定义触发器
        module = importlib.import_module(config['module'])
        trigger_class = getattr(module, config['class'])
        trigger = trigger_class(config.get('params', {}))
        trigger.on_trigger(self.sync_job)
    def sync_job(self):
        # 执行同步
        pass
# 配置文件示例 (sync_config.yaml)
"""
triggers:
  - type: file_watch
    path: /data/sync
    patterns: ["*.csv", "*.json"]
    recursive: true
  - type: schedule
    interval: 30
    unit: minutes
  - type: webhook
    endpoint: /webhook/github
    secret: your-secret
  - type: custom
    module: my_triggers
    class: KafkaTrigger
    params:
      topic: sync-topic
      bootstrap_servers: localhost:9092
"""

关键设计原则

解耦触发与执行：使用任务队列分离触发逻辑和执行逻辑
错误处理：实现重试机制和错误日志
幂等性：确保多次触发不会产生副作用
监控告警：添加健康检查和性能监控
可配置性：使用配置驱动的方式管理触发策略

根据具体场景选择合适的触发方式,组合使用多种触发机制可以实现更灵活可靠的同步系统。