Initial commit - Crew Portal API

.gitignore

@@ -0,0 +1,4 @@
+venv/
+*.pyc
+__pycache__/
+settings.ini

classes/adonis.py

@@ -0,0 +1,76 @@
+from urllib.request import urlopen, Request, urljoin
+
+import json
+
+from classes.exceptions import AdonisAuthenticationDeniedError, AdonisError
+from classes.crew import Crew
+
+from const import USER_AGENT
+
+
+class Adonis:
+    def __init__(self, crew_portal_base_url: str, login: str, password: str):
+        self.crew_portal_base_url = crew_portal_base_url
+        self.login = login
+        self.password = password
+
+    @classmethod
+    def fromConfig(cls, config):
+        return cls(config.crew_portal_base_url, config.login, config.password)
+
+    def requestCrewPortal(self, endpoint: str, payload: dict, add_token: bool = True, wrap_request: bool = True, extract_response: bool = True):
+        url = urljoin(self.crew_portal_base_url, endpoint)
+
+        req = Request(url)
+        req.add_header("User-Agent", USER_AGENT)
+        req.add_header("Content-Type", "application/json")
+        req.add_header("Accept", "application/json")
+
+        if add_token:
+            payload["Authentication_Token"] = self.getToken()
+
+        if wrap_request:
+            payload = {"request": payload}
+
+        req.data = json.dumps(payload).encode()
+
+        res = json.loads(urlopen(req).read())
+
+        if extract_response:
+            res = res[endpoint + "Result"]
+
+        if not res["Authentication_Approved"]:
+            raise AdonisAuthenticationDeniedError(
+                res["Authentication_ReasonDenied"])
+
+        if res["ErrorText"]:
+            raise AdonisError(res["ErrorText"])
+
+        return res
+
+    def getTokenCrewPortal(self, lifetime: int = 60):
+        data = {
+            "credentials": {
+                "Login": self.login,
+                "Password": self.password,
+                "LifeTime": lifetime
+            }
+        }
+
+        result = self.requestCrewPortal("GNL_API_AUTHENTICATION", data, False, False)
+
+        return result["Authentication_Token"]
+
+    def getCrewPersonalData(self, pin):
+        data = {
+            "Pin": pin
+        }
+
+        result = self.requestCrewPortal("DG_PersonalInformationRead", data)
+
+        return result
+
+    def getCrew(self, pin):
+        crew = Crew(pin)
+        crew._personal_data = self.getCrewPersonalData(pin)
+        return crew

classes/config.py

@@ -0,0 +1,18 @@
+from configparser import ConfigParser
+
+class Config:
+    def __init__(self, path):
+        self.config = ConfigParser()
+        self.config.read(path)
+
+    @property
+    def crew_portal_base_url(self):
+        return self.config["ADONIS"]["CrewPortalBaseURL"]
+
+    @property
+    def login(self):
+        return self.config["ADONIS"]["Login"]
+
+    @property
+    def password(self):
+        return self.config["ADONIS"]["Password"]

classes/crew.py

@@ -0,0 +1,20 @@
+class Crew:
+    def __init__(self, pin):
+        self.pin = pin
+        self._personal_data = None
+
+    @property
+    def firstName(self):
+        return self._personal_data["DetailsAll"]["Details"]["FirstName"]
+
+    @property
+    def middleName(self):
+        return self._personal_data["DetailsAll"]["Details"]["MiddleName"]
+
+    @property
+    def lastName(self):
+        return self._personal_data["DetailsAll"]["Details"]["LastName"]
+
+    @property
+    def birthDate(self):
+        return self._personal_data["DetailsAll"]["Details"]["BirthDate"]

classes/exceptions.py

@@ -0,0 +1,5 @@
+class AdonisError(Exception):
+    pass
+
+class AdonisAuthenticationDeniedError(AdonisError):
+    pass

const.py

@@ -0,0 +1 @@
+USER_AGENT = "PyAdonis / dev (https://kumi.systems)"

settings.dist.ini

@@ -0,0 +1,4 @@
+[ADONIS]
+CrewPortalBaseURL=http://adonis/AdonisWebServices/CrewPortalWebService.svc/
+Login=Adonis_API
+Password=

worker.py

@@ -0,0 +1,30 @@
+from argparse import ArgumentParser
+
+from classes.config import Config
+from classes.adonis import Adonis
+
+
+if __name__ == "__main__":
+    parser = ArgumentParser()
+    parser.add_argument("pin", nargs="+")
+    args = parser.parse_args()
+
+    config = Config("settings.ini")
+    api = Adonis.fromConfig(config)
+
+    data = dict()
+
+    for pin in args.pin:
+        crewdata = dict()
+
+        crew = api.getCrew(pin)
+
+        crewdata["firstName"] = crew.firstName
+        crewdata["middleName"] = crew.middleName
+        crewdata["lastName"] = crew.lastName
+        crewdata["dob"] = crew.birthDate
+        crewdata["password"] = crew.birthDate.replace("-","")
+
+        data[pin] = crewdata
+
+    print(data)