告別手動(dòng)編輯：如何用Python快速創(chuàng)建 Ansible hosts 文件？

作者：didiplus 2025-01-21 11:52:12

在實(shí)際應(yīng)用中，我們可能會(huì)有一個(gè)包含大量服務(wù)器設(shè)備臺(tái)賬信息的文件，例如ip.txt，我們希望將這些IP地址轉(zhuǎn)換為Ansible hosts清單的格式。本文將介紹如何使用Python腳本來實(shí)現(xiàn)這一目標(biāo)。

在自動(dòng)化運(yùn)維領(lǐng)域，Ansible是一款非常強(qiáng)大的工具，它可以幫助我們管理和配置大量的服務(wù)器。為了讓Ansible能夠有效地管理這些服務(wù)器，我們需要一個(gè)hosts清單文件，該文件定義了Ansible要管理的目標(biāo)主機(jī)。

一、場景描述

假設(shè)我們有一個(gè)ip.txt文件，其中包含了一系列需要管理的服務(wù)器IP地址。這些IP地址可能是分散的，也可能是連續(xù)的。我們希望將這些IP地址轉(zhuǎn)換為Ansible hosts清單的格式，以便Ansible能夠識(shí)別和管理這些服務(wù)器。具體來說，我們希望實(shí)現(xiàn)以下功能：

讀取ip.txt文件中的IP地址。
對IP地址進(jìn)行排序。
將連續(xù)的IP地址范圍合并為一個(gè)范圍表示，例如192.168.1.[100:105]。
生成Ansible hosts清單文件，其中包含每個(gè)主機(jī)的IP地址或IP地址范圍，以及相應(yīng)的連接信息，如用戶名、密碼和端口號。

二、實(shí)現(xiàn)步驟

1. 讀取IP地址

首先，我們需要編寫一個(gè)Python腳本來讀取ip.txt文件中的IP地址。我們可以使用Python的內(nèi)置函數(shù)open()來打開文件，并使用readlines()方法讀取文件的每一行。以下是讀取IP地址的代碼示例：

def read_data_from_csv(file_path):
    data = []
    with open(file_path, 'r') as file:
        reader = csv.DictReader(file)
        for row in reader:
            data.append(row)
    return data

2. 對IP地址進(jìn)行排序

讀取IP地址后，我們需要對它們進(jìn)行排序。Python的內(nèi)置函數(shù)sorted()可以很方便地對列表進(jìn)行排序。以下是對IP地址進(jìn)行排序的代碼示例：

def sort_ips_in_group(grouped_data):
    for subnet, rows in grouped_data.items():
        grouped_data[subnet] = sorted(rows, key=lambda x: ip_to_int(x['IP']))
    return grouped_data

def ip_to_int(ip):
    parts = list(map(int, ip.split('.')))
    return (parts[0] << 24) + (parts[1] << 16) + (parts[2] << 8) + parts[3]

3. 合并連續(xù)的IP地址范圍

為了將連續(xù)的IP地址范圍合并為一個(gè)范圍表示，我們可以遍歷排序后的IP地址列表，并檢查每個(gè)IP地址是否與前一個(gè)IP地址連續(xù)。如果連續(xù)，則將它們合并為一個(gè)范圍。以下是合并連續(xù)IP地址范圍的代碼示例：

def merge_ip_ranges(sorted_ips):
    merged_ranges = []
    current_range_start = None
    current_range_end = None

    for ip in sorted_ips:
        if current_range_start isNone:
            current_range_start = ip
            current_range_end = ip
        elif int(ip.split('.')[-1]) == int(current_range_end.split('.')[-1]) + 1:
            current_range_end = ip
        else:
            if current_range_start == current_range_end:
                merged_ranges.append(f"{current_range_start}")
            else:
                merged_ranges.append(f"{current_range_start.split('.')[0]}.{current_range_start.split('.')[1]}.{current_range_start.split('.')[2]}.[{current_range_start.split('.')[-1]}:{current_range_end.split('.')[-1]}]")
            current_range_start = ip
            current_range_end = ip

    # 處理最后一個(gè)范圍
    if current_range_start isnotNone:
        if current_range_start == current_range_end:
            merged_ranges.append(f"{current_range_start}")
        else:
            merged_ranges.append(f"{current_range_start.split('.')[0]}.{current_range_start.split('.')[1]}.{current_range_start.split('.')[2]}.[{current_range_start.split('.')[-1]}:{current_range_end.split('.')[-1]}]")

    return merged_ranges

4. 生成Ansible hosts清單

為了簡化IP地址管理，我們可以將連續(xù)的IP地址范圍合并成一個(gè)更大的范圍。具體來說，我們可以通過遍歷已經(jīng)排序的IP地址列表，并逐一檢查每個(gè)IP地址是否與前一個(gè)地址連續(xù)。如果發(fā)現(xiàn)它們是連續(xù)的，我們就把它們合并在一起。這樣做的好處是，不僅便于后續(xù)的管理和使用，還能讓我們以每個(gè)IP段作為組名進(jìn)行組織。最后，我們將這些合并后的IP地址范圍轉(zhuǎn)換成Ansible hosts清單的格式。Ansible hosts清單文件一般包括主機(jī)組、主機(jī)名或IP地址以及相關(guān)的連接信息等。下面是一個(gè)生成Ansible hosts清單的代碼示例：

def generate_ansible_hosts(grouped_data):
    ansible_hosts = []
    added_ips = set()  # 使用集合來存儲(chǔ)已經(jīng)添加的IP地址

    def add_ip_range(start, end):
        start_str = ip_to_str(start)
        end_str = ip_to_str(end)
        start_parts = start_str.split('.')
        end_parts = end_str.split('.')
        
        if start_parts[:3] == end_parts[:3]:
            if start == end:
                # 如果 start 和 end 相同，直接添加單個(gè) IP 地址
                ansible_hosts.append(start_str)
            else:
                # 合并相同的前三個(gè)部分
                ansible_hosts.append(f"{start_parts[0]}.{start_parts[1]}.{start_parts[2]}.[{start_parts[3]}:{end_parts[3]}]")
        else:
            ansible_hosts.append(f"{start_str}")

    for subnet, rows in grouped_data.items():
        ifnot rows:
            continue# 跳過空的子網(wǎng)

        ansible_hosts.append(f"[{subnet}]")
        current_range_start = None
        current_range_end = None

        for row in rows:
            try:
                ip = row['IP']
                ip_int = ip_to_int(ip)

                if ip_int notin added_ips:  # 檢查IP地址是否已經(jīng)添加過
                    if current_range_start isNone:
                        current_range_start = ip_int
                        current_range_end = ip_int
                    elif ip_int == current_range_end + 1:
                        current_range_end = ip_int
                    else:
                        add_ip_range(current_range_start, current_range_end)
                        current_range_start = ip_int
                        current_range_end = ip_int
                    added_ips.add(ip_int)  # 將IP地址添加到集合中
                elif current_range_start isNoneornot (current_range_start <= ip_int <= current_range_end):
                    ansible_hosts.append(ip)

            except KeyError as e:
                print(f"Missing key in row data: {e}")
                continue

        if current_range_start isnotNone:
            add_ip_range(current_range_start, current_range_end)

        ansible_hosts.append(f"[{subnet}:vars]")
        try:
            ansible_hosts.append(f"ansible_ssh_user={rows[0]['Username']}")
            ansible_hosts.append(f"ansible_ssh_pass={rows[0]['Password']}")
            ansible_hosts.append(f"ansible_ssh_port={rows[0]['Port']}")
        except (KeyError, IndexError) as e:
            print(f"Error accessing vars for subnet {subnet}: {e}")
            continue

        ansible_hosts.append("")

    return ansible_hosts

三、腳本使用演示

首先，通過一個(gè)腳本生成一下測試IP數(shù)據(jù)，腳本如下：

import random

def generate_random_ip(subnet):
    """生成一個(gè)隨機(jī)的IP地址"""
    parts = subnet.split('.')
    if len(parts) < 4:
        parts.extend(['0'] * (4 - len(parts)))
    parts[3] = str(random.randint(1, 254))
    return'.'.join(parts)

def generate_test_data(file_path, num_records):
    """生成測試數(shù)據(jù)并寫入CSV文件"""
    with open(file_path, 'w') as file:
        file.write("IP,Port,Username,Password\n")
        for _ in range(num_records):
            subnet = random.choice(['192.168.31', '192.168.32', '192.168.33', '192.168.34'])
            ip = generate_random_ip(subnet)
            port = 10022
            if subnet == "192.168.31":
                username = "root"
                password = "pass@123"
            elif subnet == "192.168.33":
                username = "root"
                password = "P@ssw0rd"
            else:
                username = "admin"
                password = "password"
            file.write(f"{ip},{port},{username},{password}\n")

# 示例文件路徑
file_path = 'test_data.csv'
num_records = 800

# 生成測試數(shù)據(jù)并寫入文件
generate_test_data(file_path, num_records)

執(zhí)行上述腳本后，會(huì)在當(dāng)前目錄生成一個(gè)IP數(shù)據(jù)，這些IP數(shù)據(jù)，有連續(xù)的和不連續(xù)的。如下圖所示：

然后，通過執(zhí)行g(shù)enerate_ansible_hosts.py即可生成如下圖的結(jié)果：

四、總結(jié)

通過以上步驟，我們成功地使用Python腳本將ip.txt文件轉(zhuǎn)換為Ansible hosts清單。這個(gè)腳本可以幫助我們自動(dòng)化生成Ansible hosts清單，提高運(yùn)維效率。在實(shí)際應(yīng)用中，我們可以根據(jù)需要對腳本進(jìn)行進(jìn)一步的擴(kuò)展和優(yōu)化，例如支持更多的連接信息、處理不同格式的IP地址等。希望本文對你有所幫助，如果你有任何問題或建議，歡迎留言討論。