ACP-260门禁面板日志获取问题:Python调用PullSDK无效求助
我有一台正常运行的ACP-260双门门禁面板,此前使用ZKAccess 3.5进行管理,现在希望自行编写Python脚本获取设备数据。调研后基于ZKTeco的PullSDK编写了连接代码,但无法获取任何日志。需要获取的是每次人员通过1/2号门时的实时日志,包含时间、用户信息、准入方式(指纹/刷卡)及失败尝试等门禁使用详情。试过python库pyzkaccess,但它不支持ACP-260,因此该方案不可行。
我的实现代码如下:
import time import threading from ctypes import windll, create_string_buffer plcommpro_dll_path = "E:/pyTCP/deps/plcommpro.dll" class DeviceController: def __init__(self, ip, port=4370, timeout=4000, passwd=""): self.ip = ip self.port = port self.timeout = timeout self.passwd = passwd self.hcommpro = None self.commpro = windll.LoadLibrary(plcommpro_dll_path) self.is_running = False self.log_thread = None def connect(self): params = f"protocol=TCP,ipaddress={self.ip},port={self.port},timeout={self.timeout},passwd={self.passwd}" constr = create_string_buffer(params.encode()) self.hcommpro = self.commpro.Connect(constr) return self.hcommpro def disconnect(self): if self.hcommpro: self.commpro.Disconnect(self.hcommpro) self.hcommpro = None def get_real_time_logs(self, buffer_size=256): rt_log = create_string_buffer(buffer_size) ret = self.commpro.GetRTLog(self.hcommpro, rt_log, buffer_size) return rt_log.value.decode() def retrieve_logs_continuously(self, interval=60): while self.is_running: rt_logs = self.get_real_time_logs() print(rt_logs) time.sleep(interval) def start_logging(self, interval=60): if not self.is_running: self.is_running = True self.log_thread = threading.Thread(target=self.retrieve_logs_continuously, args=(interval,)) self.log_thread.start() def stop_logging(self): if self.is_running: self.is_running = False self.log_thread.join() if __name__ == "__main__": device = DeviceController(ip="10.0.0.240") device.connect() device.start_logging(interval=600) try: while True: time.sleep(1) except KeyboardInterrupt: device.stop_logging() device.disconnect()
排查与修复步骤
校验DLL版本兼容性
确保使用的plcommpro.dll是ACP-260专属版本,ZKTeco不同系列设备对SDK版本有要求,不要用通用版,去官方下载ACP系列对应的PullSDK包替换现有文件。验证设备连接状态
在connect方法中增加连接失败校验,Connect函数成功返回非0句柄,失败返回0:def connect(self): params = f"protocol=TCP,ipaddress={self.ip},port={self.port},timeout={self.timeout},passwd={self.passwd}" constr = create_string_buffer(params.encode()) self.hcommpro = self.commpro.Connect(constr) if self.hcommpro == 0: raise Exception("连接失败,请检查IP、端口、密码或网络连通性") return self.hcommpro修正实时日志获取逻辑
PullSDK的GetRTLog是被动等待设备推送新日志,并非主动拉取,且单条日志可能超过256字节,建议调整缓冲区大小并优化监听间隔:def get_real_time_logs(self, buffer_size=1024): rt_log = create_string_buffer(buffer_size) log_count = self.commpro.GetRTLog(self.hcommpro, rt_log, buffer_size) # ZKTeco SDK默认使用GBK编码,避免乱码 if log_count > 0: return rt_log.value.decode('gbk') return "" def retrieve_logs_continuously(self): while self.is_running: rt_logs = self.get_real_time_logs() if rt_logs: print(rt_logs) # 短间隔轮询,避免错过实时日志 time.sleep(1)同时修改
start_logging方法,去掉不必要的interval参数:def start_logging(self): if not self.is_running: self.is_running = True self.log_thread = threading.Thread(target=self.retrieve_logs_continuously) self.log_thread.start()检查设备配置
登录设备Web管理后台(若支持),确认实时日志推送功能已开启;同时检查设备日志存储是否已满,网络防火墙是否允许设备向脚本所在机器发送数据(PullSDK本质是设备主动推日志到客户端)。先验证历史日志读取
添加历史日志读取方法,确认SDK能正常和设备交互:def get_history_logs(self, buffer_size=4096): log_buf = create_string_buffer(buffer_size) log_count = self.commpro.GetGeneralLog(self.hcommpro, log_buf, buffer_size) if log_count > 0: return log_buf.value.decode('gbk') return ""在主函数中调用该方法,若能获取历史日志,说明实时日志监听逻辑需要调整;若仍无法获取,需排查连接权限或设备SDK支持问题。
内容的提问来源于stack exchange,提问作者winston1420

