Mass Algorithm Changes

Changing Algorithm Mode

The following is a fully fleshed out command line script for setting some or all Zones AT8 filter mode to auto or monitor.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "requests",
# ]
# ///
"""Change AT8 to auto or monitor mode."""
import json
import urllib.parse
import requests
import traceback
import argparse

HEADERS = {'Content-Type':'application/json',
            'Accept': 'application/json'}

# Put in your user credentials here
CREDENTIALS = {'login':'admin', 'password':'admin123'}

def url(resource, params=None):
    """Turn a resource into a url.

    Uses base_url and optionally encodes a hash of parameters"""
    querystring = ""
    if params != None:
        querystring = "?"+urllib.parse.urlencode(params)
    return BASE_URL + resource + querystring

def wrap(data):
    """Convert an object to an rWeb suitable json string

    Wraps the data in the standard rweb {'data': } object """
    return json.dumps({'data': data})

# API functions
def login(session):
    """Log in to a session.

    The authentication cookie is saved to the session"""
    result = session.post(url('/authentication'),
                          wrap(CREDENTIALS))
    assert(result.status_code == requests.codes.created)

def logout(session):
    """Close a session"""
    result = session.delete(url('/authentication'))
    assert(result.status_code == requests.codes.ok)

def get_group(session, group_name):
    """This gets the group resources of the group with the required name"""
    result = session.get(url('/groups', {'name':group_name}))
    assert(result.status_code == requests.codes.ok)
    group = json.loads(result.content)
    if not group['data']:
        raise ValueError(f"Group {group_name} doesn't exist")
    return group['data'][0]

def get_zone(session, group_name, zone_name=None):
    """This gets the zone resources from the provided group and zone name
    If no zone is specified it returns all the zones resources in the group"""
    group = get_group(session, group_name)
    if zone_name is not None:
        result = session.get(url(group['zones_resource'], {'name':zone_name}))
    else:
        result = session.get(url(group['zones_resource']))
    assert(result.status_code == requests.codes.ok)
    zone = json.loads(result.content)
    if not zone['data']:
        raise ValueError(f"Zone {zone_name} does not exist in group")
    return zone['data']

def get_all_zones(session):
    """returns the resources of all the zones in the rweb"""
    result = session.get(url('/zones'))
    assert(result.status_code == requests.codes.ok)
    zones = json.loads(result.content)
    return zones['data']

def set_at8(session, mode ,zone_resources):
    zone = zone_resources
    result = session.get(url(zone['revision_current_resource']))
    current = json.loads(result.content)
    assert(result.status_code == requests.codes.ok)
    result = session.post(url(current['configuration_resource']),
                          wrap({'configurations':{'at_8_filter_mode':str(mode)}}))
    assert(result.status_code == requests.codes.created)
    result = json.loads(result.content)
    return result

def main():
    """Main CLI interface"""
    global BASE_URL

    # Handle some command line arguments
    parser = argparse.ArgumentParser(description="Change AT8 filter mode to auto or monitor mode.\n"
                                     "If no group and no zone is specifed, all zones in rweb are affected.\n"
                                     "If only group is specifed, all zones in the group are affected.\n"
                                     "If both group and zone are specified, that specific zone is affected.")
    parser.add_argument('server', help="rWeb server address")
    parser.add_argument('-g', '--group', nargs='?', type=str,
                        help="Group name")
    parser.add_argument('-z', '--zone', nargs='?', type=str,
                        help="Zone name")
    parser.add_argument('mode', choices=['auto', 'monitor'],
                        help="AT8 Filter Mode")

    args = parser.parse_args()

    BASE_URL = "https://{0}".format(args.server) #change to specific rWeb instance

    with requests.session() as session:
        session.headers = HEADERS
        # This is insecure but may be needed if you haven't set up an SSL certificate for rWeb yet
        session.verify = False
        login(session)

        if args.group:
            zones = get_zone(session, args.group, args.zone)
        else:
            zones = get_all_zones(session)

        for i in zones:
            try:
                set_at8(session, args.mode, i)
                print(f"Set AT8 to {args.mode} mode in Zone {i['name']} (Group: {i['group_name']})")
            except Exception as e:
                trace = traceback.format_exc()
                print(trace)
                print(f"Error changing Zone {i['name']} (Group: {i['group_name']})")

        logout(session)

if __name__ == '__main__':
    main()

Changing Algorithm Advanced Settings

The following is a fully fleshed out command line script for setting AT1 "Maximum Duplicates" to a specified value in some or all Zones.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "requests",
# ]
# ///
"""Change AT1 "Maximum duplicates" to a specified value."""
import json
import urllib.parse
import requests
import traceback
import argparse

HEADERS = {'Content-Type':'application/json',
            'Accept': 'application/json'}

# Put in your user credentials here
CREDENTIALS = {'login':'admin', 'password':'admin123'}

def url(resource, params=None):
    """Turn a resource into a url.

    Uses base_url and optionally encodes a hash of parameters"""
    querystring = ""
    if params != None:
        querystring = "?"+urllib.parse.urlencode(params)
    return BASE_URL + resource + querystring

def wrap(data):
    """Convert an object to an rWeb suitable json string

    Wraps the data in the standard rweb {'data': } object """
    return json.dumps({'data': data})

# API functions
def login(session):
    """Log in to a session.

    The authentication cookie is saved to the session"""
    result = session.post(url('/authentication'),
                          wrap(CREDENTIALS))
    assert(result.status_code == requests.codes.created)

def logout(session):
    """Close a session"""
    result = session.delete(url('/authentication'))
    assert(result.status_code == requests.codes.ok)

def get_group(session, group_name):
    """This gets the group resources of the group with the required name"""
    result = session.get(url('/groups', {'name':group_name}))
    assert(result.status_code == requests.codes.ok)
    group = json.loads(result.content)
    if not group['data']:
        raise ValueError(f"Group {group_name} doesn't exist")
    return group['data'][0]

def get_zone(session, group_name, zone_name=None):
    """This gets the zone resources from the provided group and zone name
    If no zone is specified it returns all the zones resources in the group"""
    group = get_group(session, group_name)
    if zone_name is not None:
        result = session.get(url(group['zones_resource'], {'name':zone_name}))
    else:
        result = session.get(url(group['zones_resource']))
    assert(result.status_code == requests.codes.ok)
    zone = json.loads(result.content)
    if not zone['data']:
        raise ValueError(f"Zone {zone_name} does not exist in group")
    return zone['data']

def get_all_zones(session):
    """returns the resources of all the zones in the rweb"""
    result = session.get(url('/zones'))
    assert(result.status_code == requests.codes.ok)
    zones = json.loads(result.content)
    return zones['data']

def set_at1(session, value, zone_resources):
    zone = zone_resources
    result = session.get(url(zone['revision_current_resource']))
    current = json.loads(result.content)
    assert(result.status_code == requests.codes.ok)
    result = session.post(url(current['configuration_resource'])
            , wrap({'configurations': {'at_1_st_2_duplicates_allowed_max': str(value)}}))
    assert(result.status_code == requests.codes.created)
    result = json.loads(result.content)
    return result

def main():
    """Main CLI interface"""
    global BASE_URL

    # Handle some command line arguments
    parser = argparse.ArgumentParser(description="Change AT1 'Maximum duplicates' to a specified value.\n"
                                     "If no group and no zone is specifed, all zones in rweb are affected.\n"
                                     "If only group is specifed, all zones in the group are affected.\n"
                                     "If both group and zone are specified, that specific zone is affected.")
    parser.add_argument('server', help="rWeb server address")
    parser.add_argument('-g', '--group', nargs='?', type=str,
                        help="Group name")
    parser.add_argument('-z', '--zone', nargs='?', type=str,
                        help="Zone name")
    parser.add_argument('number', type=int,
                        help="Number of AT1 max duplicates to set")

    args = parser.parse_args()

    BASE_URL = "https://{0}".format(args.server) # Change to specific rWeb instance

    with requests.session() as session:
        session.headers = HEADERS
        # This is insecure but may be needed if you haven't set up an SSL certificate for rWeb yet
        session.verify = False
        login(session)

        if args.group:
            zones = get_zone(session, args.group, args.zone)
        else:
            zones = get_all_zones(session)

        for i in zones:
            try:
                set_at1(session, args.number, i)
                print(f"Set AT1 Maximum duplicates to {args.number} in Zone {i['name']} (Group: {i['group_name']})")
            except Exception as e:
                trace= traceback.format_exc()
                print(trace)
                print(f"Error changing Zone {i['name']} (Group: {i['group_name']})")

        logout(session)

if __name__ == '__main__':
    main()