autoficher/autoficher.py

#!/usr/bin/env python3
# encoding: utf-8
#
# Automarcaje para timenet.gpisoftware.com
# usage: autoficher.py -u USER -p PIN -t TYPE [-f] [-h]
#
# Argumentos obligatorios:
#   -u USER, --user USER  Usuario
#   -p PIN, --pin PIN     Contraseña
#   -t TYPE, --type TYPE  Tipo marcado. 0 = Entrada, 1 = Salida (Si no se utiliza -bt)
#
#
# Creado: Omar Sánchez 04-05-2019

# Importamos librerias
import json
from datetime import datetime
import os
import pytz
import requests
import argparse
import telegram

from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError

SCOPES = ['https://www.googleapis.com/auth/calendar.readonly']

from dateutil.relativedelta import relativedelta

# Introducimos los argumentos obligatorios y opcionales
parser = argparse.ArgumentParser(add_help=False)
parser._action_groups.pop()

obligatoryArgs = parser.add_argument_group("Argumentos obligatorios")
obligatoryArgs.add_argument('-p', '--pin', help="Contraseña", type=int, required=True)
obligatoryArgs.add_argument('-t', '--type', help="Tipo marcado. 0 = Entrada, 1 = Salida", type=int)

optionalArgs = parser.add_argument_group("Argumentos opcionales")
optionalArgs.add_argument('-bt', '--basedtime', help="Hacer check o descheck según la hora del dia",
                          action="store_true")
optionalArgs.add_argument('-h', '--help', action="help", help="Esta ayuda")
optionalArgs.add_argument('-d', '--debug', action='store_true')

optionalArgs.add_argument('-c', '--config', help="Rúta al archivo de configuración")
optionalArgs.add_argument('-u', '--user', help="Usuario")
optionalArgs.add_argument('-glo', '--geoLongitude', help="GEO Longitud", type=float)
optionalArgs.add_argument('-gla', '--geoLatitude', help="GEO Latitud", type=float)
optionalArgs.add_argument('-ga', '--geoAccuracy', help="GEO Accuracy", type=float)

args = parser.parse_args()

if not args.type and not args.basedtime:
    print("Hay que indicar al menos un metodo de fichaje -t <0 o 1> o -bt")
    exit(20)

ROOT_DIR = os.path.dirname(os.path.abspath(__file__))

class AutoFicher():
    # URL de la api
    headers = {
        "Content-type": "application/x-www-form-urlencoded; charset=UTF-8",
        "tn-v": "3.0.9",
        "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/93.0.4577.82 Safari/537.36"
    }
    config = None

    def __init__(self):
        try:
            self.loadConfig()
        except:
            self.sendReport("No se ha podido cargar el archivo de configuración")
            exit(20)

    def loadConfig(self):
        ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
        path = os.path.join(ROOT_DIR, 'config.json')

        if args.config:
            path = args.config

        with open(path, 'r') as f:
            self.config = json.load(f)

            if args.geoLatitude:
                self.config['geo']['latitude'] = args.geoLatitude
            if args.geoLongitude:
                self.config['geo']['longitude'] = args.geoLongitude
            if args.user:
                self.config['user'] = args.user
            if args.geoAccuracy:
                self.config['geo']['accuracy'] = args.geoAccuracy

    def sendReport(self, message):
        print(message)
        bot = telegram.Bot(token=self.config['telegram']['token'])
        bot.sendMessage(chat_id=self.config['telegram']['chat_id'], text=str(message))

    def get(self, url, data={}, headers={}):
        furl = self.config['api_url'] + url
        if url != "version":
            furl = self.config['api_url'] + "v1/" + url
        return requests.get(furl, data=data, headers=headers)

    def post(self, url, data={}, headers={}):
        furl = self.config['api_url'] + url
        if url != "version":
            furl = self.config['api_url'] + "v1/" + url
        return requests.post(furl, data=data, headers=headers)

    def updateTime(self):
        self.headers["tn-d"] = "\"" + datetime.now().astimezone(pytz.UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + "\""

    def getVersion(self):
        self.updateTime()
        return self.get('version', self.headers)

    def getInfo(self):
        return self.get('check/info', {"guid": args.user})

    def sendUpdate(self):
        self.updateTime()

        calendar = self.calendarCheck()
        if calendar:
            self.sendReport("🟥🕐 Comprobación de calendario: " + calendar)
            return

        typ = 0
        if args.basedtime and not args.type:
            hour = datetime.now().strftime("%H")

            hIN = self.config['in_hours']

            hOUT = self.config['out_hours']

            if hour in hIN:
                typ = 0
            elif hour in hOUT:
                typ = 1
            else:
                self.sendReport("No se ha fichado ni desfichado ya que estamos en horario laboral, fuerze el estado con el parametro --type <0 = Entrada, 1 = Salida>")
        else:
            typ = args.type

        data = {
            "cp": self.config['user'],
            "pin": args.pin,
            "typ": typ,
            "date": self.headers['tn-d'],
            "geoLatitude": self.config['geo']['latitude'],
            "geoLongitude": self.config['geo']['longitude'],
            "geoError": "",
            "c": 1,
            "geoAccuracy": self.config['geo']['accuracy']
        }

        response = None
        if not args.debug:
            self.sendReport("####HACIENDO CHECK####")
            response = self.post("check", data, self.headers)
        else:
            self.sendReport('Corriendo en modo debug. No se realizará ninguna acción')
            exit(20)

        try:
            rj = json.loads(response.text)
        except:
            self.sendReport("La respuesta al hacer check no es correcta... algo ha pasado :/")
            exit(20)

        if response.status_code == 200:
            if rj['Repeated']:
                time = datetime.strptime(rj['RepeatedTime'], "%Y-%m-%dT%H:%M:%SZ").replace(
                    tzinfo=pytz.utc).astimezone(pytz.timezone("Europe/Madrid")).strftime("%H:%M")
                self.sendReport("🕐 Ya se ha realizado este marcaje antes a las %s" % time)
            else:
                time = datetime.now().astimezone(pytz.timezone("Europe/Madrid")).strftime("%H:%M")

                conditional_response = [{
                    "text": "fichado",
                    "emoji": "🕐🏢🏃‍♂️🏡"
                },{
                    "text": "desfichado",
                    "emoji": "🏡🏃‍♂️🏢🕐"
                }]
                self.sendReport('%s Has %s correctamente a las %s' % (conditional_response[int(typ)]['emoji'], conditional_response[int(typ)]['text'], time))
        else:
            self.sendReport("🟥🕐 No se ha podido fichar, la web no ha devuelto 200 OK")

    def calendarCheck(self):
        creds = None
        configpath = os.path.join(ROOT_DIR, 'googleoauth.json')
        if os.path.exists(configpath):
            try:
                creds = Credentials.from_authorized_user_file(configpath, SCOPES)
            except ValueError as error:
                print(error)
        else:
            return False


        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(configpath, SCOPES)
                creds = flow.run_local_server(port=0)

            with open(configpath, 'w') as token:
                token.write(creds.to_json())

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

            # Call the Calendar API
            now = datetime.utcnow().strftime('%Y-%m-%dT00:00:00Z')  # 'Z' indicates UTC time

            now_after_one_day = (datetime.utcnow() + relativedelta(days=1)).strftime('%Y-%m-%dT00:00:00Z')
            events_result_autoficher = service.events().list(calendarId='e51d3bfb6b352c48afb8bd7a5d494599cc3eaa686800a856796306f50c6d72fd@group.calendar.google.com', timeMin=now,
                                                  timeMax=now_after_one_day, maxResults=10, singleEvents=True,
                                                  orderBy='startTime').execute()
            events_result_holiday = service.events().list(
                calendarId='es.spain#holiday@group.v.calendar.google.com',
                timeMin=now,
                timeMax=now_after_one_day,
                maxResults=10, singleEvents=True,
                orderBy='startTime').execute()


            events = events_result_autoficher.get('items', []) + events_result_holiday.get('items', [])

            if not events:
                return False

            # Prints the start and name of the next 10 events
            for event in events:
                start = event['start'].get('dateTime', event['start'].get('date'))
                end = event['end'].get('dateTime', event['end'].get('date'))

                if datetime.fromisoformat(start).timestamp() <= datetime.now().timestamp() <= datetime.fromisoformat(end).timestamp():
                    if event['organizer']['displayName'] == 'autoficher':
                        return 'Forzado desde autoficher - ' + event['summary']

                    if 'description' in event and ("Cataluña" in event['description'] or 'Día festivo' in event['description']or 'Celebración\n' in event['description']) and not 'Cambio de horario' in event['summary']:
                        return event['summary']
        except HttpError as error:
            print(f'An error occurred: {error}')
            return 'Error de llamada a api'

        return 'No se que ha pasado :/'

if __name__ == '__main__':
    af = AutoFicher()
    af.sendUpdate()