告別手動編輯：如何用Python快速創建 Ansible hosts 文件？
在自動化運維領域，Ansible是一款非常強大的工具，它可以幫助我們管理和配置大量的服務器。為了讓Ansible能夠有效地管理這些服務器，我們需要一個hosts清單文件，該文件定義了Ansible要管理的目標主機。
在實際應用中，我們可能會有一個包含大量服務器設備臺賬信息的文件，例如ip.txt，我們希望將這些IP地址轉換為Ansible hosts清單的格式。本文將介紹如何使用Python腳本來實現這一目標。
一、場景描述
假設我們有一個ip.txt文件，其中包含了一系列需要管理的服務器IP地址。這些IP地址可能是分散的，也可能是連續的。我們希望將這些IP地址轉換為Ansible hosts清單的格式，以便Ansible能夠識別和管理這些服務器。具體來說，我們希望實現以下功能：
- 讀取ip.txt文件中的IP地址。
- 對IP地址進行排序。
- 將連續的IP地址范圍合并為一個范圍表示，例如192.168.1.[100:105]。
- 生成Ansible hosts清單文件，其中包含每個主機的IP地址或IP地址范圍，以及相應的連接信息，如用戶名、密碼和端口號。
二、實現步驟
1. 讀取IP地址
首先，我們需要編寫一個Python腳本來讀取ip.txt文件中的IP地址。我們可以使用Python的內置函數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 data2. 對IP地址進行排序
讀取IP地址后，我們需要對它們進行排序。Python的內置函數sorted()可以很方便地對列表進行排序。以下是對IP地址進行排序的代碼示例：
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. 合并連續的IP地址范圍
為了將連續的IP地址范圍合并為一個范圍表示，我們可以遍歷排序后的IP地址列表，并檢查每個IP地址是否與前一個IP地址連續。如果連續，則將它們合并為一個范圍。以下是合并連續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
# 處理最后一個范圍
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_ranges4. 生成Ansible hosts清單
為了簡化IP地址管理，我們可以將連續的IP地址范圍合并成一個更大的范圍。具體來說，我們可以通過遍歷已經排序的IP地址列表，并逐一檢查每個IP地址是否與前一個地址連續。如果發現它們是連續的，我們就把它們合并在一起。這樣做的好處是，不僅便于后續的管理和使用，還能讓我們以每個IP段作為組名進行組織。最后，我們將這些合并后的IP地址范圍轉換成Ansible hosts清單的格式。Ansible hosts清單文件一般包括主機組、主機名或IP地址以及相關的連接信息等。下面是一個生成Ansible hosts清單的代碼示例：
def generate_ansible_hosts(grouped_data):
ansible_hosts = []
added_ips = set() # 使用集合來存儲已經添加的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 相同，直接添加單個 IP 地址
ansible_hosts.append(start_str)
else:
# 合并相同的前三個部分
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# 跳過空的子網
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地址是否已經添加過
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三、腳本使用演示
首先，通過一個腳本生成一下測試IP數據，腳本如下：
import random
def generate_random_ip(subnet):
"""生成一個隨機的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):
"""生成測試數據并寫入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
# 生成測試數據并寫入文件
generate_test_data(file_path, num_records)執行上述腳本后，會在當前目錄生成一個IP數據，這些IP數據，有連續的和不連續的。如下圖所示：
然后，通過執行generate_ansible_hosts.py即可生成如下圖的結果：
四、總結
通過以上步驟，我們成功地使用Python腳本將ip.txt文件轉換為Ansible hosts清單。這個腳本可以幫助我們自動化生成Ansible hosts清單，提高運維效率。在實際應用中，我們可以根據需要對腳本進行進一步的擴展和優化，例如支持更多的連接信息、處理不同格式的IP地址等。希望本文對你有所幫助，如果你有任何問題或建議，歡迎留言討論。
五、腳本獲取方式
上述腳本已經上傳上傳到gitee,有需要的小伙伴可以自行獲取。gitee上的倉庫主要是分享一些工作中常用的腳本。小伙伴可以frok或者watch倉庫，這樣有更新可以及時關注到。
倉庫地址：https://gitee.com/didiplus/script
