Python-Projekte/crewmeister.py

import requests
import pandas as pd
import datetime
from colorama import Fore
from colorama import Style


def pt_to_hours(x):
    delta = int(x[2:-1]) / 3600
    return round(delta, 2)
    # return str(round(delta)) + ":" + str(round((delta - round(delta)) * 60))


def identity(x):
    return x


def type_id(x):
    if x == 1:
        return "Urlaub"
    if x == 2:
        return "Krank"
    if x == 3:
        return "Freizeit"
    return "anwesend"


class crewmeister:
    webservice = "https://api.crewmeister.com/api/v2"

    user_name = "bedner@global-cube.de"
    user_pass = "7pmicg1w"
    crew_id = "26515"

    suffix_count = 0

    def auth(self):
        cookies = {}
        r = requests.post(self.webservice + "/user/authentication", {'userIdentifier': self.user_name, 'password': self.user_pass})
        payload = r.json()['payload']
        cookies['cmAuthenticationUserToken'] = payload['token']
        cookies['cmAuthenticationUserId'] = str(payload['id'])

        r = requests.post(self.webservice + "/crew/authentication", {'crewId': self.crew_id}, cookies=cookies)
        payload = r.json()['payload']
        cookies['cmAuthenticationCrewToken'] = payload['token']
        cookies['cmAuthenticationCrewId'] = payload['id']
        return cookies

    def userlist(self):
        r = requests.get(self.webservice + f"/crew/{self.crew_id}/member", cookies=self.cookies)
        return pd.DataFrame(r.json()['payload']).set_index("userId")

    def add_to_userlist(self, df):
        self.users = self.users.join(df, rsuffix="_" + str(self.suffix_count))
        self.suffix_count += 1

    def __init__(self):
        self.cookies = self.auth()
        self.users = self.userlist()

    def cm_request(self, querystring, params, cols):
        r = requests.get(self.webservice + querystring, params, cookies=self.cookies)
        try:
            df = pd.DataFrame(r.json()['payload'])
            if 'userId' not in df.columns:
                df['userId'] = df['groupBy'].map(lambda x: x['userId'])
            for (key, (col, map_fun)) in cols.items():
                df[key] = df[col].map(map_fun)
            items = list(cols.keys())
            items.append("userId")
            self.add_to_userlist(df.filter(items=items).set_index("userId"))
            return True
        except Exception as e:
            print(r.json())
            print(e)
            return self.zerofill(cols)

    def zerofill(self, cols):
        for key in cols.keys():
            self.users[key] = 0
        return False

    def statistics(self):
        timestamp = datetime.datetime.now().isoformat()
        today = timestamp[:10]
        first_of_year = today[:4] + "-01-01"
        end_of_year = today[:4] + "-12-31"
        prev_month = (datetime.date.fromisoformat(today[:7] + "-01") + datetime.timedelta(days=-1)).isoformat()[:10]

        r = requests.get(self.webservice + f"/crew/{self.crew_id}/time-tracking/stamps", {'startTime': timestamp, 'endTime': timestamp}, cookies=self.cookies)
        stamps = pd.DataFrame(r.json()['payload'])

        if len(stamps.index) > 0:
            self.add_to_userlist(stamps.query("timeAccount==1").groupby('userId').max())

        self.cm_request(f"/crew/{self.crew_id}/absence-management/absences", {'from': today, 'to': today}, {'absence_today': ("typeId", identity)})
        self.cm_request(f"/crew/{self.crew_id}/absence-management/absence-balances", {'date': first_of_year, 'typeId': 1}, {'absence_vac_prev': ("value", identity)})
        self.cm_request(f"/crew/{self.crew_id}/absence-management/absence-balances", {'date': today, 'typeId': 1}, {'absence_vacation': ("value", identity)})
        self.cm_request(f"/crew/{self.crew_id}/absence-management/absence-balances", {'date': end_of_year, 'typeId': 1}, {'absence_planned': ("value", identity)})
        self.cm_request(f"/crew/{self.crew_id}/duration-balances?groupBy%5B%5D=user_id&date%5B%5D={prev_month}", {}, {'duration_prev_month': ("value", pt_to_hours)})
        self.cm_request(f"/crew/{self.crew_id}/duration-balances?groupBy%5B%5D=user_id&date%5B%5D={today}", {}, {'duration_today': ("value", pt_to_hours)})
        return self.users


if __name__ == '__main__':
    cm = crewmeister()
    user = cm.statistics().loc[86043]
    print("Gerade anwesend: " + user['clockInTimestamp'])
    print("geplante Abwesenheit heute: " + type_id(user['absence_today']))
    print("Urlaub Jahresanfang: " + str(user['absence_vac_prev']))
    print("Urlaub genommen: " + str(user['absence_vacation']))
    print("Urlaub geplant: " + str(user['absence_planned']))
    print("Überstunden Vormonat: " + str(user['duration_prev_month']))
    print("Überstunden aktuell: " + str(user['duration_today']))

    print(f"{Fore.GREEN}Test!{Style.RESET_ALL}")