introduction

Google Agenda est très pratique, n'est-ce pas? Il peut être lié à Gmail, peut être ouvert à partir d'un PC et est simple. Cependant, il y a quelques points d'insatisfaction. C'est-à-dire qu'il ne calcule pas le salaire de l'emploi à temps partiel. Je souhaite que cela fonctionne avec l'application de gestion des équipes. .. .. Cela dit, j'ai essayé de le faire en utilisant diverses API et AWS qui m'intéressaient en tant que débutant.

Ce que j'ai utilisé

Slack
AWS
Google Agenda
Python

Vue d'ensemble de l'architecture

J'ai créé un bot avec la configuration suivante.

procédure

Créez une application pour créer un bot sur le site API Slack
Créez un point de terminaison sur AWS qui reçoit les événements Slack
Créez un programme pour calculer le salaire en octets à partir du calendrier Google à l'aide de l'API Google
Introduction à AWS lambda

[Procédure 1] [Procédure 2] Créez un bot Slack et créez un point de terminaison sur AWS

Cette procédure a été très utile pour l'article suivant. Si vous suivez l'article, vous pouvez le faire sans aucun problème. Même les débutants AWS peuvent comprendre! Terminé sur le navigateur! Super introduction à Slack Bot à l'aide d'AWS + API d'événement Slack

[Procédure 3] Créez un programme pour calculer le salaire d'octets de Google Agenda

Je me suis référé à l'article suivant pour savoir comment obtenir le calendrier du calendrier Google à l'aide de Python. Obtenir les rendez-vous Google Agenda à partir de Python 3

Ensuite, reportez-vous à Official Sample et Official Reference. ， Nous allons faire un programme pour calculer le salaire.

Le programme est divisé en gros comme suit.

handle_slack_event (): point d'entrée
MakePayMsg (): Calcule le salaire correspondant au texte de l'utilisateur et crée un message
CalculatePay (): Calculer le salaire journalier

Le code complet peut être trouvé sur github, veuillez donc vous y référer.

handle_slack_event()

C'est le point d'entrée. Il analyse le texte envoyé par l'utilisateur et publie un message basé sur celui-ci.

# -----Point d'accès-----
def handle_slack_event(slack_event, context):

    #Exporter les informations d'événement reçues dans le journal Cloud Watch
    logging.info(json.dumps(slack_event))

    #Authentification API d'événement
    if "challenge" in slack_event:
        return slack_event.get("challenge")

    #Autres que les événements de bot ou les événements de publication de messages
    #Revenir tel quel pour ne pas réagir
    #Je dois renvoyer une réponse à Slack, donc je retourne OK
    #(S'il n'est pas retourné, il sera considéré comme un échec et la même demande sera envoyée plusieurs fois)
    if is_bot(slack_event) or not is_message_event(slack_event):
        return "OK"

    #Extraire le texte du message de l'utilisateur
    text = slack_event.get("event").get("text")

    #Déclaration de classe de paie
    pay_msg = MakePayMsg()
    
    #Rédiger un message en analysant le texte de l'utilisateur
    if 'help' in text:
        msg = 'Entrez le numéro correspondant aux informations que vous souhaitez connaître!\n'
        msg += '(1)Salaire du mois prochain\n'
        msg += '(2)Salaire de cette année\n'
        msg += '(3)Journal des salaires\n'
    elif text == '1':
        msg = 'Le salaire du mois prochain est de ¥{}est!'.format(pay_msg.monthpay())
    elif text == '2':
        msg = '{}'.format(pay_msg.yearpay())
    elif text == '3':
        msg = 'Journal des salaires\n{}'.format(pay_msg.paylog())
    else:
        msg = '\\Quai/'
    
    #Publier un message
    post_message_to_slack_channel(msg, slack_event.get("event").get("channel"))

    #Suite à une demande de l'API Event, en plus de publier le message
    #Je dois renvoyer une réponse à Slack, donc je retourne OK
    #(S'il n'est pas retourné, il sera considéré comme un échec et la même demande sera envoyée plusieurs fois)
    return "OK"

MakePayMsg() Calculez le salaire correspondant au texte de l'utilisateur et créez un message.

# -----Calculer le salaire et rédiger un message-----
class MakePayMsg():
    def __init__(self):
    
        self.SCOPES = ['https://www.googleapis.com/auth/calendar.readonly']
        self.now = datetime.datetime.now()
        self.events = self.get_event() #Événements récupérés de Google Agenda
        self.pay_log = self.make_paylog() #Journal des salaires de cette année

    # ---Récupération des événements de Google Agenda---
    def get_event(self):
        creds = None
        if os.path.exists('token.pickle'):
            with open('token.pickle', 'rb') as token:
                creds = pickle.load(token)
        # If there are no (valid) credentials available, let the user log in.
        if not creds or not creds.valid:
            if creds and creds.expired and creds.refresh_token:
                creds.refresh(Request())
            else:
                flow = InstalledAppFlow.from_client_secrets_file(
                    'credentials.json', self.SCOPES)
                creds = flow.run_local_server(port=0)
            # Save the credentials for the next run
            with open('/tmp/token.pickle', 'wb') as token:
                pickle.dump(creds, token)

        service = build('calendar', 'v3', credentials=creds)

        #Sélectionnez le calendrier sur lequel le décalage d'octet est enregistré
        calender_id = os.environ['CALENDER_ID']

        page_token = None
        events = service.events().list(calendarId=calender_id, pageToken=page_token).execute()
        return events

    # ---Créez un journal des salaires pour cette année---
    def make_paylog(self):
        pay_log = []
        cal = CalculatePay(1013, 1063, 1.25, 22) #Entrez les informations de salaire horaire
        
        #Extraire l'heure de début et l'heure de fin de l'octet de l'événement et calculer le salaire
        for event in self.events['items']:
            #Transformez l'heure de début et l'heure de fin en datetime
            stime = event['start']['dateTime']
            stime = datetime.datetime(
                int(stime[0:4]), int(stime[5:7]), int(stime[8:10]),
                int(stime[11:13]), int(stime[14:16]))
            etime = event['end']['dateTime']
            etime = datetime.datetime(
                int(etime[0:4]), int(etime[5:7]), int(etime[8:10]),
                int(etime[11:13]), int(etime[14:16]))
            
            #Période de calcul de la paie
            # (x-1)décembre~La quantité de travail en novembre x est le salaire de x
            if self.now.month != 12:
                sdate = datetime.date(self.now.year-1, 12, 1)
                edate = datetime.date(self.now.year, 11, 30)
            else:
                sdate = datetime.date(self.now.year, 12, 1)
                edate = datetime.date(self.now.year+1, 11, 30)

            #Enregistrer le salaire d'un an sous forme de journal
            if (stime.date() >= sdate) and (etime.date() <= edate):
                #Calcul du salaire d'une journée à partir de l'heure de début et de fin
                daypay = cal.calculate(stime, etime)
                #Ajuster pour que la quantité de travail soit le salaire du mois suivant
                if stime.month==12:
                    daypay_dir = {'date':stime.date(), 'month':1, 'pay':daypay}
                else:
                    daypay_dir = {'date':stime.date(), 'month':stime.month+1, 'pay':daypay}
                pay_log += [daypay_dir]
        pay_log = sorted(pay_log, key=lambda x:x['date'])
        return pay_log
            
    # ---Créez un message pour afficher le salaire du mois prochain---
    def monthpay(self):
        mpay = 0
        for i in self.pay_log:
            if self.now.month!=12:
                if i['month'] == (self.now.month+1):
                    mpay += i['pay']
            else:
                if i['month'] == 1:
                    mpay += i['pay']
        return mpay

    # ---Créer un message pour afficher le salaire pendant un an---
    def yearpay(self):
        mpay_list = [0] * 12
        for i in self.pay_log:
            mpay_list[i['month']-1] += i['pay']
        msg = ''
        for i, mpay in enumerate(mpay_list):
            msg += '{}Mois ¥{:,}\n'.format(i+1, mpay)
        msg += '\n total ¥{}'.format(sum(mpay_list))
        return msg

    # ---Créez un message pour afficher le journal d'un an---
    def paylog(self):
        msg = ''
        month = 0
        for i in self.pay_log:
            while i['month'] != month:
                msg += '\n{}Mois\n'.format(month+1)
                month += 1
            msg += '{} ￥{:,}\n'.format(i['date'], i['pay'])
        return msg

Notez que Lambda ne peut écrire que dans des fichiers sous / tmp. Par conséquent, lors de l'écriture dans un certain fichier, on peut dire que «[Errno 30] Système de fichiers en lecture seule» se produit lors de l'exécution avec Lambda, même si aucune erreur ne s'est produite lors de l'exécution locale. Puisqu'une erreur s'est également produite dans ce programme, je l'ai modifiée comme suit.

`Changer avant`


with open('token.pickle', 'wb') as token:
    pickle.dump(creds, token)

`Après le changement`


with open('/tmp/token.pickle', 'wb') as token:
    pickle.dump(creds, token)

CalculatePay() Calculez le salaire journalier.


# -----Calculer le salaire journalier-----
class CalculatePay():
    def __init__(
        self, basic_pay, irregular_pay, night_rate, night_time):

        self.basic_pay = basic_pay #Salaire horaire en semaine
        self.irregular_pay = irregular_pay #Salaire horaire les samedis, dimanches et jours fériés
        self.night_rate = night_rate #Taux d'augmentation de salaire de fin de soirée
        self.night_time = night_time #Il est temps d'obtenir un salaire de minuit

    # ---Calculer le salaire quotidien---
    def calculate(self, stime, etime):
        night_time = datetime.datetime(stime.year, stime.month, stime.day, self.night_time)
        
        if stime.weekday() >= 5 or jpholiday.is_holiday(stime.date()):
            pay = self.irregular_pay
        else:
            pay = self.basic_pay

        if etime >= night_time:
            normal_time = self.get_h(night_time - stime)
            night_time = self.get_h(etime - night_time)
            daypay = normal_time * pay + night_time * (pay * self.night_rate)
        else:
            normal_time = self.get_h(etime - stime)
            daypay = normal_time * pay

        return round(daypay)

    # ---Convertir à partir de l'affichage de l'heure x heures y minutes en h---
    def get_h(self, delta_time):
        h = delta_time.seconds / 3600
        return h

[Procédure 4] Introduit dans AWS lambda

J'ai installé des modules pour écrire un programme pour calculer le salaire à partir de Google Agenda. Si j'exécute ce programme sur AWS Lambda sans rien faire, j'obtiens un ModuleNotFoundError. Créez un package de déploiement AWS Lambda pour Python afin que les différents modules puissent également être utilisés sur AWS Lambda. En termes simples, installez tous les modules dépendants dans le répertoire de votre projet et fermez-les avec votre exécutable. Cela peut être fait selon le site suivant. [Python] Utilisation d'un module externe avec AWS Lambda

Résultat d'exécution

J'ai récemment commencé à enregistrer des décalages d'octets dans Google Agenda, il n'y a donc pas beaucoup de données. Depuis que j'ai introduit ce bot, j'aimerais gérer les changements avec Google Agenda à l'avenir. (C'est une icône facile à comprendre où se trouve la destination de l'octet)