From 2c7229d58eec816aed572e79f3e0b3d6a3cd4892 Mon Sep 17 00:00:00 2001 From: Kumi Date: Mon, 3 Jul 2023 11:05:13 +0200 Subject: [PATCH] More functions, blacked --- pyproject.toml | 2 +- requirements.txt | 3 +- src/pyadonis/classes/adonis.py | 165 ++++++++++++++++++++++++----- src/pyadonis/classes/config.py | 1 + src/pyadonis/classes/crew.py | 8 +- src/pyadonis/classes/exceptions.py | 3 +- src/pyadonis/pindata.py | 2 +- 7 files changed, 154 insertions(+), 30 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 48931d6..0e97107 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "pyadonis" -version = "0.9.5" +version = "0.9.6" authors = [ { name="Kumi Systems e.U.", email="office@kumi.systems" }, ] diff --git a/requirements.txt b/requirements.txt index 037103e..77e4a79 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,2 @@ -pillow \ No newline at end of file +pillow +python-dateutil diff --git a/src/pyadonis/classes/adonis.py b/src/pyadonis/classes/adonis.py index 312d98a..2280381 100644 --- a/src/pyadonis/classes/adonis.py +++ b/src/pyadonis/classes/adonis.py @@ -15,7 +15,13 @@ INTEGRATION = 1 class Adonis: - def __init__(self, crew_portal_base_url: str, integration_base_url: str, login: str, password: str): + def __init__( + self, + crew_portal_base_url: str, + integration_base_url: str, + login: str, + password: str, + ): self.crew_portal_base_url = crew_portal_base_url self.integration_base_url = integration_base_url self.login = login @@ -23,10 +29,27 @@ class Adonis: @classmethod def fromConfig(cls, config): - return cls(config.crew_portal_base_url, config.integration_base_url, config.login, config.password) + return cls( + config.crew_portal_base_url, + config.integration_base_url, + config.login, + config.password, + ) - def request(self, endpoint: str, payload: dict, add_token: bool = True, wrap_request: bool = True, extract_response: bool = True, api: int = CREW_PORTAL): - base_url = self.crew_portal_base_url if api == CREW_PORTAL else self.integration_base_url + def request( + self, + endpoint: str, + payload: dict, + add_token: bool = True, + wrap_request: bool = True, + extract_response: bool = True, + api: int = CREW_PORTAL, + ): + base_url = ( + self.crew_portal_base_url + if api == CREW_PORTAL + else self.integration_base_url + ) url = urljoin(base_url, endpoint) req = Request(url) @@ -48,48 +71,62 @@ class Adonis: res = res[endpoint + "Result"] if not res["Authentication_Approved"]: - raise AdonisAuthenticationDeniedError( - res["Authentication_ReasonDenied"]) + raise AdonisAuthenticationDeniedError(res["Authentication_ReasonDenied"]) if res["ErrorText"]: raise AdonisError(res["ErrorText"]) return res + def requestIntegration( + self, + endpoint: str, + payload: dict, + add_token: bool = True, + wrap_request: bool = True, + extract_response: bool = True, + ): + return self.request( + endpoint, payload, add_token, wrap_request, extract_response, INTEGRATION + ) - def requestIntegration(self, endpoint: str, payload: dict, add_token: bool = True, wrap_request: bool = True, extract_response: bool = True): - return self.request(endpoint, payload, add_token, wrap_request, extract_response, INTEGRATION) - - def requestCrewPortal(self, endpoint: str, payload: dict, add_token: bool = True, wrap_request: bool = True, extract_response: bool = True): - return self.request(endpoint, payload, add_token, wrap_request, extract_response, CREW_PORTAL) + def requestCrewPortal( + self, + endpoint: str, + payload: dict, + add_token: bool = True, + wrap_request: bool = True, + extract_response: bool = True, + ): + return self.request( + endpoint, payload, add_token, wrap_request, extract_response, CREW_PORTAL + ) def getToken(self, lifetime: int = 60, api: int = CREW_PORTAL): data = { "credentials": { "Login": self.login, "Password": self.password, - "LifeTime": lifetime + "LifeTime": lifetime, } } - method = self.requestCrewPortal if api == CREW_PORTAL else self.requestIntegration + method = ( + self.requestCrewPortal if api == CREW_PORTAL else self.requestIntegration + ) result = method("GNL_API_AUTHENTICATION", data, False, False) return result["Authentication_Token"] def getCrewPersonalData(self, pin): - data = { - "Pin": pin - } + data = {"Pin": pin} result = self.requestCrewPortal("DG_GeneralDetailsRead", data) return result def getCrewProfilePicture(self, pin): - data = { - "Pin": pin - } + data = {"Pin": pin} try: result = self.requestCrewPortal("DG_ProfilePictureRead", data) @@ -117,8 +154,8 @@ class Adonis: "Filter": filters, "Pagination": pagination or "None", "RowsByPage": paginate_by, - "Page": page - } + "Page": page, + } try: result = self.requestIntegration("GNL_APMCrewListViews", data) @@ -134,8 +171,10 @@ class Adonis: pnum = 1 while True: - page = self.getCrewListView(view, pagination="True", paginate_by=str(paginate_by), page=str(pnum)) - + page = self.getCrewListView( + view, pagination="True", paginate_by=str(paginate_by), page=str(pnum) + ) + for record in json.loads(page["Result"]): requirements.append((record["PIN"], record["Document code"])) @@ -144,4 +183,82 @@ class Adonis: pnum += 1 - return requirements \ No newline at end of file + return requirements + + def getCodes(self): + try: + return self.requestIntegration("DG_GetCodes") + except AdonisError: + return None + + def writeCompetence( + self, + pin, + competence_code, + certificate_number, + from_timestamp=None, + to_timestamp=None, + issued_by="pyadonis", + attachment=None, + **kwargs, + ): + from_dt = ( + datetime.datetime.fromtimestamp(from_timestamp) + if from_timestamp + else datetime.datetime.now() + ) + from_dt = from_dt.replace(hour=0, minute=0, second=0, microsecond=0) + from_dt_string = from_dt.isoformat() + + if to_timestamp: + to_dt = datetime.datetime.fromtimestamp(to_timestamp) + else: + codes = self.getCodes() + expiry_days = 0 + + for competence in codes["Competence"]: + if competence["Code"] == competence_code: + expiry_days = int(competence["DefaultExpiryDays"]) + + if expiry_days == 0: + to_dt_string = "" + else: + if not expiry_days % 365: + to_dt = from_dt + relativedelta(years=expiry_days / 365) + elif not expiry_days % 30: + to_dt = from_dt + relativedelta(months=expiry_days / 30) + else: + to_dt = from_dt + relativedelta(days=expiry_days) + + to_dt_string = to_dt.isoformat() + + payload = { + "Certificate": { + "Pin": pin, + "Code": competence_code, + "Number": certificate_number, + "DateFrom": from_dt_string, + "ExpiryDate": to_dt_string, + "IssuedBy": issued_by, + "Planned": "False", + } + } + + if attachment: + payload["Certificate"]["FileName"] = "certificate.pdf" + + if isinstance(attachment, bytes): + attachment = attachment.decode() + + payload["Certificate"]["DocumentBase64"] = ( + "data:application/pdf;" + attachment + ) + + payload.update(kwargs) + + return self.requestCrewPortal("DG_CertificateCreate", payload) + + def getCompetences(self, pin): + payload = {"Pin": pin} + + return self.requestCrewPortal("DG_CertificateRead", payload) diff --git a/src/pyadonis/classes/config.py b/src/pyadonis/classes/config.py index 4d1c9ce..cf2e32c 100644 --- a/src/pyadonis/classes/config.py +++ b/src/pyadonis/classes/config.py @@ -1,5 +1,6 @@ from configparser import ConfigParser + class Config: def __init__(self, path): self.config = ConfigParser() diff --git a/src/pyadonis/classes/crew.py b/src/pyadonis/classes/crew.py index 5085dfe..47342a4 100644 --- a/src/pyadonis/classes/crew.py +++ b/src/pyadonis/classes/crew.py @@ -22,8 +22,12 @@ class Crew: @property def email(self): - return self._personal_data["Emails"][0]["Email"] if self._personal_data["Emails"] else "" + return ( + self._personal_data["Emails"][0]["Email"] + if self._personal_data["Emails"] + else "" + ) @property def profilePicture(self): - return self._profile_picture \ No newline at end of file + return self._profile_picture diff --git a/src/pyadonis/classes/exceptions.py b/src/pyadonis/classes/exceptions.py index b792002..60d48d7 100644 --- a/src/pyadonis/classes/exceptions.py +++ b/src/pyadonis/classes/exceptions.py @@ -1,5 +1,6 @@ class AdonisError(Exception): pass + class AdonisAuthenticationDeniedError(AdonisError): - pass \ No newline at end of file + pass diff --git a/src/pyadonis/pindata.py b/src/pyadonis/pindata.py index 136249c..dc59168 100644 --- a/src/pyadonis/pindata.py +++ b/src/pyadonis/pindata.py @@ -24,7 +24,7 @@ if __name__ == "__main__": crewdata["middleName"] = crew.middleName crewdata["lastName"] = crew.lastName crewdata["dob"] = crew.birthDate - crewdata["password"] = crew.birthDate.replace("-","") + crewdata["password"] = crew.birthDate.replace("-", "") crewdata["email"] = crew.email crewdata["profilePicture"] = crew.profilePicture