ipam_control/scripts/seed_demo.py

"""Заполняет БД вымышленными данными из макетов IPAM Manager (через API). Запуск: venv/bin/python scripts/seed_demo.py"""
import os
import pathlib
import sys

import httpx

env = dict(l.split("=", 1) for l in pathlib.Path(__file__).resolve().parent.parent.joinpath(".env").read_text().split() if "=" in l)
base = os.environ.get("BASE_URL", f"http://127.0.0.1:{env['APP_PORT']}") + "/api/v1"
c = httpx.Client(base_url=base, timeout=30)
tok = c.post("/auth/login", json={"username": env["ADMIN_USERNAME"], "password": env["ADMIN_PASSWORD"]}).json()["access_token"]
c.headers["Authorization"] = f"Bearer {tok}"


def post(path, **body):
    r = c.post(path, json=body)
    if r.status_code >= 400:
        sys.exit(f"{path}: {r.status_code} {r.text}")
    return r.json()


orgs_data = [
    ("ООО «Технологии связи»", "ТехСвязь", "7701234567", "Москва, ул. Ленина, 15"),
    ("АО «Центр-Телеком»", "ЦентрТК", "7709876543", "СПб, пр. Невский, 100"),
    ("ИП Петров А.В.", "Петров", "770100001234", "Казань, ул. Баумана, 5"),
    ("ООО «СтройМонтаж»", "СтройМ", "7705551234", "Москва, Кутузовский, 32"),
    ("ООО «Дата-Центр»", "ДЦ", "7704443322", "Москва, Бутово, пром. зона"),
]
orgs = [post("/organizations", name=n, short_name=s, inn=i, address=a) for n, s, i, a in orgs_data]
org = orgs[0]
vrf = {v["name"]: v["id"] for v in c.get("/vrfs", params={"organization_id": org["id"]}).json()["items"]}
vrf["management"] = post("/vrfs", organization_id=org["id"], name="management", route_target="65000:2")["id"]
vrf["guest"] = post("/vrfs", organization_id=org["id"], name="guest", route_target="65000:3")["id"]

prefixes = {}
for cidr, desc, v, status, pool in [
    ("10.0.0.0/8", "Внутренняя сеть", "default", "active", False),
    ("10.10.0.0/16", "Москва — серверы и офис", "default", "active", False),
    ("10.10.1.0/24", "Серверная — production", "default", "active", True),
    ("10.10.2.0/24", "Серверная — staging", "default", "active", True),
    ("10.20.0.0/16", "Санкт-Петербург", "default", "active", False),
    ("10.20.0.0/24", "СПб — офис", "default", "active", False),
    ("192.168.0.0/16", "Офисные сети", "default", "active", False),
    ("192.168.10.0/24", "Офис — пользователи", "default", "active", False),
    ("192.168.20.0/24", "Офис — принтеры", "default", "active", False),
    ("172.16.0.0/12", "Резерв — будущее расширение", "default", "reserved", False),
    ("fd00::/8", "ULA — внутренняя IPv6", "default", "active", False),
    ("10.99.0.0/24", "Управление оборудованием", "management", "active", False),
    ("10.98.0.0/24", "Гостевой Wi-Fi", "guest", "active", False),
]:
    prefixes[cidr] = post("/prefixes", organization_id=org["id"], vrf_id=vrf[v], prefix=cidr, description=desc, status=status, is_pool=pool)["id"]

types = {t["name"]: t["id"] for t in c.get("/device-types").json()["items"]}
devs = {}
for name, t, note in [
    ("gw-prod.internal", "Сетевое оборудование", "Шлюз — MSK-Core-01"),
    ("db-master.internal", "Сервер", "PostgreSQL primary"),
    ("db-replica.internal", "Сервер", "PostgreSQL replica"),
    ("app-01.internal", "Сервер", "Backend API node 1"),
    ("monitoring.internal", "Сервер", "Prometheus + Grafana"),
    ("old-nas.internal", "Сетевое хранилище", "NAS — выведен из эксплуатации"),
]:
    devs[name] = post("/devices", name=name, device_type_id=types[t], organization_id=org["id"], note=note)["id"]


def addr(cidr, ip, **kw):
    post(f"/prefixes/{prefixes[cidr]}/addresses", address=ip, **kw)


P = "10.10.1.0/24"
named = {1: ("gw-prod.internal", "Шлюз — MSK-Core-01"), 2: ("db-master.internal", "PostgreSQL primary"), 3: ("db-replica.internal", "PostgreSQL replica"), 5: ("app-01.internal", "Backend API node 1"), 105: ("monitoring.internal", "Prometheus + Grafana")}
reserved = {4: "VIP для балансировки", 6: "Резерв", 7: "Резерв", 8: "Резерв"}
free = {100, 101, 102, 103, 104, 106, 107, 108}  # 8 свободных
n_assigned = 0
for i in range(1, 255):
    ip = f"10.10.1.{i}"
    if i in free:
        continue
    if i == 200:
        addr(P, ip, status="deprecated", dns_name="old-nas.internal", description="NAS — выведен", device_id=devs["old-nas.internal"])
    elif i in reserved:
        addr(P, ip, status="reserved", description=reserved[i])
    elif i in named:
        dns, d = named[i]
        addr(P, ip, dns_name=dns, description=d, device_id=devs[dns])
    else:
        addr(P, ip, dns_name=f"host-{i}.internal", description="Сервис")
# второй IP у monitoring — в staging-подсети
addr("10.10.2.0/24", "10.10.2.10", dns_name="monitoring.internal", description="Prometheus + Grafana", device_id=devs["monitoring.internal"])
for cidr, prefix, count in [("10.10.2.0/24", "10.10.2", 217), ("192.168.10.0/24", "192.168.10", 196), ("10.20.0.0/24", "10.20.0", 162), ("192.168.20.0/24", "192.168.20", 30)]:
    for i in range(20 if cidr == "10.10.2.0/24" else 1, count + (20 if cidr == "10.10.2.0/24" else 1)):
        addr(cidr, f"{prefix}.{i}", dns_name=f"h{i}.{prefix.replace('.', '-')}.internal")
for i in range(1, 49):
    addr("fd00::/8", f"fd00::{i:x}", description="IPv6 узел")

for name, org_i, nets, hot, contract, note in [
    ("Ростелеком", 0, ["85.249.12.0/24", "85.249.13.0/24"], "8-800-100-0800", "РТ-2024/156", "Основной канал 1G"),
    ("МТС", 1, ["213.87.0.0/16"], "8-800-250-0890", "МТС-2025/042", "Резервный канал"),
    ("Мегафон", 0, ["178.176.72.0/21", "178.176.80.0/21", "178.176.88.0/21"], "8-800-550-0500", "МФ-2024/089", "VPN филиалов"),
    ("ТТК", 4, ["195.66.128.0/19"], "8-800-775-0775", "ТТК-2025/011", "ДЦ Бутово"),
]:
    post("/isps", name=name, organization_id=orgs[org_i]["id"], networks=nets, hotline=hot, contract_number=contract, note=note)
print("seed ok")