autoficher/app/timenet_manager.py

import asyncio

import requests, json, pytz
from datetime import datetime
from time import sleep
import random
from app import arg_parser, config_parser, telegram_bot, google_calendar, utils

args = arg_parser.ArgParser().parse()
config = config_parser.ConfigParser().loadConfig()

tnv_string = "Tn-V"
tnd_string = "Tn-D"


conditional_response = [{
    "text": "fichado",
    "text2": "fichar",
    "emoji": "🕐🏢🏃‍♂️🏡"
},{
    "text": "desfichado",
    "text2": "desfichar",
    "emoji": "🏡🏃‍♂️🏢🕐"
}]

class timenetManager:
    telegram = None
    message_acumulator = ""
    headers = {}

    def __init__(self):
        self.telegram = telegram_bot.telegramBot()

        if args.pin is None:
            # if pin is None, prompt for it
            args.pin = int(input("Introduzca su pin: "))

        self.headers = {
            "Content-type": "application/x-www-form-urlencoded; charset=UTF-8",
            "Accept": "application/json, text/plain, */*",
            "Referer": "https://timenet-wcp.gpisoftware.com/",
            "Sec-Ch-Ua-Mobile": "?0",
            "Sec-Ch-Ua-Platform": "\"Linux\"",
            "Sec-Ch-Ua": "\"Chromium\";v=\"138\", \"Not=A?Brand\";v=\"8\", \"Google Chrome\";v=\"138\"",
            "Sec-Fetch-Dest": "empty",
            "Sec-Fetch-Mode": "cors",
            "Sec-Fetch-Site": "same-origin",
            tnv_string: "wcp_8.0.0.2",
            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36",
            "Tn-U": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36"
        }
        self.login()


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

        print("[GET] %s con headers %s y data %s" % (furl, headers, data))
        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

        print("[POST] %s con headers %s y data %s" % (furl, headers, data))
        return requests.post(furl, data=data, headers=headers)
    
    def login(self):
        self.updateTime()

        response = self.post("cp/login/" + config['user'], {
            "p": args.pin,
        }, self.headers)
        
        if response.status_code != 200:
            print("Error al iniciar sesión: %s" % response.text)
            exit(20)

        self.headers["Token"] = response.text.replace("\"", "")

    def addMessage(self, message):
        self.message_acumulator += message + "\n"

    def sendReport(self):
        # self.telegram.sendMessage(self.message_acumulator)
        asyncio.run(self.telegram.sendMessage(self.message_acumulator))
        self.message_acumulator = ""

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

    def updateVersion(self):
        self.headers[tnv_string] = self.getVersion()

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

    def getInfo(self):
        return self.get('check/info?guid=' + config['user'], {}, self.headers)
    
    def checkLastCheck(self, typ):
        response =  self.get('cp/checks?start=' + datetime.now().strftime("%d/%m/%Y") + '&end=' + datetime.now().strftime("%d/%m/%Y"), {}, self.headers)
        if response.status_code != 200:
            self.addMessage("Error al obtener los checks: %s" % response.text)
            self.sendReport()
            return False
        
        
        today_checks = json.loads(response.text)
        if today_checks['C'] and today_checks['C'][0]['C'][-1]['T'] == typ:
            self.addMessage("Ya se ha realizado este marcaje antes a las %s" % today_checks['C'][0]['C'][-1]['H'])
            self.sendReport()
            return True
        

        return False


    def sendUpdate(self):

        calendar = google_calendar.GoogleCalendar()
        calendar = calendar.getEvent()
        if calendar and not args.force:
            self.addMessage("🟥🕐 Comprobación de calendario: " + calendar)
            self.sendReport()
            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.addMessage("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


        if self.checkLastCheck(typ):
            return;

        response = None
        if not args.debug:
            self.addMessage("####HACIENDO CHECK####")
            rand = utils.numero_aleatorio_con_probabilidad(230, 0.5)
            self.addMessage("Esperando %s segundos para %s en un minuto aleatorio" % (rand,conditional_response[int(typ)]["text2"]))
            self.sendReport()
            sleep(rand)
            self.updateTime()
            data = {
                "typ": typ,
                "date": self.headers[tnd_string],
                "geoLatitude": config['geo']['latitude'],
                "geoLongitude": config['geo']['longitude'],
                "geoError": "",
                "dStr": self.headers["dStr"].replace("\"", "")
            }
            response = self.post("cp/checks", data, self.headers)
        else:
            self.addMessage('Corriendo en modo debug. No se realizará ninguna acción')
            self.sendReport()
            exit(20)

        if response.text == "no valid worker":
            self.addMessage("El trabajador no ha podido ser identificado")
            self.sendReport()
            exit(20)

        try:
            rj = json.loads(response.text)
        except:
            self.addMessage("La respuesta al hacer check no es correcta... algo ha pasado :/ - %s - %s" % (response.text, response.status_code))
            self.sendReport()
            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.addMessage("🕐 Ya se ha realizado este marcaje antes a las %s" % time)
            else:
                time = datetime.now().astimezone(pytz.timezone("Europe/Madrid")).strftime("%H:%M")
                self.addMessage('%s Has %s correctamente a las %s' % (conditional_response[int(typ)]['emoji'], conditional_response[int(typ)]['text'], time))
        else:
            self.addMessage("🟥🕐 No se ha podido fichar, la web no ha devuelto 200 OK")

        self.sendReport()