omar
f2ed694f98
Cinco graficos generados con matplotlib y enviados como foto a Telegram, con
botones para elegir. La paleta es la de referencia del proyecto, validada con el
comprobador de daltonismo en claro y oscuro (peor par CVD delta-E 24.7 / 26.8,
umbral 8).
- saldo acumulado (linea) + saldo de cada dia (barras), dos paneles
- puntualidad desviacion de cada entrada en el tiempo. Es el que explica el
problema de un vistazo: en 6 meses no hay ni un solo punto por
debajo de cero, o sea que nunca se ha llegado antes de la hora
- horas histograma de la hora de entrada, historico frente a planificado.
Las dos distribuciones no se solapan en absoluto
- regalado minutos al mes presente sin que corra el reloj
- prevision linea de tiempo de los proximos dias con la franja que cuenta
Un PNG no tiene hover, asi que las cifras que en una pagina resolveria el
tooltip van en el pie de foto y en etiquetas directas sobre las marcas.
Al mirarlos renderizados aparecieron cuatro defectos que el codigo no delata:
titulo y subtitulo pisandose, la etiqueta del peor dia chocando con el eje de
fechas, un dia de +64 min aplastando el resto del eje contra el cero, y un bug
de formato que pintaba "09:64" en vez de "10:04". Corregidos: ejes acotados con
los valores extremos anotados aparte, y las horas formateadas con un helper.
Executor.historico() extrae los dias de varios meses ya parseados; el servicio
lo cachea media hora porque son seis llamadas a la API.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
171 lines
6.8 KiB
Python
171 lines
6.8 KiB
Python
"""Ejecuta los fichajes contra Timenet y consulta el estado real.
|
|
|
|
Todo lo de este modulo es sincrono (usa ``requests``). El daemon lo invoca con
|
|
``asyncio.to_thread`` para no bloquear el bucle de eventos.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import calendar
|
|
import logging
|
|
import random
|
|
from datetime import date, datetime, timedelta
|
|
|
|
from app import jitter, schedule
|
|
from app.timenet_api import TimenetClient, TimenetError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ENTRADA, SALIDA = 0, 1
|
|
CALENDAR_TTL = timedelta(hours=12)
|
|
|
|
|
|
class Executor:
|
|
def __init__(self, cfg: dict, store, pin: str):
|
|
self.cfg = cfg
|
|
self.store = store
|
|
self.pin = pin
|
|
self.cli = TimenetClient(cfg["api_url"], cfg["user"])
|
|
self._cal: schedule.Calendar | None = None
|
|
self._cal_at: datetime | None = None
|
|
|
|
# ------------------------------------------------------------------ login
|
|
def ensure_login(self, force: bool = False) -> None:
|
|
if self.cli.token and not force:
|
|
return
|
|
self.cli.login(self.pin)
|
|
logger.info("login en Timenet OK")
|
|
|
|
def _with_login(self, fn, *a, **kw):
|
|
"""Ejecuta y reintenta una vez con login nuevo si el token ha caducado."""
|
|
self.ensure_login()
|
|
try:
|
|
return fn(*a, **kw)
|
|
except TimenetError as e:
|
|
logger.warning("fallo en %s (%s); reintentando con login nuevo", fn.__name__, e)
|
|
self.ensure_login(force=True)
|
|
return fn(*a, **kw)
|
|
|
|
# -------------------------------------------------------------- calendario
|
|
def calendar(self, force: bool = False) -> schedule.Calendar:
|
|
fresh = self._cal_at and datetime.now() - self._cal_at < CALENDAR_TTL
|
|
if self._cal and fresh and not force:
|
|
return self._cal
|
|
raw = self._with_login(self.cli.raw_calendar)
|
|
self._cal = schedule.Calendar(raw)
|
|
self._cal_at = datetime.now()
|
|
logger.info("calendario refrescado: %d dias, %d tipos de jornada",
|
|
len(self._cal.days), len(self._cal.day_types))
|
|
return self._cal
|
|
|
|
# ------------------------------------------------------------------ fichar
|
|
def do_check(self, typ: int) -> dict:
|
|
"""Ficha de verdad. Devuelve un resumen con lo que Timenet ha registrado."""
|
|
g = self.cfg.get("geo") or {}
|
|
lat, lon = jitter.jitter_geo(
|
|
g.get("latitude", 0), g.get("longitude", 0),
|
|
float(self.cfg.get("geo_jitter_meters") or 0), random.Random())
|
|
|
|
logger.info("fichando typ=%s en %.6f,%.6f", typ, lat, lon)
|
|
rj = self._with_login(self.cli.post_check, typ, lat, lon)
|
|
|
|
out = {"typ": typ, "lat": lat, "lon": lon, "raw": rj,
|
|
"repeated": bool(rj.get("Repeated")), "check_id": None,
|
|
"recorded": None, "real": None}
|
|
try:
|
|
out.update(self._confirm(typ))
|
|
except Exception as e: # la confirmacion nunca debe tumbar el fichaje
|
|
logger.warning("no se pudo confirmar el fichaje: %s", e)
|
|
return out
|
|
|
|
def _confirm(self, typ: int) -> dict:
|
|
"""Relee los fichajes de hoy y devuelve el ultimo, tal y como lo ve Timenet."""
|
|
today = self.today_raw()
|
|
dias = today.get("C") or []
|
|
if not dias:
|
|
return {}
|
|
checks = dias[-1].get("C") or []
|
|
if not checks:
|
|
return {}
|
|
last = checks[-1]
|
|
return {"check_id": last.get("I"),
|
|
"recorded": last.get("H"), # hora que CUENTA
|
|
"real": last.get("HR")} # hora realmente fichada
|
|
|
|
# ------------------------------------------------------------------ estado
|
|
def today_raw(self, day: date | None = None) -> dict:
|
|
d = (day or date.today()).strftime("%d/%m/%Y")
|
|
return self._with_login(self.cli.raw_checks, d, d)
|
|
|
|
def month_raw(self, day: date | None = None) -> dict:
|
|
d = day or date.today()
|
|
first = d.replace(day=1)
|
|
last = (first + timedelta(days=32)).replace(day=1) - timedelta(days=1)
|
|
return self._with_login(self.cli.raw_checks,
|
|
first.strftime("%d/%m/%Y"), last.strftime("%d/%m/%Y"))
|
|
|
|
def balance(self, day: date | None = None) -> dict:
|
|
m = self.month_raw(day)
|
|
x, xant = m.get("X") or 0, m.get("XAnt") or 0
|
|
return {"P": m.get("P") or 0, "W": m.get("W") or 0,
|
|
"X": x, "XAnt": xant, "total": x + xant}
|
|
|
|
def historico(self, meses: int = 6) -> list[dict]:
|
|
"""Un registro por dia laborable de los ultimos `meses`, ordenado.
|
|
|
|
Cada registro: fecha, previsto (P), contado (W), saldo (X), presencia real
|
|
(A, el tiempo entre fichajes sin recortar) y los fichajes con su hora real
|
|
(HR) y la que cuenta (H).
|
|
"""
|
|
hoy = date.today()
|
|
y, m = hoy.year, hoy.month
|
|
periodos = []
|
|
for _ in range(meses):
|
|
periodos.append((y, m))
|
|
m -= 1
|
|
if m == 0:
|
|
m, y = 12, y - 1
|
|
|
|
out: list[dict] = []
|
|
for y, m in reversed(periodos):
|
|
ultimo = calendar.monthrange(y, m)[1]
|
|
try:
|
|
raw = self._with_login(self.cli.raw_checks,
|
|
"01/%02d/%d" % (m, y), "%02d/%02d/%d" % (ultimo, m, y))
|
|
except TimenetError as e:
|
|
logger.warning("no se pudo leer %d-%02d: %s", y, m, e)
|
|
continue
|
|
for dia in raw.get("C") or []:
|
|
cs = dia.get("C") or []
|
|
if not cs or not (dia.get("P") or 0):
|
|
continue
|
|
try:
|
|
f = datetime.strptime(dia["D"][:10], "%d/%m/%Y").date()
|
|
except (KeyError, ValueError):
|
|
continue
|
|
out.append({
|
|
"fecha": f,
|
|
"P": dia.get("P") or 0,
|
|
"W": dia.get("W") or 0,
|
|
"X": dia.get("X") or 0,
|
|
"A": (dia.get("A") or [{}])[0].get("M") or 0,
|
|
"tipo": (dia.get("T") or "").strip(),
|
|
"checks": [{"typ": c.get("T"), "H": c.get("H"), "HR": c.get("HR")}
|
|
for c in cs],
|
|
})
|
|
out.sort(key=lambda r: r["fecha"])
|
|
return out
|
|
|
|
def today_summary(self, day: date | None = None) -> dict:
|
|
"""Fichajes de hoy segun Timenet, con hora real y hora contada."""
|
|
raw = self.today_raw(day)
|
|
dias = raw.get("C") or []
|
|
if not dias:
|
|
return {"checks": [], "P": 0, "W": 0, "X": 0, "tipo": None}
|
|
d0 = dias[-1]
|
|
return {
|
|
"checks": [{"typ": c.get("T"), "recorded": c.get("H"), "real": c.get("HR"),
|
|
"id": c.get("I")} for c in (d0.get("C") or [])],
|
|
"P": d0.get("P") or 0, "W": d0.get("W") or 0, "X": d0.get("X") or 0,
|
|
"tipo": (d0.get("T") or "").strip() or None,
|
|
}
|