#!/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 args.type is None and args.basedtime is False: 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 updateVersion(self): self.headers["tn-v"] = self.getVersion() 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() self.updateVersion() 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) if response.text == "no valid worker": self.sendReport("El trabajador no ha podido ser identificado") 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.get("summary", "Sin Título") 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 False if __name__ == '__main__': af = AutoFicher() af.sendUpdate()