可以通过 Ansible 提供的 Python API 来实现使用 Python 脚本调用 Ansible，并加载 `host.ini` 主机清单和 `task.yml` 任务文件，同时传递变量。以下是一个详细的实现方式： ### 示例代码 ```python import os from ansible import context from ansible.module_utils.common.collections import ImmutableDict from ansible.inventory.manager import InventoryManager from ansible.parsing.dataloader import DataLoader from ansible.playbook.play import Play from ansible.plugins.callback import CallbackBase from ansible.executor.task_queue_manager import TaskQueueManager from ansible.vars.manager import VariableManager class ResultCallback(CallbackBase): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.host_ok = {} self.host_unreachable = {} self.host_failed = {} def v2_runner_on_ok(self, result, **kwargs): self.host_ok[result._host.get_name()] = result._result def v2_runner_on_failed(self, result, **kwargs): self.host_failed[result._host.get_name()] = result._result def v2_runner_on_unreachable(self, result, **kwargs): self.host_unreachable[result._host.get_name()] = result._result def run_ansible_playbook(inventory_path, playbook_path, extra_vars=None): # 初始化数据加载器 loader = DataLoader() # 设置上下文 context.CLIARGS = ImmutableDict(tags={}, listtags=False, listtasks=False, listhosts=False, syntax=False, connection='ssh', module_path=None, forks=5, remote_user=None, private_key_file=None, ssh_common_args=None, ssh_extra_args=None, sftp_extra_args=None, scp_extra_args=None, become=None, become_method=None, become_user=None, verbosity=0, check=False, diff=False) # 加载主机清单 inventory = InventoryManager(loader=loader, sources=inventory_path) # 变量管理器 variable_manager = VariableManager(loader=loader, inventory=inventory) # 添加额外变量 if extra_vars: variable_manager.extra_vars = extra_vars # 创建回调对象 results_callback = ResultCallback() # 创建任务队列管理器 tqm = TaskQueueManager( inventory=inventory, variable_manager=variable_manager, loader=loader, passwords={}, stdout_callback=results_callback, ) try: # 创建 Play 对象 play_source = dict( name="Ansible Play", hosts='all', gather_facts='no', tasks=[ dict(action=dict(module='include_tasks', args=dict(file=playbook_path)), loop=[],), ], ) play = Play().load(play_source, variable_manager=variable_manager, loader=loader) # 执行 Play tqm.run(play=play) finally: tqm.cleanup() return results_callback # 示例：运行 playbook 并传递变量 if __name__ == "__main__": inventory_file = "host.ini" playbook_file = "task.yml" extra_variables = {"my_variable": "my_value"} # 自定义变量 results = run_ansible_playbook(inventory_file, playbook_file, extra_vars=extra_variables) print("成功主机:", results.host_ok) print("失败主机:", results.host_failed) print("不可达主机:", results.host_unreachable) ``` ### 实现说明 1. **初始化组件**： - 使用 `DataLoader` 加载数据。 - 使用 `InventoryManager` 加载主机清单文件（如 `host.ini`）。 - 使用 `VariableManager` 管理变量，包括通过 `extra_vars` 参数传递的自定义变量。 2. **执行 Playbook**： - 创建一个 `Play` 对象并指定要执行的任务。 - 使用 `TaskQueueManager` 管理任务队列，并调用 `tqm.run()` 执行任务。 3. **结果处理**： - 自定义 `ResultCallback` 类来捕获任务执行的结果，包括成功、失败和不可达的主机信息。 4. **传递变量**： - 通过 `variable_manager.extra_vars` 方法传递自定义变量，这些变量可以在 Playbook 中直接使用。 ### 注意事项 - 确保 `host.ini` 和 `task.yml` 文件路径正确。 - 根据实际需求修改 Playbook 内容或任务逻辑。 - 如果需要更复杂的配置（如连接参数），可以结合引用中提到的 `ansible_host`、`ansible_port` 等特殊变量进行设置[^2]。 ###