autoficher/app/timenet_api.py

"""Cliente HTTP de la API de Timenet.

Notas sobre la API (descubiertas el 2026-09-24 contra el servidor 9.1.0):

* Las cabeceras ``Tn-D`` y ``dStr`` (timestamp UTC de la peticion) son OBLIGATORIAS:
  sin ellas el servidor responde 403 incluso en endpoints publicos como ``version``.
  Hay que refrescarlas en cada peticion.
* Todos los endpoints cuelgan de ``v1/`` menos ``version``.
* El login devuelve el token en el cuerpo, entrecomillado.
"""
from __future__ import annotations

import logging
from datetime import datetime, timezone
from typing import Any

import requests

from app import strings

logger = logging.getLogger(__name__)

WCP_VERSION = "wcp_8.0.0.2"
TIMEOUT = 30


class TimenetError(RuntimeError):
    """Fallo de la API de Timenet."""


class TimenetClient:
    def __init__(self, api_url: str, user_guid: str, wcp_version: str = WCP_VERSION):
        self.api_url = api_url if api_url.endswith("/") else api_url + "/"
        self.user_guid = user_guid
        self.wcp_version = wcp_version
        self.session = requests.Session()
        self.token: str | None = None

    # ---------------------------------------------------------------- headers
    def _headers(self) -> dict[str, str]:
        now = datetime.now(timezone.utc)
        h = {
            "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",
            "User-Agent": strings.user_agent,
            "Tn-U": strings.tn_u,
            strings.tnv_string: self.wcp_version,
            # obligatorias: sin ellas -> 403
            strings.tnd_string: '"' + now.strftime("%Y-%m-%dT%H:%M:%SZ") + '"',
            "dStr": '"' + now.strftime("%d/%m/%Y %H:%M:%S") + '"',
        }
        if self.token:
            h["Token"] = self.token
        return h

    # ------------------------------------------------------------------ verbs
    def _request(self, method: str, path: str, data: dict | None = None) -> requests.Response:
        url = self.api_url + ("" if path == "version" else "v1/") + path
        try:
            r = self.session.request(method, url, data=data or {},
                                     headers=self._headers(), timeout=TIMEOUT)
        except requests.exceptions.RequestException as e:
            raise TimenetError(f"{method} {path}: {type(e).__name__}: {e}") from e
        logger.debug("%s %s -> %s", method, path, r.status_code)
        return r

    def _json(self, path: str) -> Any:
        r = self._request("GET", path)
        if r.status_code != 200:
            raise TimenetError(f"GET {path}: HTTP {r.status_code} {r.text[:200]!r}")
        try:
            return r.json()
        except ValueError as e:
            raise TimenetError(f"GET {path}: respuesta no es JSON: {r.text[:200]!r}") from e

    # -------------------------------------------------------------- endpoints
    def version(self) -> str:
        r = self._request("GET", "version")
        if r.status_code != 200:
            raise TimenetError(f"version: HTTP {r.status_code}")
        return r.text.strip().strip('"')

    def login(self, pin: int | str) -> str:
        self.token = None
        r = self._request("POST", f"cp/login/{self.user_guid}", {"p": pin})
        if r.status_code != 200:
            raise TimenetError(f"login rechazado: HTTP {r.status_code} {r.text[:200]!r}")
        self.token = r.text.replace('"', "").strip()
        if not self.token:
            raise TimenetError("login devolvio un token vacio")
        return self.token

    def info(self) -> dict:
        return self._json(f"check/info?guid={self.user_guid}")

    def config(self) -> dict:
        return self._json("cp/config")

    def dashboard(self) -> dict:
        return self._json("cp/dashboard")

    def raw_calendar(self, start: str | None = None, end: str | None = None) -> dict:
        """cp/calendar: tipos de jornada + asignacion por dia. Sin fechas -> año en curso."""
        q = f"?start={start}&end={end}" if start and end else ""
        return self._json("cp/calendar" + q)

    def raw_checks(self, start: str, end: str) -> dict:
        """cp/checks con fechas dd/mm/yyyy. Devuelve dias, fichajes y saldos."""
        return self._json(f"cp/checks?start={start}&end={end}")

    # ----------------------------------------------------------------- fichar
    def post_check(self, typ: int, latitude, longitude, when: datetime | None = None) -> dict:
        """Registra un fichaje. typ: 0 = Entrada, 1 = Salida.

        El unico metodo de esta clase que escribe. Devuelve el JSON de respuesta.
        """
        now = (when or datetime.now(timezone.utc)).astimezone(timezone.utc)
        data = {
            "typ": typ,
            "date": '"' + now.strftime("%Y-%m-%dT%H:%M:%SZ") + '"',
            "geoLatitude": latitude,
            "geoLongitude": longitude,
            "geoError": "",
            "dStr": now.strftime("%d/%m/%Y %H:%M:%S"),
        }
        r = self._request("POST", "cp/checks", data)
        if r.text.strip() == "no valid worker":
            raise TimenetError("el trabajador no ha podido ser identificado")
        if r.status_code != 200:
            raise TimenetError(f"fichaje rechazado: HTTP {r.status_code} {r.text[:200]!r}")
        try:
            return r.json()
        except ValueError as e:
            raise TimenetError(f"respuesta del fichaje no es JSON: {r.text[:200]!r}") from e