ITEAmplify commented 2 months ago

Hey,

I know my Problem is not From the PLugin itself but i cant find the mistake. I Upgraded to Version 4.0.1 today and im used to search Free Asset Tags for the Netbox_Inventory Plugin with an Custom Report. Now i Migrated it to an Custom Script as Reports are now Depricated and now it dosnt work anymore. Does someone has a hint for me? (Sorry for the bad programming, im not used to python)

##!/opt/netbox/venv/bin/python
#import django, os, sys
#sys.path.append('/opt/netbox/netbox')
#os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'netbox.settings')
#django.setup()
from extras.choices import JobResultStatusChoices
from extras.scripts import Script
from dcim.models import Device
from collections import OrderedDict
from netbox_inventory.models import Asset
import numpy as np
import logging
import traceback
from django.utils import timezone

logger = logging.getLogger(__name__)
nachricht = []

class asset_tagReport(Script):
    description = "Nach freien Asset Tags suchen"

    def __init__(self):
        repeated = self._check_repeated()
        if len(repeated) > 0:
            for asset_tag in repeated:
                setattr(self.__class__,
                    f'test_asset_tag_{asset_tag}_repeated',
                    staticmethod(self._repeated_asset_tag_wrapper(asset_tag, repeated[asset_tag] ))
                    )
        self.setup()

    def _repeated_asset_tag_wrapper(self, asset_tag, devices):

        def run_test():
            self._repeated_asset_tag(asset_tag, devices)

        return run_test

    def _repeated_asset_tag(self, asset_tag, devices):
        for device in devices:
            for ausgabe in nachricht:
                self.log_success(ausgabe, obj=device)
            break
    def _check_repeated(self):
        nummern=np.arange(0, 999)
        device_asset_tags = {}
        ergebnis = []
        for device in Asset.objects.all():
            if device.asset_tag != '':
                if device.asset_tag not in device_asset_tags:
                    device_asset_tags[device.asset_tag] = [device]
                else:
                    device_asset_tags[device.asset_tag].append(device)
                for num in nummern:
                  if str(device.asset_tag) == str(num):
                    nummern = np.delete(nummern, np.where(nummern==num))

        for erg in nummern:
            ergebnis.append(erg)
        repeated_asset_tags = {}
        for asset_tag in device_asset_tags:
           if len(device_asset_tags[asset_tag]) > 1:
                repeated_asset_tags[asset_tag] = device_asset_tags[asset_tag]
        for device in Asset.objects.all().filter(asset_tag=1):
            if device.asset_tag != '':
                for erg in ergebnis:
                    if erg <= 100:
                        continue
                    nachricht.append(f"Asset Tag {erg} ist frei")
        return repeated_asset_tags

    @property
    def name(self):
        return "Device asset_tag Validation"

    def test_Device_has_asset_tag(self):
        for device in Asset.objects.all():
            if device.asset_tag != '':
                print("tja")
            else:
                self.log_failure("Device hasn't asset_tag configured", obj=device)

    def setup(self):
        self._results = OrderedDict()
        self.active_test = None
        self.failed = False

        self.logger = logging.getLogger(f"netbox.reports.{self.full_name}")

        # Compile test methods and initialize results skeleton
        test_methods = {}
        for method in dir(self):
            if method.startswith('test_') and callable(getattr(self, method)):
                method_array = method.split("_")
                method_array = method_array[1:]
                name = " ".join(method_array)
                test_methods[name] = method
                self._results[name] = OrderedDict([
                    ('success', 0),
                    ('info', 0),
                    ('warning', 0),
                    ('failure', 0),
                    ('log', []),
                ])
        if not test_methods:
            raise Exception("A report must contain at least one test method.")
        self.test_methods = test_methods

    def run(self, job_result):
        """
        Run the report and save its results. Each test method will be executed in order.
        """
        self.logger.info(f"Running report")
        job_result.status = JobResultStatusChoices.STATUS_RUNNING
        job_result.save()

        try:

            for method_name in self.test_methods:
                self.active_test = method_name
                test_method = getattr(self, self.test_methods[method_name])
                test_method()

            if self.failed:
                self.logger.warning("Report failed")
                job_result.status = JobResultStatusChoices.STATUS_FAILED
            else:
                self.logger.info("Report completed successfully")
                job_result.status = JobResultStatusChoices.STATUS_COMPLETED

        except Exception as e:
            stacktrace = traceback.format_exc()
            self.log_failure(None, f"An exception occurred: {type(e).__name__}: {e} <pre>{stacktrace}</pre>")
            logger.error(f"Exception raised during report execution: {e}")
            job_result.set_status(JobResultStatusChoices.STATUS_ERRORED)

        job_result.data = self._results
        job_result.completed = timezone.now()
        job_result.save()

        # Perform any post-run tasks
        self.post_run()

matejv commented 2 months ago

Apparently any existing reports you have from netbox3.x should be automatically converted to scripts during the upgrade. With no code changes needed.

I only have one simple report, but I can add netbox 4 and it runs fine.

Could you post the code of your original report without modifications? The thing you posted looks quite complex.

ITEAmplify commented 2 months ago

Thanks for you help,

Sure here is the orginal code: `##!/opt/netbox/venv/bin/python

import django, os, sys

sys.path.append('/opt/netbox/netbox')

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'netbox.settings')

django.setup()

from extras.choices import JobResultStatusChoices from extras.reports import Report from dcim.models import Device from collections import OrderedDict from netbox_inventory.models import Asset import numpy as np import logging import traceback from django.utils import timezone

logger = logging.getLogger(name) nachricht = []

class asset_tagReport(Report): description = "Nach freien Asset Tags suchen"

def __init__(self):
    repeated = self._check_repeated()
    if len(repeated) > 0:
        for asset_tag in repeated:
            setattr(self.__class__,
                f'test_asset_tag_{asset_tag}_repeated',
                staticmethod(self._repeated_asset_tag_wrapper(asset_tag, repeated[asset_tag] ))
                )
    self.setup()

def _repeated_asset_tag_wrapper(self, asset_tag, devices):

    def run_test():
        self._repeated_asset_tag(asset_tag, devices)

    return run_test

def _repeated_asset_tag(self, asset_tag, devices):
    for device in devices:
        for ausgabe in nachricht:
            self.log_success(device, ausgabe)
        break
def _check_repeated(self):
    nummern=np.arange(0, 999)
    device_asset_tags = {}
    ergebnis = []
    for device in Asset.objects.all():
        if device.asset_tag != '':
            if device.asset_tag not in device_asset_tags:
                device_asset_tags[device.asset_tag] = [device]
            else:
                device_asset_tags[device.asset_tag].append(device)
            for num in nummern:
              if str(device.asset_tag) == str(num):
                nummern = np.delete(nummern, np.where(nummern==num))

    for erg in nummern:
        ergebnis.append(erg)
    repeated_asset_tags = {}
    for asset_tag in device_asset_tags:
       if len(device_asset_tags[asset_tag]) > 1:
            repeated_asset_tags[asset_tag] = device_asset_tags[asset_tag]
    for device in Asset.objects.all().filter(asset_tag=1):
        if device.asset_tag != '':
            for erg in ergebnis:
                if erg <= 100:
                    continue
                nachricht.append(f"Asset Tag {erg} ist frei")
    return repeated_asset_tags

@property
def name(self):
    return "Device asset_tag Validation"

def test_Device_has_asset_tag(self):
    for device in Asset.objects.all():
        if device.asset_tag != '':
            print("tja")
        else:
            self.log_failure(device, "Device hasn't asset_tag configured")

def setup(self):
    self._results = OrderedDict()
    self.active_test = None
    self.failed = False

    self.logger = logging.getLogger(f"netbox.reports.{self.full_name}")

    # Compile test methods and initialize results skeleton
    test_methods = {}
    for method in dir(self):
        if method.startswith('test_') and callable(getattr(self, method)):
            method_array = method.split("_")
            method_array = method_array[1:]
            name = " ".join(method_array)
            test_methods[name] = method
            self._results[name] = OrderedDict([
                ('success', 0),
                ('info', 0),
                ('warning', 0),
                ('failure', 0),
                ('log', []),
            ])
    if not test_methods:
        raise Exception("A report must contain at least one test method.")
    self.test_methods = test_methods

def run(self, job_result):
    """
    Run the report and save its results. Each test method will be executed in order.
    """
    self.logger.info(f"Running report")
    job_result.status = JobResultStatusChoices.STATUS_RUNNING
    job_result.save()

    try:

        for method_name in self.test_methods:
            self.active_test = method_name
            test_method = getattr(self, self.test_methods[method_name])
            test_method()

        if self.failed:
            self.logger.warning("Report failed")
            job_result.status = JobResultStatusChoices.STATUS_FAILED
        else:
            self.logger.info("Report completed successfully")
            job_result.status = JobResultStatusChoices.STATUS_COMPLETED

    except Exception as e:
        stacktrace = traceback.format_exc()
        self.log_failure(None, f"An exception occurred: {type(e).__name__}: {e} <pre>{stacktrace}</pre>")
        logger.error(f"Exception raised during report execution: {e}")
        job_result.set_status(JobResultStatusChoices.STATUS_ERRORED)

    job_result.data = self._results
    job_result.completed = timezone.now()
    job_result.save()

    # Perform any post-run tasks
    self.post_run()

`

matejv commented 2 months ago

Oh my... That seems just needlessly complicated. I'm struggling to even figure out what this code is supposed to do.

So, I'm assuming you want to do three things:

check for assets that are missing asset_tag
check for assets that have duplicated asset_tag values
list asset tag values that are free

Here is a script that should do that. I tried to do the minimum needed changes, hopefully you'll be able to see what I'v changed.

A couple more notes:

If an asset doesn't have an asset tag, the asset.asset_tag is not an empty string (''), but None. In an if statement if you want to do something for asset that has asset_tag, you can just write: if device.asset_tag: Or if you want to do something if an asset doesn't have tag you do: if not device.asset.tag:

You're using numpy library to generate an array of numbers and then remove certain values. You could do that using range() function and set() data type.

Anyhow, here is the script:

##!/opt/netbox/venv/bin/python
#import django, os, sys
#sys.path.append('/opt/netbox/netbox')
#os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'netbox.settings')
#django.setup()
from extras.scripts import Script
from netbox_inventory.models import Asset
import numpy as np

nachricht = []

class asset_tagReport(Script):
    description = "Nach freien Asset Tags suchen"

    def _check_repeated(self):
        nummern=np.arange(0, 999)
        device_asset_tags = {}
        for device in Asset.objects.all():
            if device.asset_tag:
                if device.asset_tag not in device_asset_tags:
                    device_asset_tags[device.asset_tag] = [device]
                else:
                    device_asset_tags[device.asset_tag].append(device)
                for num in nummern:
                  if str(device.asset_tag) == str(num):
                    nummern = np.delete(nummern, np.where(nummern==num))

        for erg in nummern:
            if erg <= 100:
                continue
            nachricht.append(f"Asset Tag {erg} ist frei")
        repeated_asset_tags = {}
        for asset_tag in device_asset_tags:
           if len(device_asset_tags[asset_tag]) > 1:
                repeated_asset_tags[asset_tag] = device_asset_tags[asset_tag]
        return repeated_asset_tags

    def run(self, data, commit):
        repeated = self._check_repeated()
        for asset_tag in repeated:
            for device in repeated[asset_tag]:
                self.log_failure(f"Device {device} has asset_tag repeated")

        self.check_device_has_asset_tag()

        return '\n'.join(nachricht)

    def check_device_has_asset_tag(self):
        for device in Asset.objects.all():
            if not device.asset_tag:
                self.log_failure(f"Device {device} hasn't asset_tag configured")

Hope you can customize it further according to your needs.

ITEAmplify commented 2 months ago

Thanks many many times, i got your point that its to complicated, im gonna rebuild the script myself with the use of range() and set() to further undestand the logic 😊 Thanks again you really helped me

ArnesSI / netbox-inventory

Custom Script #167

import django, os, sys

sys.path.append('/opt/netbox/netbox')

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'netbox.settings')

django.setup()