separación de codigo para facil edición

Signed-off-by: Omar Sánchez Pizarro <omar.sanchez@pistacero.net>

app/arg_parser.py

@@ -0,0 +1,36 @@
+import argparse
+
+
+class ArgParser:
+    parser = None
+    args = None
+    def __init__(self):
+        self.parser = argparse.ArgumentParser(add_help=False)
+        self.parser._action_groups.pop()
+
+        obligatoryArgs = self.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 = self.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)
+
+    def check(self):
+        if self.args.type is None and self.args.basedtime is False:
+            print("Hay que indicar al menos un metodo de fichaje -t <0 o 1> o -bt")
+            exit(20)
+
+    def parse(self):
+        self.args = self.parser.parse_args()
+        self.check()
+
+        return self.args

app/config_parser.py

@@ -0,0 +1,32 @@
+import json, os
+from app import arg_parser
+
+args = arg_parser.ArgParser().parse()
+
+
+class ConfigParser:
+    config = None
+
+    def __init__(self):
+        self.loadConfig()
+
+    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
+
+        return self.config

app/google_calendar.py

@@ -0,0 +1,104 @@
+import google.oauth2.credentials
+
+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
+
+from datetime import datetime
+from dateutil.relativedelta import relativedelta
+
+import os
+
+
+class GoogleCalendar:
+    def __init__(self):
+        self.creds = None
+        self.service = None
+        self.SCOPES = ['https://www.googleapis.com/auth/calendar.readonly']
+        self.authfile = os.path.join(os.path.dirname(__file__), '../googleoauth.json')
+        self.grantfile = os.path.join(os.path.dirname(__file__), '../googleoauth_grant.json')
+
+        if os.path.exists(self.authfile):
+            try:
+                self.creds = Credentials.from_authorized_user_file(self.authfile, self.SCOPES)
+            except ValueError as error:
+                print(error)
+
+        self.checkAuth()
+        self.buildService()
+
+    def serverAuth(self):
+        flow = InstalledAppFlow.from_client_secrets_file(self.grantfile, self.SCOPES)
+        self.creds = flow.run_local_server(port=0)
+
+    def saveAuth(self):
+        with open(self.authfile, 'w') as token:
+            token.write(self.creds.to_json())
+
+    def refreshAuth(self):
+        self.creds.refresh(Request())
+
+    def checkAuth(self):
+        if not self.creds or not self.creds.valid:
+            if self.creds and self.creds.expired and self.creds.refresh_token:
+                try:
+                    self.refreshAuth()
+                except google.oauth2.credentials.exceptions.RefreshError as rerror:
+                    self.serverAuth()
+            else:
+                self.serverAuth()
+
+            self.saveAuth()
+
+    def buildService(self):
+        self.service = build('calendar', 'v3', credentials=self.creds)
+
+    def getNow(self):
+        return datetime.utcnow().strftime('%Y-%m-%dT00:00:00Z')
+
+    def getNowAfterOneDay(self):
+        return (datetime.utcnow() + relativedelta(days=1)).strftime('%Y-%m-%dT00:00:00Z')
+
+    def getEvent(self):
+        try:
+            now = self.getNow()
+            now_after_one_day = self.getNowAfterOneDay()
+
+            events_result_autoficher = self.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 = self.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 llama a Google Calendar'
+
+        return False

app/telegram_bot.py

@@ -0,0 +1,15 @@
+import telegram
+from app import config_parser
+
+config = config_parser.ConfigParser().loadConfig()
+
+
+class telegramBot:
+    bot = None
+
+    def __init__(self):
+        self.bot = telegram.Bot(token=config['telegram']['token'])
+
+    def sendMessage(self, message):
+        print(message)
+        self.bot.sendMessage(chat_id=config['telegram']['chat_id'], text=str(message))

app/timenet_manager.py

@@ -0,0 +1,124 @@
+import requests, json, pytz
+from datetime import datetime
+from app import arg_parser, config_parser, telegram_bot, google_calendar
+
+args = arg_parser.ArgParser().parse()
+config = config_parser.ConfigParser().loadConfig()
+
+
+class timenetManager:
+    telegram = None
+
+    def __init__(self):
+        self.telegram = telegram_bot.telegramBot()
+
+    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"
+    }
+
+    def get(self, url, data={}, headers={}):
+        furl = config['api_url'] + url
+        if url != "version":
+            furl = config['api_url'] + "v1/" + url
+        return requests.get(furl, data=data, headers=headers)
+
+    def post(self, url, data={}, headers={}):
+        furl = config['api_url'] + url
+        if url != "version":
+            furl = config['api_url'] + "v1/" + url
+        return requests.post(furl, data=data, headers=headers)
+
+    def sendReport(self, message):
+        self.telegram.sendMessage(message)
+
+    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()
+        print(self.headers)
+        exit(20)
+
+    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 = google_calendar.GoogleCalendar()
+        calendar = calendar.getEvent()
+        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 = config['in_hours']
+
+            hOUT = 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": config['user'],
+            "pin": args.pin,
+            "typ": typ,
+            "date": self.headers['tn-d'],
+            "geoLatitude": config['geo']['latitude'],
+            "geoLongitude": config['geo']['longitude'],
+            "geoError": "",
+            "c": 1,
+            "geoAccuracy": 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")

autoficher.py

@@ -1,266 +0,0 @@
-#!/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()

googleoauth_grant.json

@@ -0,0 +1 @@
+{"installed":{"client_id":"962199633636-qu0te9fbj06fe4p44gskbb1p6f2s0d5s.apps.googleusercontent.com","project_id":"autoficher","auth_uri":"https://accounts.google.com/o/oauth2/auth","token_uri":"https://oauth2.googleapis.com/token","auth_provider_x509_cert_url":"https://www.googleapis.com/oauth2/v1/certs","client_secret":"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","redirect_uris":["http://localhost"]}}

main.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+# encoding: utf-8
+#
+# Automarcaje para timenet.gpisoftware.com
+# usage: main.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 App Principal
+from app import timenet_manager
+
+
+if __name__ == '__main__':
+    af = timenet_manager.timenetManager()
+    af.sendUpdate()