Detect printer name and IP with Get-Printer command from powershell and python (v2)

detect_printers.py

import subprocess
import sys
import platform
import ipaddress
from datetime import datetime
import time
import queue
from threading import Thread
import requests
from bs4 import BeautifulSoup
import pandas as pd

server = sys.argv[1]

def execute(cmd):
    """
    Execute a powershell command in local machine
    Params:
        cmd (str): valid powershell command
    Return:
        (str): default output of command
    Exceptions:
        OSerror: failed execution
    """
    process = subprocess.Popen(["powershell.exe",cmd],
                        stdout = subprocess.PIPE, 
                        stderr = subprocess.PIPE,
                        text = True,
                        shell = True
                        )
    std_out, std_err = process.communicate()
    if len(std_err) == 0:
        return std_out.strip()
    else:
        raise OSError(std_err)

def get_printer(server):
    """
    Execute Get-Printer command in a printer server, 
    get the values of the name and address of printer
    filter through valid IPs and single name printers

    Params:
        server (str): address or hostname of the printer server
    Return:
        (list): list of dicts with the keys: "printer_name" and "ip".
    """
    print("Executing get_printer")
    list_final = []
    out = execute("get-printer -computer " + server + " | select-object name,portname")
    list_printers = out.split("\n")
    del list_printers[:2] # first two lines are the column title and horizontal separator
    for i in list_printers:
        p = i.split()
        if len(p) <= 2: # lists bigger than this not fit in my scope, dispensable
            try:
                printer_address = p[1]
                # Treat printers with "_2" or "_1" to represent duplicated ports, increasing accuracy
                if "_" in printer_address:
                    printer_address = printer_address.split("_")[0]
                ip = ipaddress.ip_address(printer_address) # check if address is a valid IP
                list_final.append(
                    {
                        "printer_name":p[0],
                        "address": printer_address
                        })
            except ValueError:
                pass # discard ip not valid
    return list_final

def printer_title(ip):
    """
    Parse the HTML from the webpage server of a printer
    Often the model is in the title tag
    Parameters:
        ip (str): ip address of the printer
    Return:
        (str): the title of page, none if fail
    """
    try:
        data = requests.get(f"http://{ip}", timeout=10, verify=False).text
        soup = BeautifulSoup(data, 'html.parser')
        return soup.title.text
    # at any error, return none
    except AttributeError:
        return None
    except requests.exceptions.ConnectTimeout:
        return None
    except requests.exceptions.ConnectionError:
        return None
    except requests.exceptions.SSLError:
            return None

def run_with_threads(printer_list:list):
    """
    Run the requests to parse page title with threads
    Huge improvement in performance
    Parameters:
        printer_list (list): list of dicts with the data obtained from powershell command
            Keys:
            address (str)
            printer_name (str)
    Return:
        (list): List of dict, similar to input but added the key "model"
        Keys:
            address (str)
            printer_name (str)
            model (str)
    """
    start_time = time.time()
    def execute_command():
        """
        Function to be executed by the thread, run the request and parse
        """
        while True:
            printer = queue_printers.get()
            model = printer_title(printer["address"])
            queue_responses.put(
                {
                    "machine_instance": printer["address"],
                    "model": model,
                    "printer_name": printer["printer_name"]
                }
            )
            queue_printers.task_done()

    
    
    queue_printers = queue.Queue()
    queue_responses = queue.Queue()

    # run threads
    # one thread for list item
    for x in range(1, len(printer_list)):
            proletariat = Thread(target=execute_command)
            proletariat.setDaemon(True)
            proletariat.start()

    # fill the queue
    for x in printer_list:
        queue_printers.put(x)
    
    queue_printers.join() # execute and wait

    list_responses = []
    while True:
        try:
            # get value without waiting execution
            response = queue_responses.get_nowait()
            list_responses.append(response)
        except queue.Empty:
            break  # break the loop when the queue is empty

    end_time = time.time()
    print("Time to execute:  ", end_time - start_time)
    return list_responses

if __name__ == "__main__":
    if platform.system() != "Windows":
        raise OSerror("Script only run in Windows")
    if server is not None:
        _ = get_printer(server)
        printer_list = run_with_threads(_)
        """
        for x in _:
            x["model"] = printer_title(x["address"])
        """
        df = pd.DataFrame(printer_list)
        df = df.sort_values("model")
        # export to CSV at end
        # know bug: duplicated entries in CSV
        df.to_csv(f"{server}_{datetime.now().strftime("%d-%m-%Y_%Hh%Mm")}.csv", sep=";", header=True, index=False)