Python脚本如何生成ansible playbook

wen 实用脚本 2026-07-09 169

本文目录导读：

使用YAML库直接生成（推荐）
使用Jinja2模板生成
动态生成复杂Playbook
从JSON/CSV导入生成

我来介绍几种用Python生成Ansible Playbook的方法：

使用YAML库直接生成（推荐）

import yaml
def generate_playbook(hosts, tasks, name="Deploy Application"):
    playbook = {
        'name': name,
        'hosts': hosts,
        'become': True,
        'vars': {
            'app_name': 'myapp',
            'app_port': 8080
        },
        'tasks': tasks
    }
    return playbook
# 生成包含多个play的playbook
def generate_multi_play_playbook():
    playbooks = []
    # Play 1: 安装nginx
    play1 = {
        'name': 'Install and configure nginx',
        'hosts': 'webservers',
        'become': True,
        'tasks': [
            {
                'name': 'Install nginx',
                'apt': {
                    'name': 'nginx',
                    'state': 'present',
                    'update_cache': True
                }
            },
            {
                'name': 'Start nginx',
                'service': {
                    'name': 'nginx',
                    'state': 'started',
                    'enabled': True
                }
            }
        ]
    }
    playbooks.append(play1)
    # Play 2: 部署应用
    play2 = {
        'name': 'Deploy application',
        'hosts': 'appservers',
        'become': True,
        'tasks': [
            {
                'name': 'Create app directory',
                'file': {
                    'path': '/opt/myapp',
                    'state': 'directory',
                    'mode': '0755'
                }
            },
            {
                'name': 'Copy app config',
                'template': {
                    'src': 'app.conf.j2',
                    'dest': '/opt/myapp/app.conf'
                }
            }
        ]
    }
    playbooks.append(play2)
    return playbooks
# 保存为YAML文件
def save_playbook(playbook_data, filename='deploy.yml'):
    # 处理Ansible的特殊标签
    class AnsibleDumper(yaml.Dumper):
        pass
    def represent_none(self, _):
        return self.represent_scalar('tag:yaml.org,2002:null', 'None')
    AnsibleDumper.add_representer(type(None), represent_none)
    with open(filename, 'w', encoding='utf-8') as f:
        if isinstance(playbook_data, list):
            yaml.dump(playbook_data, f, Dumper=AnsibleDumper, default_flow_style=False, allow_unicode=True)
        else:
            yaml.dump([playbook_data], f, Dumper=AnsibleDumper, default_flow_style=False, allow_unicode=True)
# 使用示例
if __name__ == "__main__":
    tasks = [
        {
            'name': 'Ensure package is installed',
            'apt': {
                'name': 'git',
                'state': 'present'
            }
        },
        {
            'name': 'Clone repository',
            'git': {
                'repo': 'https://github.com/example/repo.git',
                'dest': '/opt/app'
            }
        }
    ]
    playbook = generate_playbook('webservers', tasks)
    save_playbook(playbook)

使用Jinja2模板生成

from jinja2 import Environment, FileSystemLoader
import os
# 模板文件: playbook_template.j2
jinja_template = """
---
- name: {{ play_name }}
  hosts: {{ hosts }}
  become: {{ become | default(true) }}
  vars:
    app_name: "{{ app_name }}"
    app_port: {{ app_port }}
  tasks:
    {% for task in tasks %}
    - name: {{ task.name }}
      {{ task.module }}:
        {% for key, value in task.args.items() %}
        {{ key }}: "{{ value }}"
        {% endfor %}
      {% if task.notify is defined %}
      notify: {{ task.notify }}
      {% endif %}
    {% endfor %}
  handlers:
    {% for handler in handlers %}
    - name: {{ handler.name }}
      {{ handler.module }}:
        {% for key, value in handler.args.items() %}
        {{ key }}: "{{ value }}"
        {% endfor %}
    {% endfor %}
"""
def generate_playbook_with_jinja2(playbook_data, template_str):
    env = Environment()
    template = env.from_string(template_str)
    return template.render(**playbook_data)
# 使用示例
handlers = [
    {
        'name': 'Restart nginx',
        'module': 'service',
        'args': {
            'name': 'nginx',
            'state': 'restarted'
        }
    }
]
tasks = [
    {
        'name': 'Install nginx',
        'module': 'apt',
        'args': {
            'name': 'nginx',
            'state': 'present'
        },
        'notify': 'Restart nginx'
    }
]
playbook_content = generate_playbook_with_jinja2({
    'play_name': 'WebServer Setup',
    'hosts': 'webservers',
    'become': True,
    'app_name': 'myapp',
    'app_port': 8080,
    'tasks': tasks,
    'handlers': handlers
})
# 保存文件
with open('playbook.yml', 'w') as f:
    f.write(playbook_content)

动态生成复杂Playbook

import yaml
from datetime import datetime
class PlaybookGenerator:
    def __init__(self):
        self.plays = []
    def add_play(self, name, hosts, become=True, vars=None, tasks=None, handlers=None):
        play = {
            'name': name,
            'hosts': hosts,
            'become': become
        }
        if vars:
            play['vars'] = vars
        if tasks:
            play['tasks'] = tasks
        if handlers:
            play['handlers'] = handlers
        self.plays.append(play)
    def add_task(self, play_index, name, module, args, notify=None):
        task = {
            'name': name,
            module: args
        }
        if notify:
            task['notify'] = notify
        if play_index < len(self.plays):
            if 'tasks' not in self.plays[play_index]:
                self.plays[play_index]['tasks'] = []
            self.plays[play_index]['tasks'].append(task)
    def add_role(self, play_index, role_name):
        if play_index < len(self.plays):
            if 'roles' not in self.plays[play_index]:
                self.plays[play_index]['roles'] = []
            self.plays[play_index]['roles'].append(role_name)
    def save(self, filename=None):
        if not filename:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"playbook_{timestamp}.yml"
        with open(filename, 'w') as f:
            yaml.dump(self.plays, f, default_flow_style=False, allow_unicode=True)
        return filename
# 使用示例
generator = PlaybookGenerator()
# Play 1: Web服务器配置
generator.add_play(
    name="Configure Web Servers",
    hosts="webservers",
    vars={
        'http_port': 80,
        'max_clients': 200
    }
)
generator.add_task(
    0,
    name="Install nginx",
    module="apt",
    args={'name': 'nginx', 'state': 'present'},
    notify="Restart nginx"
)
generator.add_task(
    0,
    name="Configure nginx",
    module="template",
    args={
        'src': 'nginx.conf.j2',
        'dest': '/etc/nginx/nginx.conf'
    },
    notify="Restart nginx"
)
generator.add_role(0, "security")
# Play 2: 数据库配置
generator.add_play(
    name="Configure Database",
    hosts="dbservers",
    become=False
)
generator.add_task(
    1,
    name="Install PostgreSQL",
    module="apt",
    args={'name': 'postgresql', 'state': 'present'}
)
# 保存playbook
filename = generator.save()
print(f"Playbook saved as: {filename}")

从JSON/CSV导入生成

import yaml
import json
import csv
import os
def generate_playbook_from_json(json_file):
    with open(json_file, 'r') as f:
        data = json.load(f)
    playbook = []
    for play_data in data:
        play = {
            'name': play_data.get('name', 'Default Play'),
            'hosts': play_data.get('hosts', 'all'),
            'become': play_data.get('become', True),
            'tasks': []
        }
        for task_data in play_data.get('tasks', []):
            task = {
                'name': task_data['name'],
                task_data['module']: task_data['args']
            }
            play['tasks'].append(task)
        playbook.append(play)
    return playbook
def generate_playbook_from_csv(csv_file):
    with open(csv_file, 'r') as f:
        reader = csv.DictReader(f)
        tasks = []
        for row in reader:
            task = {
                'name': row['task_name'],
                row['module']: {
                    'name': row['package_name'],
                    'state': row.get('state', 'present')
                }
            }
            tasks.append(task)
    playbook = [{
        'name': 'Generated from CSV',
        'hosts': 'all',
        'become': True,
        'tasks': tasks
    }]
    return playbook
# 使用示例
def generate_from_csv_example():
    csv_data = """task_name,module,package_name,state
Install git,apt,git,present
Install curl,apt,curl,present
Install vim,apt,vim,present"""
    with open('packages.csv', 'w') as f:
        f.write(csv_data)
    playbook = generate_playbook_from_csv('packages.csv')
    with open('from_csv_playbook.yml', 'w') as f:
        yaml.dump(playbook, f, default_flow_style=False)
    os.remove('packages.csv')

这些方法各有特点：