pip3 install ansible
pip3 install ansible_runner
pip3 install ansible_inventory
pip3 install ansible_playbook

[root@dev-technology-215l shell]# cat ansible2.py   
import json  
import shutil  
from ansible.module_utils.common.collections import ImmutableDict  
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  
import ansible.constants as C
class ResultCallback(CallbackBase):  
    """  
    重寫callbackBase類的部分方法  
    """  
    def __init__(self, *args, **kwargs):  
        super().__init__(*args, **kwargs)  
        self.host_ok = {}  
        self.host_unreachable = {}  
        self.host_failed = {}  
        self.task_ok = {}  
    def v2_runner_on_unreachable(self, result):  
        self.host_unreachable[result._host.get_name()] = result
    def v2_runner_on_ok(self, result, **kwargs):  
        self.host_ok[result._host.get_name()] = result
    def v2_runner_on_failed(self, result, **kwargs):  
        self.host_failed[result._host.get_name()] = result
class MyAnsiable2():  
    def __init__(self,  
        connection='local',  # 連接方式 local 本地方式，smart ssh方式  
        remote_user=None,    # ssh 用戶  
        remote_password=None,  # ssh 用戶的密碼，應該是一個字典, key 必須是 conn_pass  
        private_key_file=None,  # 指定自定義的私鑰地址  
        sudo=None, sudo_user=None, ask_sudo_pass=None,  
        module_path=None,    # 模塊路徑，可以指定一個自定義模塊的路徑  
        become=None,         # 是否提權  
        become_method=None,  # 提權方式 默認 sudo 可以是 su  
        become_user=None,  # 提權后，要成為的用戶，并非登錄用戶  
        check=False, diff=False,  
        listhosts=None, listtasks=None,listtags=None,  
        verbosity=3,  
        syntax=None,  
        start_at_task=None,  
        inventory=None):
        # 函數(shù)文檔注釋  
        """  
        初始化函數(shù)，定義的默認的選項值，  
        在初始化的時候可以傳參，以便覆蓋默認選項的值  
        """  
        context.CLIARGS = ImmutableDict(  
            connection=connection,  
            remote_user=remote_user,  
            private_key_file=private_key_file,  
            sudo=sudo,  
            sudo_user=sudo_user,  
            ask_sudo_pass=ask_sudo_pass,  
            module_path=module_path,  
            become=become,  
            become_method=become_method,  
            become_user=become_user,  
            verbosity=verbosity,  
            listhosts=listhosts,  
            listtasks=listtasks,  
            listtags=listtags,  
            syntax=syntax,  
            start_at_task=start_at_task,  
        )
        # 三元表達式，假如沒有傳遞 inventory, 就使用 "localhost,"  
        # 指定 inventory 文件  
        # inventory 的值可以是一個 資產(chǎn)清單文件  
        # 也可以是一個包含主機的元組，這個僅僅適用于測試  
        #  比如 ： 1.1.1.1,    # 如果只有一個 IP 最后必須有英文的逗號  
        #  或者： 1.1.1.1, 2.2.2.2
        self.inventory = inventory if inventory else "localhost,"
        # 實例化數(shù)據(jù)解析器  
        self.loader = DataLoader()
        # 實例化 資產(chǎn)配置對象  
        self.inv_obj = InventoryManager(loader=self.loader, sources=self.inventory)
        # 設置密碼  
        self.passwords = remote_password
        # 實例化回調(diào)插件對象  
        self.results_callback = ResultCallback()
        # 變量管理器  
        self.variable_manager = VariableManager(self.loader, self.inv_obj)
    def run(self, hosts='localhost', gether_facts="no", module="ping", args='', task_time=0):  
        """  
        參數(shù)說明：  
        task_time -- 執(zhí)行異步任務時等待的秒數(shù)，這個需要大于 0 ，等于 0 的時候不支持異步（默認值）。這個值應該等于執(zhí)行任務實際耗時時間為好  
        """  
        play_source =  dict(  
            name = "Ad-hoc",  
            hosts = hosts,  
            gather_facts = gether_facts,  
            tasks = [  
                # 這里每個 task 就是這個列表中的一個元素，格式是嵌套的字典  
                # 也可以作為參數(shù)傳遞過來，這里就簡單化了。  
               {"action":{"module": module, "args": args}, "async": task_time, "poll": 0}])
        play = Play().load(play_source, variable_manager=self.variable_manager, loader=self.loader)
        tqm = None  
        try:  
            tqm = TaskQueueManager(  
                      inventory=self.inv_obj ,  
                      variable_manager=self.variable_manager,  
                      loader=self.loader,  
                      passwords=self.passwords,  
                      stdout_callback=self.results_callback)
            result = tqm.run(play)  
        finally:  
            if tqm is not None:  
                tqm.cleanup()  
            shutil.rmtree(C.DEFAULT_LOCAL_TMP, True)
    def playbook(self,playbooks):  
        """  
        Keyword arguments:  
        playbooks --  需要是一個列表類型  
        """  
        from ansible.executor.playbook_executor import PlaybookExecutor
        playbook = PlaybookExecutor(playbooks=playbooks,  
                        inventory=self.inv_obj,  
                        variable_manager=self.variable_manager,  
                        loader=self.loader,  
                        passwords=self.passwords)
        # 使用回調(diào)函數(shù)  
        playbook._tqm._stdout_callback = self.results_callback
        result = playbook.run()
    def get_result(self):  
      result_raw = {'success':{},'failed':{},'unreachable':{}}
      # print(self.results_callback.host_ok)  
      for host,result in self.results_callback.host_ok.items():  
          result_raw['success'][host] = result._result  
      for host,result in self.results_callback.host_failed.items():  
          result_raw['failed'][host] = result._result  
      for host,result in self.results_callback.host_unreachable.items():  
          result_raw['unreachable'][host] = result._result
      # 最終打印結(jié)果，并且使用 JSON 繼續(xù)格式化  
      print(json.dumps(result_raw, indent=4))  
      return json.dumps(result_raw)

[root@dev-technology-215l shell]# cat ansible_run_wc.py   
from ansible2 import *   #引用修改過的ansible2.py 的所有模塊  
import ansible_runner  
import subprocess  
import os, sys, json, datetime, time
ansible3 = MyAnsiable2(inventory='/data/ansible/host/hosts', connection='smart')      #創(chuàng)建資源庫對象
ansible3.run(hosts= "192.168.0.94", module="shell", args='pm2 ls | grep sk_service_design_calculate_rda | wc -l')
stdout_dict = json.loads(ansible3.get_result())  
print(stdout_dict,type(stdout_dict))  
print(stdout_dict['success']['192.168.0.94']['stdout'],'######wc')

[root@dev-technology-215l shell]# cat ansible_playbook_test.py 
from ansible2 import *
import ansible_runner
import subprocess
import os, sys, json, datetime, time
import urllib.request
ansible3 = MyAnsiable2(inventory='/data/ansible/host/hosts', connection='smart')
ansible3.playbook(playbooks=['test.yml'])
stdout_dict = json.loads(ansible3.get_result())
print(stdout_dict,type(stdout_dict))
#print(stdout_dict['success']['192.168.0.94']['stdout'])

[root@dev-technology-215l shell]# grep -C5 192.168.0.94 /data/ansible/host/hosts 
###rda-環(huán)境node高負載應用linux標準A組###
[rda-highload-node-service-linux-stdA]
192.168.0.94 ansible_ssh_user=root ansible_ssh_pass="Sxxx-xxxx"

- name: test server
  hosts: 'rda-highload-node-service-linux-stdA'
  tasks:
        - name: online pm2 ls
          raw: ls /data/