Python脚本如何独立执行模块同步任务

wen python案例 2026-06-25 31

本文目录导读：

模式">使用 if __name__ == "__main__" 模式
使用命令行参数解析（更适合复杂任务）
使用配置文件和定时执行
使用进程池并行执行
使用装饰器模式简化任务定义
使用方法示例：

我来介绍几种让Python脚本独立执行模块同步任务的方法：

使用 `if name == "main"` 模式

这是最基础的方式,让模块既可以导入又可以独立运行：

# sync_module.py
import time
import logging
logging.basicConfig(level=logging.INFO)
def sync_data(source, target):
    """同步数据的主函数"""
    logging.info(f"开始同步从 {source} 到 {target}")
    # 模拟同步过程
    time.sleep(2)
    logging.info("同步完成")
    return True
class SyncTask:
    def __init__(self, name):
        self.name = name
    def run(self):
        logging.info(f"执行同步任务: {self.name}")
        # 执行同步逻辑
        return sync_data("source", "target")
# 独立执行时的入口
if __name__ == "__main__":
    import sys
    task = SyncTask(sys.argv[1] if len(sys.argv) > 1 else "default")
    task.run()

使用命令行参数解析（更适合复杂任务）

# advanced_sync.py
import argparse
import logging
import time
from pathlib import Path
logging.basicConfig(level=logging.INFO, 
                   format='%(asctime)s - %(levelname)s - %(message)s')
class FileSync:
    """文件同步器"""
    def __init__(self, source, target, recursive=False, dry_run=False):
        self.source = Path(source)
        self.target = Path(target)
        self.recursive = recursive
        self.dry_run = dry_run
    def sync(self):
        """执行同步"""
        if not self.source.exists():
            logging.error(f"源路径不存在: {self.source}")
            return False
        logging.info(f"开始同步: {self.source} -> {self.target}")
        if self.dry_run:
            logging.info("dry-run模式，仅模拟同步")
            return True
        # 实际的同步逻辑
        self._copy_files(self.source, self.target)
        return True
    def _copy_files(self, src, dst):
        """递归复制文件"""
        if src.is_file():
            if not dst.parent.exists():
                dst.parent.mkdir(parents=True)
            # 这里添加实际的文件复制逻辑
            time.sleep(0.1)
        elif src.is_dir() and self.recursive:
            for item in src.iterdir():
                self._copy_files(item, dst / item.name)
def main():
    parser = argparse.ArgumentParser(description='文件同步工具')
    parser.add_argument('source', help='源路径')
    parser.add_argument('target', help='目标路径')
    parser.add_argument('-r', '--recursive', action='store_true', 
                       help='递归同步子目录')
    parser.add_argument('--dry-run', action='store_true',
                       help='模拟运行，不实际执行')
    parser.add_argument('--log-level', default='INFO',
                       choices=['DEBUG', 'INFO', 'WARNING', 'ERROR'])
    args = parser.parse_args()
    # 设置日志级别
    logging.getLogger().setLevel(getattr(logging, args.log_level))
    syncer = FileSync(args.source, args.target, 
                      args.recursive, args.dry_run)
    success = syncer.sync()
    return 0 if success else 1
if __name__ == "__main__":
    exit(main())

使用配置文件和定时执行

# scheduled_sync.py
import json
import schedule
import time
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO)
class TaskConfig:
    def __init__(self, config_file):
        with open(config_file, 'r') as f:
            self.config = json.load(f)
    def get_tasks(self):
        return self.config.get('tasks', [])
class TaskRunner:
    def __init__(self):
        self.tasks = []
    def add_task(self, name, func, schedule_time):
        """添加定时任务"""
        self.tasks.append({
            'name': name,
            'func': func,
            'schedule': schedule_time
        })
        # 设置定时执行
        schedule.every().day.at(schedule_time).do(self._run_task, name, func)
        logging.info(f"已添加定时任务: {name} 于 {schedule_time}")
    def _run_task(self, name, func):
        """执行单个任务"""
        logging.info(f"开始执行定时任务: {name} at {datetime.now()}")
        try:
            result = func()
            logging.info(f"任务 {name} 完成: {result}")
        except Exception as e:
            logging.error(f"任务 {name} 失败: {e}")
    def run_forever(self):
        """持续运行"""
        logging.info("开始定时任务调度...")
        while True:
            schedule.run_pending()
            time.sleep(1)
# 具体的同步函数
def sync_database():
    """同步数据库"""
    logging.info("同步数据库...")
    # 实际的数据库同步逻辑
    return True
def sync_files():
    """同步文件"""
    logging.info("同步文件...")
    # 实际的文件同步逻辑
    return True
if __name__ == "__main__":
    # 创建任务运行器
    runner = TaskRunner()
    # 添加定时任务
    runner.add_task("数据库同步", sync_database, "01:00")  # 每天凌晨1点
    runner.add_task("文件同步", sync_files, "02:00")      # 每天凌晨2点
    # 也可以立即执行一次
    if len(sys.argv) > 1 and sys.argv[1] == "--now":
        logging.info("立即执行所有任务")
        for task in runner.tasks:
            task['func']()
    else:
        runner.run_forever()

使用进程池并行执行

# parallel_sync.py
import multiprocessing
import time
import logging
from typing import List, Dict
logging.basicConfig(level=logging.INFO)
class SyncWorker:
    """同步工作器"""
    def __init__(self, worker_id):
        self.worker_id = worker_id
    def sync_module(self, module_name):
        """同步单个模块"""
        logging.info(f"Worker {self.worker_id}: 开始同步模块 {module_name}")
        # 模拟同步过程
        time.sleep(2)
        logging.info(f"Worker {self.worker_id}: 完成模块 {module_name}")
        return f"{module_name} 同步成功"
def sync_all_modules(modules: List[str], max_workers: int = 4):
    """并行同步所有模块"""
    with multiprocessing.Pool(max_workers) as pool:
        workers = [SyncWorker(i) for i in range(max_workers)]
        # 分配任务
        results = []
        for i, module in enumerate(modules):
            worker = workers[i % max_workers]
            result = pool.apply_async(worker.sync_module, (module,))
            results.append(result)
        # 收集结果
        for result in results:
            try:
                output = result.get(timeout=30)
                logging.info(f"结果: {output}")
            except Exception as e:
                logging.error(f"同步失败: {e}")
if __name__ == "__main__":
    modules = [
        "用户模块",
        "订单模块", 
        "产品模块",
        "支付模块",
        "库存模块",
        "物流模块",
        "报表模块",
        "配置模块"
    ]
    import sys
    max_workers = int(sys.argv[1]) if len(sys.argv) > 1 else 4
    start_time = time.time()
    sync_all_modules(modules, max_workers)
    duration = time.time() - start_time
    logging.info(f"全部同步完成，耗时: {duration:.2f}秒")

使用装饰器模式简化任务定义

# decorator_sync.py
import functools
import logging
import time
logging.basicConfig(level=logging.INFO)
class SyncTaskRegistry:
    """同步任务注册器"""
    _tasks = {}
    @classmethod
    def register(cls, name=None):
        """注册任务的装饰器"""
        def decorator(func):
            task_name = name or func.__name__
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                logging.info(f"开始执行: {task_name}")
                start_time = time.time()
                try:
                    result = func(*args, **kwargs)
                    duration = time.time() - start_time
                    logging.info(f"完成 {task_name}, 耗时: {duration:.2f}秒")
                    return result
                except Exception as e:
                    logging.error(f"{task_name} 失败: {e}")
                    raise
            cls._tasks[task_name] = wrapper
            return wrapper
        return decorator
    @classmethod
    def run_all(cls):
        """运行所有注册的任务"""
        results = {}
        for name, task in cls._tasks.items():
            try:
                results[name] = task()
            except Exception as e:
                results[name] = str(e)
        return results
    @classmethod
    def run_specific(cls, *names):
        """运行指定任务"""
        results = {}
        for name in names:
            if name in cls._tasks:
                try:
                    results[name] = cls._tasks[name]()
                except Exception as e:
                    results[name] = str(e)
            else:
                logging.warning(f"任务 {name} 不存在")
        return results
# 定义同步任务
@SyncTaskRegistry.register("用户数据同步")
def sync_users():
    time.sleep(1)
    return "同步了1000个用户"
@SyncTaskRegistry.register("订单同步")
def sync_orders():
    time.sleep(2)
    return "同步了500个订单"
@SyncTaskRegistry.register("产品数据同步")
def sync_products():
    time.sleep(1.5)
    return "同步了200个产品"
if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1:
        # 执行指定的任务
        task_names = sys.argv[1:]
        results = SyncTaskRegistry.run_specific(*task_names)
    else:
        # 执行所有任务
        results = SyncTaskRegistry.run_all()
    for task, result in results.items():
        print(f"任务: {task} -> 结果: {result}")

使用方法示例：

# 直接运行
python sync_module.py
python sync_module.py my_task
# 带参数运行
python advanced_sync.py /source/path /target/path -r --dry-run
# 定时运行
python scheduled_sync.py
python scheduled_sync.py --now
# 并行运行
python parallel_sync.py 4  # 使用4个进程
# 选择性运行
python decorator_sync.py "用户数据同步" "订单同步"

这些方法都能让Python脚本独立执行模块同步任务,你可以根据实际需求选择合适的方式。