371 lines
11 KiB
Python
371 lines
11 KiB
Python
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)
|
|
]) |