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

本文目录导读：

1. 命令行参数触发（最常用）
2. 环境变量触发
3. 信号处理触发
4. HTTP接口触发
5. 文件监控触发
6. 组合使用示例
7. 使用建议

我来详细说明Python脚本适配手动触发同步场景的几种方法：

命令行参数触发（最常用）

import argparse
import sys
def sync_data(source, target, mode='incremental'):
    """执行数据同步"""
    print(f"同步开始: {source} -> {target}")
    print(f"同步模式: {mode}")
    # 实际同步逻辑
    return True
def main():
    parser = argparse.ArgumentParser(description='数据同步工具')
    parser.add_argument('--source', '-s', required=True, help='源路径')
    parser.add_argument('--target', '-t', required=True, help='目标路径')
    parser.add_argument('--mode', '-m', choices=['full', 'incremental'], 
                       default='incremental', help='同步模式')
    parser.add_argument('--force', '-f', action='store_true', help='强制同步')
    args = parser.parse_args()
    # 执行同步
    success = sync_data(args.source, args.target, args.mode)
    if success:
        print("同步完成")
        return 0
    else:
        print("同步失败")
        return 1
if __name__ == "__main__":
    sys.exit(main())

使用示例：

# 基本用法
python sync.py --source /data/a --target /data/b
# 全量同步
python sync.py -s /data/a -t /data/b -m full
# 强制同步
python sync.py -s /data/a -t /data/b --force

环境变量触发

import os
class SyncTrigger:
    def __init__(self):
        # 从环境变量读取配置
        self.source = os.getenv('SYNC_SOURCE', '/default/source')
        self.target = os.getenv('SYNC_TARGET', '/default/target')
        self.mode = os.getenv('SYNC_MODE', 'incremental')
        self.interval = int(os.getenv('SYNC_INTERVAL', '300'))  # 默认5分钟
    def check_manual_trigger(self):
        """检查是否手动触发"""
        trigger_file = os.getenv('SYNC_TRIGGER_FILE', '/tmp/sync.trigger')
        if os.path.exists(trigger_file):
            print(f"检测到手动触发文件: {trigger_file}")
            # 读取触发参数
            with open(trigger_file, 'r') as f:
                params = f.read().strip()
            # 删除触发文件
            os.remove(trigger_file)
            return True, params
        return False, None
    def run_sync(self, params=None):
        """执行同步"""
        print(f"手动触发同步: {params}")
        # 执行同步逻辑
        return True
# 主循环
def main_loop():
    trigger = SyncTrigger()
    while True:
        # 检查手动触发
        is_manual, params = trigger.check_manual_trigger()
        if is_manual:
            trigger.run_sync(params)
        # 定时任务逻辑
        time.sleep(1)

手动触发方式：

# 创建触发文件
echo "full_sync" > /tmp/sync.trigger

信号处理触发

import signal
import time
import sys
class SyncManager:
    def __init__(self):
        self.sync_running = False
        self.manual_sync_requested = False
        # 注册信号处理器
        signal.signal(signal.SIGUSR1, self.handle_manual_sync)
        signal.signal(signal.SIGTERM, self.handle_shutdown)
    def handle_manual_sync(self, signum, frame):
        """处理手动同步信号"""
        print(f"\n接收到手动同步信号: {signum}")
        self.manual_sync_requested = True
    def handle_shutdown(self, signum, frame):
        """处理关闭信号"""
        print(f"\n接收到关闭信号: {signum}")
        self.sync_running = False
    def perform_sync(self):
        """执行同步操作"""
        print("执行数据同步...")
        time.sleep(2)  # 模拟同步
        print("同步完成")
    def run(self):
        """主运行循环"""
        self.sync_running = True
        print("同步服务已启动，PID:", os.getpid())
        print("发送 SIGUSR1 信号手动触发同步")
        while self.sync_running:
            if self.manual_sync_requested:
                self.perform_sync()
                self.manual_sync_requested = False
            # 正常的定时检查逻辑
            time.sleep(1)
if __name__ == "__main__":
    import os
    manager = SyncManager()
    manager.run()

手动触发方式：

# 获取进程PID并发送信号
kill -SIGUSR1 <PID>
# 或者使用pgrep查找PID
kill -SIGUSR1 $(pgrep -f sync_script.py)

HTTP接口触发

from flask import Flask, request, jsonify
import threading
import time
app = Flask(__name__)
class SyncService:
    def __init__(self):
        self.sync_status = {
            'running': False,
            'last_sync': None,
            'errors': []
        }
        self.lock = threading.Lock()
    def execute_sync(self, params):
        """执行同步"""
        with self.lock:
            self.sync_status['running'] = True
        try:
            print(f"执行同步: {params}")
            time.sleep(2)  # 模拟同步
            with self.lock:
                self.sync_status['last_sync'] = time.time()
                self.sync_status['running'] = False
            return {'status': 'success', 'message': '同步完成'}
        except Exception as e:
            with self.lock:
                self.sync_status['running'] = False
                self.sync_status['errors'].append(str(e))
            return {'status': 'error', 'message': str(e)}
sync_service = SyncService()
@app.route('/sync', methods=['POST'])
def trigger_sync():
    """手动触发同步接口"""
    params = request.json or {}
    # 检查是否正在同步
    if sync_service.sync_status['running']:
        return jsonify({
            'status': 'error',
            'message': '同步正在进行中'
        }), 409
    # 异步执行同步
    thread = threading.Thread(target=sync_service.execute_sync, args=(params,))
    thread.start()
    return jsonify({
        'status': 'accepted',
        'message': '同步任务已启动'
    }), 202
@app.route('/sync/status', methods=['GET'])
def get_sync_status():
    """获取同步状态"""
    return jsonify(sync_service.sync_status)
@app.route('/sync/config', methods=['PUT'])
def update_sync_config():
    """更新同步配置"""
    config = request.json
    # 更新配置逻辑
    return jsonify({'status': 'success'})
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

HTTP触发方式：

# 触发同步
curl -X POST http://localhost:5000/sync \
  -H "Content-Type: application/json" \
  -d '{"mode": "full", "force": true}'
# 查看状态
curl http://localhost:5000/sync/status

文件监控触发

import watchdog
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time
class SyncTriggerHandler(FileSystemEventHandler):
    def __init__(self, sync_function):
        self.sync_function = sync_function
        self.last_trigger = 0
    def on_created(self, event):
        """监控文件创建事件"""
        if event.is_directory:
            return
        # 防止重复触发
        current_time = time.time()
        if current_time - self.last_trigger < 1:
            return
        file_path = event.src_path
        print(f"检测到新文件: {file_path}")
        # 执行同步
        self.sync_function(file_path)
        self.last_trigger = current_time
class FileSyncMonitor:
    def __init__(self, watch_dir, sync_function):
        self.watch_dir = watch_dir
        self.sync_function = sync_function
        self.observer = Observer()
    def start(self):
        """启动监控"""
        event_handler = SyncTriggerHandler(self.sync_function)
        self.observer.schedule(event_handler, self.watch_dir, recursive=True)
        self.observer.start()
        print(f"开始监控目录: {self.watch_dir}")
    def stop(self):
        """停止监控"""
        self.observer.stop()
        self.observer.join()
def file_sync_handler(source_file):
    """文件同步处理函数"""
    print(f"同步文件: {source_file}")
    # 实际同步逻辑
    return True
# 使用示例
if __name__ == "__main__":
    monitor = FileSyncMonitor("/data/incoming", file_sync_handler)
    try:
        monitor.start()
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        monitor.stop()

组合使用示例

#!/usr/bin/env python3
"""
综合手动触发同步脚本
支持多种触发方式：命令行、信号、HTTP API、文件监控
"""
import argparse
import signal
import threading
import time
import os
import json
import logging
from flask import Flask, request, jsonify
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SyncManager:
    def __init__(self, config=None):
        self.config = config or {}
        self.sync_queue = []
        self.running = False
        self.lock = threading.Lock()
        # 注册信号处理
        signal.signal(signal.SIGUSR1, self.signal_handler)
        signal.signal(signal.SIGTERM, self.shutdown_handler)
    def signal_handler(self, signum, frame):
        """处理信号触发"""
        logger.info("收到手动同步信号 (SIGUSR1)")
        self.trigger_sync(source="signal", params={"mode": "incremental"})
    def shutdown_handler(self, signum, frame):
        """处理关闭信号"""
        logger.info("收到关闭信号")
        self.running = False
    def trigger_sync(self, source, params):
        """触发同步"""
        with self.lock:
            task = {
                'source': source,
                'params': params,
                'time': time.time()
            }
            self.sync_queue.append(task)
            logger.info(f"同步请求已加入队列: {task}")
    def process_queue(self):
        """处理同步队列"""
        while self.running:
            task = None
            with self.lock:
                if self.sync_queue:
                    task = self.sync_queue.pop(0)
            if task:
                logger.info(f"处理同步请求: {task}")
                self.execute_sync(task['params'])
            else:
                time.sleep(0.1)
    def execute_sync(self, params):
        """执行同步操作"""
        logger.info(f"执行同步: {params}")
        time.sleep(2)  # 模拟同步
        logger.info("同步完成")
    def start(self):
        """启动服务"""
        self.running = True
        logger.info("同步服务启动")
        # 启动队列处理
        self.process_thread = threading.Thread(target=self.process_queue)
        self.process_thread.start()
    def stop(self):
        """停止服务"""
        self.running = False
        logger.info("同步服务停止")
def create_app(sync_manager):
    """创建Flask应用"""
    app = Flask(__name__)
    @app.route('/sync', methods=['POST'])
    def api_trigger_sync():
        data = request.json or {}
        sync_manager.trigger_sync("api", data)
        return jsonify({"status": "accepted"})
    @app.route('/status')
    def get_status():
        return jsonify({
            "queue_size": len(sync_manager.sync_queue),
            "running": sync_manager.running
        })
    return app
def main():
    parser = argparse.ArgumentParser(description='同步服务')
    parser.add_argument('--watch-dir', help='监控目录')
    parser.add_argument('--api-port', type=int, default=5000, help='API端口')
    parser.add_argument('--daemon', action='store_true', help='作为守护进程运行')
    args = parser.parse_args()
    # 初始化同步管理器
    sync_manager = SyncManager()
    # 启动服务
    sync_manager.start()
    # 如果指定了监控目录，启动文件监控
    if args.watch_dir:
        from watchdog.observers import Observer
        observer = Observer()
        event_handler = FileSystemEventHandler()
        def on_file_change(event):
            if not event.is_directory:
                sync_manager.trigger_sync("file_watch", {
                    "path": event.src_path,
                    "event_type": event.event_type
                })
        event_handler.on_modified = on_file_change
        event_handler.on_created = on_file_change
        observer.schedule(event_handler, args.watch_dir, recursive=True)
        observer.start()
        logger.info(f"文件监控已启动: {args.watch_dir}")
    # 启动API服务
    app = create_app(sync_manager)
    # 如果需要守护进程模式
    if args.daemon:
        import daemon
        with daemon.DaemonContext():
            app.run(host='0.0.0.0', port=args.api_port)
    else:
        app.run(host='0.0.0.0', port=args.api_port)
if __name__ == "__main__":
    main()

使用建议

1. 选择合适的触发方式：

   • 命令行：适合一次性手动同步
   • 信号：适合已有进程的运行时控制
   • HTTP API：适合远程触发和集成
   • 文件监控：适合自动化场景

2. 安全性考虑：

   • API接口添加认证
   • 信号处理验证PID
   • 文件监控限制目录范围

3. 错误处理：

   • 添加重试机制
   • 记录日志
   • 发送通知

4. 并发控制：

   • 防止重复执行
   • 队列管理
   • 超时处理

根据你的具体需求选择合适的触发方式,可以组合使用多种方式以满足复杂场景。