Initial test
This commit is contained in:
Aries Wibowo
371
ZKFingerprint.py
Normal file
371
ZKFingerprint.py
Normal file
@@ -0,0 +1,371 @@
|
||||
import datetime
|
||||
import socket
|
||||
from typing import Union, Optional, List
|
||||
from dataclasses import dataclass
|
||||
import json
|
||||
import csv
|
||||
|
||||
@dataclass
|
||||
class UserInfo:
|
||||
pin: str
|
||||
name: str
|
||||
password: str
|
||||
group: str
|
||||
privilege: str
|
||||
card: str
|
||||
pin2: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class UserAttendance:
|
||||
pin: str
|
||||
datetime: str
|
||||
verified: str
|
||||
status: str
|
||||
workcode: str
|
||||
|
||||
|
||||
class Fingerprint(object):
|
||||
|
||||
payload = {
|
||||
'GetAttLog': '<GetAttLog><ArgComKey xsi:type=\"xsd:integer\">#COMKEY</ArgComKey>#PIN</GetAttLog>', # noqa: E501
|
||||
'GetUserInfo': '<GetUserInfo><ArgComKey xsi:type=\"xsd:integer\">#COMKEY</ArgComKey>#PIN</GetUserInfo>', # noqa: E501
|
||||
'SetUserInfo': '<SetUserInfo><ArgComKey xsi:type=\"xsd:integer\">#COMKEY</ArgComKey>#Name</SetUserInfo>',
|
||||
}
|
||||
|
||||
def __init__(self, ip: str, port: int = 80, comkey: str = '') -> None:
|
||||
super(Fingerprint, self).__init__()
|
||||
self.__ip = ip
|
||||
self.__port = port if isinstance(port, int) else int(port)
|
||||
self.__comkey = comkey
|
||||
self.connect()
|
||||
|
||||
def connect(self) -> None:
|
||||
try:
|
||||
self.__conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
self.__conn.connect((self.__ip, self.__port))
|
||||
except Exception:
|
||||
raise Exception("Can't Connect")
|
||||
|
||||
def getStatus(self) -> str:
|
||||
return 'connected' if self.__conn else 'disconnected'
|
||||
|
||||
def getUserInfo(self, pin: Union[str, list] = "all") -> List[UserInfo]:
|
||||
self.connect()
|
||||
if isinstance(pin, list):
|
||||
listpin = ""
|
||||
for pinid in pin:
|
||||
listpin += "<Arg><PIN>"+pinid+"</PIN></Arg>"
|
||||
pin = listpin
|
||||
print("if condition")
|
||||
else:
|
||||
print("else condition")
|
||||
pin = "<Arg><PIN>"+pin+"</PIN></Arg>"
|
||||
print(pin)
|
||||
payload = self.__generatePayload(
|
||||
'GetUserInfo',
|
||||
{'key': "#PIN", 'value': pin}
|
||||
)
|
||||
print(payload)
|
||||
data = self.__send(payload)
|
||||
print('Data Start')
|
||||
print(data)
|
||||
print('Data End')
|
||||
if data and "<GetUserInfoResponse>" not in data:
|
||||
return []
|
||||
|
||||
return self.__parseUserInfoData(data)
|
||||
|
||||
def setUserInfo(self, pin, name):
|
||||
self.connect()
|
||||
if isinstance(pin, list):
|
||||
listpin = ""
|
||||
for pinid in pin:
|
||||
listpin += "<Arg><PIN>"+pinid+"</PIN></Arg>"
|
||||
pin = listpin
|
||||
print("if condition")
|
||||
else:
|
||||
print("else condition")
|
||||
pin = "<Arg><PIN>"+pin+"</PIN></Arg>"
|
||||
print(pin)
|
||||
payload = self.__generatePayload(
|
||||
'SetUserInfo',
|
||||
{'key': "#Name", 'value': name}
|
||||
)
|
||||
print(payload)
|
||||
data = self.__send(payload)
|
||||
print('Data Start')
|
||||
print(data)
|
||||
print('Data End')
|
||||
if data and "<SetUserInfoResponse>" not in data:
|
||||
return []
|
||||
|
||||
return self.__parseUserInfoData(data)
|
||||
|
||||
def __parseUserInfoData(self, data: str) -> List[UserInfo]:
|
||||
datas = data.split("<Row>")
|
||||
datas.pop(0)
|
||||
|
||||
userData = []
|
||||
for dataRow in datas:
|
||||
fid = self.__getValueFromTag(
|
||||
dataRow,
|
||||
"<PIN>",
|
||||
"</PIN>"
|
||||
)
|
||||
name = self.__getValueFromTag(
|
||||
dataRow,
|
||||
"<Name>",
|
||||
"</Name>"
|
||||
)
|
||||
password = self.__getValueFromTag(
|
||||
dataRow,
|
||||
"<Password>",
|
||||
"</Password>"
|
||||
)
|
||||
group = self.__getValueFromTag(
|
||||
dataRow,
|
||||
"<Group>",
|
||||
"</Group>"
|
||||
)
|
||||
privilege = self.__getValueFromTag(
|
||||
dataRow,
|
||||
"<Privilege>",
|
||||
"</Privilege>"
|
||||
)
|
||||
card = self.__getValueFromTag(
|
||||
dataRow,
|
||||
"<Card>",
|
||||
"</Card>"
|
||||
)
|
||||
pin2 = self.__getValueFromTag(
|
||||
dataRow,
|
||||
"<PIN2>",
|
||||
"</PIN2>"
|
||||
)
|
||||
|
||||
user = UserInfo(
|
||||
pin=fid,
|
||||
name=name,
|
||||
password=password,
|
||||
group=group,
|
||||
privilege=privilege,
|
||||
card=card,
|
||||
pin2=pin2
|
||||
)
|
||||
userData.append(user)
|
||||
return userData
|
||||
|
||||
def getAttendance(
|
||||
self,
|
||||
pin: Union[str, list] = "all",
|
||||
date_start: Optional[str] = None,
|
||||
date_end: Optional[str] = None
|
||||
):
|
||||
self.connect()
|
||||
if date_start is not None and date_end is None:
|
||||
date_end = date_start
|
||||
|
||||
if isinstance(pin, list):
|
||||
listpin = ""
|
||||
for pinid in pin:
|
||||
listpin += "<Arg><PIN>"+pinid+"</PIN></Arg>"
|
||||
pin = listpin
|
||||
else:
|
||||
pin = "<Arg><PIN>"+pin+"</PIN></Arg>"
|
||||
|
||||
payload = self.__generatePayload(
|
||||
'GetAttLog',
|
||||
{'key': "#PIN", 'value': pin}
|
||||
)
|
||||
data = self.__send(payload)
|
||||
|
||||
if "<GetAttLogResponse>" not in data:
|
||||
return None
|
||||
return self.__parseAttendanceData(data, date_start, date_end)
|
||||
|
||||
def __parseAttendanceData(
|
||||
self,
|
||||
data: str,
|
||||
date_start: Optional[str] = None,
|
||||
date_end: Optional[str] = None
|
||||
) -> List[UserAttendance]:
|
||||
datas = data.split("<Row>")
|
||||
datas.pop(0)
|
||||
|
||||
userData = []
|
||||
for dataRow in datas:
|
||||
fid = self.__getValueFromTag(
|
||||
dataRow,
|
||||
"<PIN>",
|
||||
"</PIN>"
|
||||
)
|
||||
dt = self.__getValueFromTag(
|
||||
dataRow,
|
||||
"<DateTime>",
|
||||
"</DateTime>"
|
||||
)
|
||||
verified = self.__getValueFromTag(
|
||||
dataRow,
|
||||
"<Verified>",
|
||||
"</Verified>"
|
||||
)
|
||||
status = self.__getValueFromTag(
|
||||
dataRow,
|
||||
"<Status>",
|
||||
"</Status>"
|
||||
)
|
||||
workcode = self.__getValueFromTag(
|
||||
dataRow,
|
||||
"<WorkCode>",
|
||||
"</WorkCode>"
|
||||
)
|
||||
|
||||
if date_start is not None and date_end is not None:
|
||||
DR = self.__dateRange(date_start, date_end)
|
||||
|
||||
dateCheck = dt.split(" ")[0]
|
||||
|
||||
if dateCheck in DR:
|
||||
user = UserAttendance(
|
||||
pin=fid,
|
||||
datetime=dt,
|
||||
verified=verified,
|
||||
status=status,
|
||||
workcode=workcode,
|
||||
)
|
||||
userData.append(user)
|
||||
else:
|
||||
user = UserAttendance(
|
||||
pin=fid,
|
||||
datetime=dt,
|
||||
verified=verified,
|
||||
status=status,
|
||||
workcode=workcode,
|
||||
)
|
||||
userData.append(user)
|
||||
return userData
|
||||
|
||||
def __send(self, payload: bytes) -> str:
|
||||
self.__conn.sendall(payload) # type: ignore
|
||||
data = ''
|
||||
while True:
|
||||
part = self.__conn.recv(1024).decode('iso-8859-1') # type: ignore
|
||||
if not part:
|
||||
break
|
||||
data += part
|
||||
self.__conn.close() # type: ignore
|
||||
return data
|
||||
|
||||
def __generatePayload(self, payload_name: str, data: dict) -> bytes:
|
||||
payload = self.payload[payload_name]
|
||||
payload = payload.replace("#COMKEY", self.__comkey)
|
||||
payload = payload.replace(data['key'], data['value'])
|
||||
|
||||
payload = f"POST /iWsService HTTP/1.0\r\nContent-Type: text/xml\r\nContent-Length: {len(payload)}\r\n\r\n{payload}\r\n" # noqa: E501
|
||||
return bytes(payload, 'utf-8')
|
||||
|
||||
def __getValueFromTag(self, data, start, end):
|
||||
res = data.split(start)[1].split(end)[0]
|
||||
return res
|
||||
|
||||
def __dateRange(self, startDate: str, endDate: str) -> List[str]:
|
||||
start = datetime.datetime.strptime(startDate, "%Y-%m-%d")
|
||||
end = datetime.datetime.strptime(endDate, "%Y-%m-%d")
|
||||
date_generated = [
|
||||
(start + datetime.timedelta(days=x)).strftime("%Y-%m-%d")
|
||||
for x in range(0, (end-start).days+1)
|
||||
]
|
||||
return date_generated
|
||||
|
||||
def deleteAttlog(
|
||||
self,
|
||||
pin: Union[str, list] = "all",
|
||||
date_start: Optional[str] = None,
|
||||
date_end: Optional[str] = None
|
||||
):
|
||||
self.connect()
|
||||
if date_start is not None and date_end is None:
|
||||
date_end = date_start
|
||||
|
||||
self.disable_device()
|
||||
self.clear_attendance()
|
||||
|
||||
#if isinstance(pin, list):
|
||||
# listpin = ""
|
||||
# for pinid in pin:
|
||||
# listpin += "<Arg><PIN>"+pinid+"</PIN></Arg>"
|
||||
# pin = listpin
|
||||
#else:
|
||||
# pin = "<Arg><PIN>"+pin+"</PIN></Arg>"
|
||||
|
||||
#payload = self.__generatePayload(
|
||||
# 'GetAttLog',
|
||||
# {'key': "#PIN", 'value': pin}
|
||||
#)
|
||||
#data = self.__send(payload)
|
||||
|
||||
#if "<GetAttLogResponse>" not in data:
|
||||
# return None
|
||||
#return self.__parseAttendanceData(data, date_start, date_end)
|
||||
|
||||
|
||||
class UserInfoEncoder(json.JSONEncoder):
|
||||
def default(self, obj):
|
||||
if isinstance(obj, UserInfo):
|
||||
return {
|
||||
"pin": obj.pin,
|
||||
"name": obj.name,
|
||||
"password": obj.password,
|
||||
"group": obj.group,
|
||||
"privilege": obj.privilege,
|
||||
"card": obj.card,
|
||||
"pin2": obj.pin2
|
||||
}
|
||||
return super(UserInfoEncoder, self).default(obj)
|
||||
|
||||
class UserAttendanceEncoder(json.JSONEncoder):
|
||||
def default(self, obj):
|
||||
if isinstance(obj, UserAttendance):
|
||||
return {
|
||||
"pin": obj.pin,
|
||||
"datetime": obj.datetime,
|
||||
"verified": obj.verified,
|
||||
"status": obj.status,
|
||||
"workcode": obj.workcode
|
||||
}
|
||||
return super(UserAttendanceEncoder, self).default(obj)
|
||||
|
||||
# Split date dan time, ubah format tanggal ke DD-MM-YYY
|
||||
def split_datetime(datetime_str):
|
||||
dt_obj = datetime.datetime.strptime(datetime_str, '%Y-%m-%d %H:%M:%S')
|
||||
date = dt_obj.strftime('%d-%m-%Y') # Convert date to DD-MM-YYYY format
|
||||
time = dt_obj.strftime('%H:%M') # Extract time '%H:%M:%S'
|
||||
return date, time
|
||||
|
||||
class UserAttendanceEncoderCSV():
|
||||
@staticmethod
|
||||
def encode_to_csv(user_attendance_list, filename):
|
||||
# Define the header for the CSV file
|
||||
#header = ["machineID","pin", "date", "time", "verified", "status", "workcode"]
|
||||
|
||||
# Writing to the CSV file
|
||||
with open(filename, mode='w', newline='') as file:
|
||||
writer = csv.writer(file, delimiter=';')
|
||||
|
||||
# Write the header
|
||||
#writer.writerow(header)
|
||||
|
||||
# Write each user's attendance data
|
||||
for user_attendance in user_attendance_list:
|
||||
print(user_attendance.pin)
|
||||
date, time = split_datetime(user_attendance.datetime)
|
||||
writer.writerow([
|
||||
'001',
|
||||
user_attendance.pin.zfill(16),
|
||||
date,
|
||||
time,
|
||||
#user_attendance.verified,
|
||||
user_attendance.status.zfill(2),
|
||||
user_attendance.workcode.zfill(2)
|
||||
])
|
||||
Reference in New Issue
Block a user