import json
from collections import namedtuple
from ansible.parsing.dataloader import DataLoader
from ansible.vars.manager import VariableManager
from ansible.inventory.manager import InventoryManager
from ansible.playbook.play import Play
from ansible.executor.task_queue_manager import TaskQueueManager
from ansible.plugins.callback import CallbackBase

from ansible import context
from ansible.module_utils.common.collections import ImmutableDict

#InventoryManager类
loader = DataLoader() #dataloader用于读取yaml\json格式的文件
inventory = InventoryManager(loader=loader,sources=['/Users/xueersi/PycharmProjects/dispatchingCenter/conf/x_hosts'])

#VariableManager类
variable_manager = VariableManager(loader=loader,inventory=inventory)

context.CLIARGS = ImmutableDict(connection='ssh',
                      remote_user=None,
                      ack_pass=None,
                      sudo_user=None,
                      forks=5,
                      sudo=None,
                      ask_sudo_pass=False,
                      verbosity=5,
                      module_path=None,
                      become=None,
                      become_method=None,
                      become_user=None,
                      check=False,
                      diff=False,
                      listhosts=None,
                      listtasks=None,
                      listtags=None,
                      syntax=None)

#Options 执行选项
Options = namedtuple('Options',
                    ['connection',
                     'remote_user',
                     'ask_sudo_pass',
                     'verbosity',
                     'ack_pass',
                     'module_path',
                     'forks',
                     'become',
                     'become_method',
                     'become_user',
                     'check',
                     'listhosts',
                     'listtasks',
                     'listtags',
                     'syntax',
                     'sudo_user',
                     'sudo',
                     'diff'])
options = Options(connection='ssh',
                      remote_user=None,
                      ack_pass=None,
                      sudo_user=None,
                      forks=5,
                      sudo=None,
                      ask_sudo_pass=False,
                      verbosity=5,
                      module_path=None,
                      become=None,
                      become_method=None,
                      become_user=None,
                      check=False,
                      diff=False,
                      listhosts=None,
                      listtasks=None,
                      listtags=None,
                      syntax=None)
#Play 执行对象和模块
play_source =  dict(
       name = "Ansible Play ad-hoc test",
       hosts = '10.26.84.29',
       # hosts = '10.26.84.12',
       # hosts = '10.26.234.32',
       gather_facts = 'no', #获取配置选项的基本信息，设置为no可以加快执行速度
       tasks = [
           dict(action=dict(module='shell', args='touch /tmp/ad_hoc_test2')),
           # dict(action=dict(module='win_ping', args='')),
           # dict(action=dict(module='debug', args=dict(msg='{{shell_out.stdout}}')))
        ]
   )
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)

class ModelResultsCollector(CallbackBase):
   """
   重写callbackBase类的部分方法
   """
   def __init__(self, *args, **kwargs):
       super(ModelResultsCollector, self).__init__(*args, **kwargs)
       self.host_ok = {}
       self.host_unreachable = {}
       self.host_failed = {}
   def v2_runner_on_unreachable(self, result):
       self.host_unreachable[result._host.get_name()] = result
   def v2_runner_on_ok(self, result):
       self.host_ok[result._host.get_name()] = result
   def v2_runner_on_failed(self, result):
       self.host_failed[result._host.get_name()] = result

callback = ModelResultsCollector()

passwords = dict()
tqm = TaskQueueManager(
         inventory=inventory,
         variable_manager=variable_manager,
         loader=loader,
         options=options,
         passwords=passwords,
         # stdout_callback=callback,
     )
result = tqm.run(play)

print(callback.host_ok.items())
# print(type(callback.host_ok.items()))

# result_raw = {'success':{},'failed':{},'unreachable':{}}
result_raw = dict(success={}, failed={}, unreachable={})
for host,result in callback.host_ok.items():
   result_raw['success'][host] = result._result
for host,result in callback.host_failed.items():
   result_raw['failed'][host] = result._result
for host,result in callback.host_unreachable.items():
   result_raw['failed'][host] = result._result

print(json.dumps(result_raw))

返回结果