Python脚本比对CSV与LDAP生产服务器数据无法生成z.csv问题求助

原脚本核心错误清单 #

• check_ldap_status函数逻辑完全错误：遍历LDAP结果时，第一次循环就会直接返回结果，不会匹配后续条目。如果第一个LDAP条目的employeeNumber不等于当前查询的empid，会直接返回matched，完全不符合比对逻辑。
• LDAP属性取值无容错：如果LDAP条目没有calStateEduPersonID属性，直接访问.value会抛出KeyError异常，导致脚本中断。
• CSV写入语法完全错误：csv.writer()是创建写入对象的方法，不能直接传入内容写入，需要先实例化writer对象再调用writerow()方法写入行。
• 变量名冲突&资源频繁释放：读写CSV都用了csvfile作为变量名，会覆盖外层读文件的对象；每次写入行都重复打开关闭输出文件，不仅效率极低，还容易触发IO异常。
• 表头写入逻辑错误：已经用next(rows)跳过了CSV表头，遍历的i=0对应的是第一行数据，此时写表头会导致第一行数据丢失，且表头字段也不符合输出需求。
• LDAP搜索结果未做预处理：每次查询empid都全量遍历LDAP结果，数据量大时性能极差，建议先把LDAP结果转成employeeNumber为key的字典，一次查询时间复杂度降到O(1)。

修正后的完整代码 #

from ldap3 import Server, Connection, ALL
import csv

# LDAP服务器连接配置
server = Server('server', get_info=ALL)
conn = Connection(server, 'uid=idmsa,ou=People,ou=auth,o=csun', 'password', auto_bind=True)

# 拉取LDAP全量数据，提前转为字典优化查询效率
# 搜索条件增加employeeNumber存在的过滤，避免返回无效条目
conn.search(
    'o=csun', 
    '(&(uid=*)(employeeNumber=*))', 
    attributes=['employeeNumber', 'calStateEduPersonID']
)
ldap_dict = {}
for entry in conn.entries:
    emp_num = entry['employeeNumber'].value
    # 处理calStateEduPersonID不存在的情况
    cs_id = entry['calStateEduPersonID'].value if 'calStateEduPersonID' in entry else None
    ldap_dict[emp_num] = cs_id

# 比对逻辑
def check_ldap_status(empid, csid):
    if empid not in ldap_dict:
        # 可根据需求调整LDAP不存在该empid的返回状态
        return 'unavailable'
    ldap_cs_id = ldap_dict[empid]
    if ldap_cs_id in ('', None):
        return 'unavailable'
    elif ldap_cs_id != csid:
        return 'unmatched'
    else:
        return 'matched'

# 读写CSV逻辑，一次性打开输出文件避免重复IO
with open('csvfile', 'r', encoding='utf-8') as in_csv, open('z.csv', 'w', encoding='utf-8', newline='') as out_csv:
    reader = csv.reader(in_csv)
    writer = csv.writer(out_csv)
    # 写入输出文件表头
    writer.writerow(['EmployeeNumber', 'Status'])
    # 跳过输入文件表头
    next(reader)
    for row in reader:
        if not row:
            continue
        empid, csid, *_ = row
        empid = empid.strip()
        if not empid:
            continue
        status = check_ldap_status(empid, csid.strip())
        if status in ('unmatched', 'unavailable'):
            writer.writerow([empid, status])

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