diff --git a/.gitignore b/.gitignore index 5d70719..b41aa53 100644 --- a/.gitignore +++ b/.gitignore @@ -158,3 +158,6 @@ venv.bak/ # mypy .mypy_cache/ + +# VScode +.vscode/ diff --git a/README.md b/README.md index a4c6951..4d02652 100644 --- a/README.md +++ b/README.md @@ -107,7 +107,7 @@ conn.set_time(newtime) ``` -* Ger Firmware Version and Extra Information +* Get Firmware Version and Extra Information ```python conn.get_firmware_version() @@ -176,10 +176,25 @@ zk.enroll_user('1') ```python # Get attendances (will return list of Attendance object) attendances = conn.get_attendance() +sorted_attendances = conn.get_sorted_attendance(by_date=False) # means sorting by uid +limited_attendances = conn.get_limited_attendance( + users=[1, 2], # only UIDs 1, 2 + start_date=datetime(2022, 1, 10, 12, 42), # from 2022,1,10 12:42:00 + end_date=datetime(2022, 1, 11) # to 2022,1,11 +) + # Clear attendances records conn.clear_attendance() ``` +* User history +```python +# Get the history of users records +hist = conn.get_user_history( + users=[1, 2], start_date=datetime(2022, 1, 10, 12, 42) +) +``` + * Test voice ```python diff --git a/zk/attendance.py b/zk/attendance.py index 3ffd9d7..821258b 100644 --- a/zk/attendance.py +++ b/zk/attendance.py @@ -2,13 +2,34 @@ class Attendance(object): def __init__(self, user_id, timestamp, status, punch=0, uid=0): self.uid = uid # not really used any more - self.user_id = user_id - self.timestamp = timestamp - self.status = status - self.punch = punch + self._user_id = user_id + self._timestamp = timestamp + self._status = status + self._punch = punch def __str__(self): - return ': {} : {} ({}, {})'.format(self.user_id, self.timestamp, self.status, self.punch) + return ': {} : {} ({}, {})'.format(self._user_id, self._timestamp, + self._status, self._punch) def __repr__(self): - return ': {} : {} ({}, {})'.format(self.user_id, self.timestamp,self.status, self.punch) + return ': {} : {} ({}, {})'.format(self._user_id, self._timestamp, + self._status, self._punch) + + def __call__(self): + return self._user_id, self._timestamp, self._status, self._punch + + @property + def user_id(self): + return self._user_id + + @property + def timestamp(self): + return self._timestamp + + @property + def status(self): + return self._status + + @property + def punch(self): + return self._punch diff --git a/zk/base.py b/zk/base.py index e4aa247..774afec 100644 --- a/zk/base.py +++ b/zk/base.py @@ -1,15 +1,19 @@ # -*- coding: utf-8 -*- import sys +import codecs + from datetime import datetime from socket import AF_INET, SOCK_DGRAM, SOCK_STREAM, socket, timeout from struct import pack, unpack -import codecs +from itertools import groupby +from typing import Union from . import const from .attendance import Attendance from .exception import ZKErrorConnection, ZKErrorResponse, ZKNetworkError from .user import User from .finger import Finger +from .utility import Utility def safe_cast(val, to_type, default=None): @@ -1640,6 +1644,78 @@ def get_attendance(self): attendance_data = attendance_data[40:] return attendances + def get_sorted_attendance(self, by_date: bool = False) -> list: + """ + Sorting attendances record wheather by date or uid. + + :param by_date: If it is True, means sorting by date. Else it will be sorted by + uid. + :return: Sorted records. + """ + return sorted(self.get_attendance(), key=lambda x: x()[int(by_date)]) + + def get_limited_attendance( + self, users: list = [], + start_date=None, end_date=None + ) -> list: + """ + Filter attendances' records with both user selection and/or datetime. + + :param users: List of users by uid or name. + :param start: The filter starts from this datetime. + :param end: The filter ends up to this datetime. + :return: Limited attendances record by the respective input filters. + """ + try: + attendances = self.get_sorted_attendance() + + if users: + attendances = Utility.filter_by_user(attendances, users) + if start_date is not None: + attendances = Utility.filter_by_date(attendances, start=start_date) + if end_date is not None: + attendances = Utility.filter_by_date(attendances, end=end_date) + + return attendances + + except Exception as e: + print(e) + raise ZKErrorResponse("Something went wrong!") + + def get_user_history( + self, users: list = [], + start_date=None, end_date=None + ) -> dict: + """ + Returns the history of attendances which is grouped by their uid. + + :param users: List of users by uid. + :param start: The filter starts from this datetime. + :param end: The filter ends up to this datetime. + :return: Grouped by attendances records including date, status, and punch + """ + history = {} + + def key_func(k): + """Returns the UID of each Attendances as a key.""" + return k.user_id + + _attendances = self.get_limited_attendance( + users=users, start_date=start_date, end_date=end_date + ) + j, i = 0, 0 + for k, g in groupby(_attendances, key_func): + '''Group by Attendance ID''' + history[f'Attendance {k}'] = list(g) + + for i, _ in enumerate(history[f'Attendance {k}']): + '''Select datetime, status, and punch from the tupple.''' + history[f'Attendance {k}'][i] = _attendances[i + j]()[1:] + + j += i + 1 + + return history + def clear_attendance(self): """ clear all attendance record diff --git a/zk/utility.py b/zk/utility.py new file mode 100644 index 0000000..f8dc1dd --- /dev/null +++ b/zk/utility.py @@ -0,0 +1,44 @@ +class Utility(object): + """Utility Class""" + + def __init__(self): + pass + + @staticmethod + def filter_by_date(attendances_list, start=None, end=None): + """ + Select only desire attendances records by datetime condition. + + :param attendances_list: The list of attendances records. + :param start: The filter of starting datetime + :param end: The filter of ending datetime + :return: Limited list of attendances records. + """ + filtered = [] + + if start is not None: + for d in attendances_list: + if d.timestamp >= start: + filtered.append(d) + + if end is not None: + for d in attendances_list: + if d.timestamp <= end: + filtered.append(d) + + return filtered + + @staticmethod + def filter_by_user(attendances_list, users_list): + """ + Select only desire attendances records by user condition. + + :param attendances_list: The list of attendances records. + :param users_list: The list of desire users to be select. + :return: Limited list of attendances records. + """ + return [ + item + for item in attendances_list + if item.user_id in list(map(str, users_list)) + ]