如何基于AD安全组与PC主机名设置默认打印机？附Python代码求助

问题背景 #

我们公司的打印服务器按成本中心划分多台打印机，部分打印机共用IP，通过Active Directory（AD）安全组自动为用户映射打印机。但公司内像医生这类需要在工作站、手术室、诊室等不同区域办公的用户，切换地点后无法手动设置默认打印机。我需要一个持续运行的Python脚本，实现用户登录域内PC时，识别登录账号、PC主机名及AD安全组成员身份，自动设置对应默认打印机（例如：用户igorcarmona属于IMP_TIC_COLOR组且登录主机名以TIC开头的PC时，设置对应打印机为默认）。以下是我的代码草稿，目前遇到实现困难：

import os
import time
import pyad.adquery
import win32print
import socket

AD_SERVER_IP = "192.168.0.193"
PRINT_SERVER_IP = "192.168.0.230"

def check_user_login():
    query = pyad.adquery.ADQuery()
    query.execute_query(
        type=AD_SERVER_IP,
        attributes=["memberOf"],
        where_clause="objectClass = 'user' AND sAMAccountName = '" + os.getlogin() + "'"
    )
    result = query.get_single_result()
    if result:
        groups = result['memberOf']
        return groups
    return []

def get_host_name():
    return socket.gethostname()

def set_default_printer(printer_name):
    printer_info = win32print.EnumPrinters(win32print.PRINTER_ENUM_CONNECTIONS, None, 1)
    for printer in printer_info:
        if printer_name in printer[2]:
            win32print.SetDefaultPrinter(printer[2])
            print("Default printer set to:", printer[2])
            break

def main():
    while True:
        groups = check_user_login()
        hostname = get_host_name()
        if groups:
            for group in groups:
                print(group)
                if "IMP_TIC_COLOR" in group and hostname.startswith("TIC"):
                    set_default_printer("IMP_TIC_COLOR")

        time.sleep(30)  # Verifica a cada 30 segundos

if __name__ == "__main__":
    main()

现有代码问题分析 #

• ADQuery.execute_query参数错误：type参数应指定查询类型（如"user"），而非AD服务器IP；AD服务器地址需通过pyad.set_defaults()提前配置。
• 查询条件存在注入风险：直接拼接os.getlogin()到SQL语句，未使用参数化查询。
• 组名匹配不准确：memberOf返回的是组的完整区分名称（DN），直接判断字符串包含会导致误匹配。
• 未处理用户切换场景：脚本启动后仅获取一次用户，无法识别注销后重新登录的其他账号。
• 缺少错误处理：AD查询、打印机枚举、默认设置过程中无异常捕获，脚本易崩溃。
• 打印机名称匹配不严谨：仅匹配部分名称可能导致设置错误，需获取完整打印机名称。

修正后的代码 #

import os
import time
import pyad.adquery
import win32print
import socket
import win32api
import win32security

# 配置AD服务器
AD_SERVER_IP = "192.168.0.193"
pyad.set_defaults(ldap_server=AD_SERVER_IP)

def get_current_user_samaccountname():
    # 通过Windows API获取当前活跃用户的SAM账号，支持用户切换场景
    try:
        token = win32security.OpenProcessToken(win32api.GetCurrentProcess(), win32security.TOKEN_QUERY)
        user_sid = win32security.GetTokenInformation(token, win32security.TokenUser)[0]
        username, _, _ = win32security.LookupAccountSid(None, user_sid)
        return username
    except Exception as e:
        print(f"获取用户账号失败: {str(e)}")
        return os.getlogin()

def get_user_ad_groups(sam_account_name):
    # 查询用户所属AD安全组，提取纯组名
    query = pyad.adquery.ADQuery()
    try:
        query.execute_query(
            attributes=["memberOf"],
            where_clause="objectClass = 'user' AND sAMAccountName = %(username)s",
            param_dict={"username": sam_account_name},
            type="user"
        )
        result = query.get_single_result()
        if not result:
            return []
        
        # 从DN中提取组名，示例DN: CN=IMP_TIC_COLOR,OU=Groups,DC=domain,DC=com
        groups = []
        for dn in result["memberOf"]:
            group_name = dn.split(",")[0].split("=")[1]
            groups.append(group_name)
        return groups
    except Exception as e:
        print(f"AD组查询失败: {str(e)}")
        return []

def get_host_name():
    return socket.gethostname()

def get_full_printer_name(target_prefix):
    # 枚举已连接打印机，返回匹配的完整名称
    try:
        printer_list = win32print.EnumPrinters(win32print.PRINTER_ENUM_CONNECTIONS, None, 1)
        for printer in printer_list:
            full_name = printer[2]
            if target_prefix in full_name:
                return full_name
        print(f"未找到包含[{target_prefix}]的已连接打印机")
        return None
    except Exception as e:
        print(f"枚举打印机失败: {str(e)}")
        return None

def set_default_printer(printer_full_name):
    if not printer_full_name:
        return
    try:
        win32print.SetDefaultPrinter(printer_full_name)
        print(f"默认打印机已设置为: {printer_full_name}")
    except Exception as e:
        print(f"设置默认打印机失败: {str(e)}")

def main():
    last_logged_user = ""
    while True:
        current_user = get_current_user_samaccountname()
        hostname = get_host_name()

        # 仅在用户变更或首次运行时执行规则匹配
        if current_user != last_logged_user:
            last_logged_user = current_user
            print(f"\n检测到当前登录用户: {current_user}")
            user_groups = get_user_ad_groups(current_user)
            print(f"用户所属AD组: {user_groups}")

            # 规则1：TIC区域+IMP_TIC_COLOR组
            if "IMP_TIC_COLOR" in user_groups and hostname.startswith("TIC"):
                target_printer = get_full_printer_name("IMP_TIC_COLOR")
                set_default_printer(target_printer)
            
            # 可添加更多规则示例
            # elif "IMP_ICU_BLACK" in user_groups and hostname.startswith("ICU"):
            #     target_printer = get_full_printer_name("IMP_ICU_BLACK")
            #     set_default_printer(target_printer)

        time.sleep(30)  # 每30秒检查一次状态

if __name__ == "__main__":
    main()

关键改进说明 #

• AD配置修正：使用pyad.set_defaults()指定AD服务器，参数化查询避免注入风险。
• 组名精准匹配：从AD返回的组DN中提取纯组名，确保匹配逻辑准确。
• 用户切换检测：通过Windows API获取当前活跃用户，支持注销后重新登录的场景。
• 全链路错误处理：为AD查询、打印机操作添加异常捕获，提升脚本稳定性。
• 规则扩展性：预留了更多组与主机名的匹配规则入口，方便后续新增成本中心的配置。

内容的提问来源于stack exchange，提问作者Igor Carmona