import os import sys from unifi_controller_api import UnifiController from unifi_controller_api.exceptions import UnifiAuthenticationError, UnifiAPIError class Db(): def __init__(self): self.devices = [] self.interfaces = [] self.ips = [] def controller_login() -> UnifiController: # try: controller = UnifiController( controller_url=os.getenv("UNIFI_URL"), username=os.getenv("UNIFI_USERNAME"), password=os.getenv("UNIFI_PASSWORD"), is_udm_pro=False, verify_ssl=True, ) # Just to check that there is a valid authentication controller.get_unifi_site(include_health=False, raw=False) return controller # except UnifiAuthenticationError: # print("Authentication failed - please check your UniFi Controller credentials and URL.") # except UnifiAPIError as e: # print(f"UniFi API error: {e}") # except Exception as e: # print(f"An unexpected error occurred: {e}") site_name=os.getenv("UNIFI_SITE") vrf=os.getenv("VRF") def collect_devices(controller: UnifiController) -> Db: db = Db() devices = [] try: devices = controller.get_unifi_site_device(site_name=site_name, detailed=True, raw=False) except UnifiAPIError as e: print(f"Error fetching device information: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") devices = sorted(devices, key=lambda d: d.name) for d in devices: print(f"{d.name}", file=sys.stderr) print(f"{d}", file=sys.stderr) db.devices.append({ "name": d.name, "mac": d.mac, "serial": d.serial, }) db.ips.append({ "address": d.ip, "status": "active", "vrf": vrf, "device": d.name, "interface": "", "is_primary": "true", }) # db.interfaces.append({ # }) return db def print_db(db: Db): def p(xs): for x in xs: print(f"---") for k, v in x.items(): print(f"{k}: {v}") p(sorted(db.devices, key=lambda d: d["name"])) p(sorted(db.interfaces, key=lambda d: d["address"])) p(sorted(db.ips, key=lambda d: d["address"])) def main(): controller = controller_login() db = collect_devices(controller) print_db(db) if __name__ == "__main__": main()