omar
9294474066
Timenet recorta el tiempo a la ventana teorica del horario:
trabajado = min(salida, fin_teorico) - max(entrada, inicio_teorico)
Verificado sobre 6 meses (104 dias laborables, 376 fichajes). La API guarda dos
horas por fichaje: HR (la que fichas) y H (la que cuenta). De 376 fichajes, 136
llevaban ajuste: todos salidas, todos recortados hacia atras. Ninguna entrada
ajustada jamas. Resultado: una entrada tardia se debita al minuto y una salida
tardia no acredita nada, asi que el jitter aleatorio positivo que habia perdia
unos 3,2 min/dia (-776 min acumulados, casi 13 horas).
Y acabar el mes en positivo es imposible: 0 dias con saldo positivo en 108, y
timeExtraHours=false en todos los tipos de jornada. El objetivo es 0 clavado.
- timenet_api: cliente limpio. Documenta que Tn-D y dStr son obligatorias (sin
ellas el servidor responde 403 hasta en endpoints publicos).
- schedule: calendario laboral desde cp/calendar. Festivos, vacaciones y tramos
horarios los da Timenet, no hace falta adivinarlos.
- planner + jitter: entradas siempre ANTES de la hora, salidas siempre DESPUES,
con costumbre que deriva lenta y cola larga hacia el lado seguro. Invariantes
que fallan en duro si una entrada se planifica tarde.
- jitter_geo: las coordenadas del historico son identicas al sexto decimal en
los 376 fichajes; ahora se desplazan dentro de un disco.
- store: SQLite con plan diario, overrides, auditoria y estado del drift.
- settings: config sin el acoplamiento a argparse del config_parser original.
tools/dryrun.py simula el año entero: saldo +0 en 226 dias laborables, con
~3,5 min/dia de presencia extra.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
118 lines
4.6 KiB
Python
118 lines
4.6 KiB
Python
"""Construye el plan de fichajes de un dia a partir del calendario de Timenet."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import random
|
|
from dataclasses import dataclass
|
|
from datetime import date, datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from app import jitter
|
|
from app.schedule import Calendar
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ENTRADA, SALIDA = 0, 1
|
|
|
|
|
|
@dataclass
|
|
class PlannedCheck:
|
|
idx: int
|
|
typ: int # 0 entrada, 1 salida
|
|
window_idx: int
|
|
target: datetime # hora teorica
|
|
planned_at: datetime # instante real en que se fichara
|
|
|
|
@property
|
|
def label(self) -> str:
|
|
return "ENTRADA" if self.typ == ENTRADA else "SALIDA"
|
|
|
|
@property
|
|
def offset_seconds(self) -> int:
|
|
"""Segundos de margen respecto de la teorica. Positivo = lado seguro."""
|
|
d = (self.planned_at - self.target).total_seconds()
|
|
return int(-d if self.typ == ENTRADA else d)
|
|
|
|
def __str__(self) -> str:
|
|
return "%-7s teorica %s ficha %s (%+ds -> %s)" % (
|
|
self.label, self.target.strftime("%H:%M:%S"),
|
|
self.planned_at.strftime("%H:%M:%S"),
|
|
int((self.planned_at - self.target).total_seconds()),
|
|
"antes" if self.typ == ENTRADA else "despues")
|
|
|
|
|
|
class Planner:
|
|
def __init__(self, cfg: dict, store=None):
|
|
self.cfg = cfg
|
|
self.store = store
|
|
self.tz = ZoneInfo(cfg.get("timezone", "Europe/Madrid"))
|
|
self.j = cfg["jitter"]
|
|
|
|
# ------------------------------------------------------------------ drift
|
|
def _drift_key(self, window_idx: int, typ: int) -> str:
|
|
return f"drift.{window_idx}.{typ}"
|
|
|
|
def _drift(self, window_idx: int, typ: int, rng: random.Random) -> float:
|
|
key = self._drift_key(window_idx, typ)
|
|
prev = float(self.store.get(key, 0.0)) if self.store else 0.0
|
|
nxt = jitter.next_drift(prev, self.j["drift_sigma"], self.j["drift_clamp"], rng)
|
|
if self.store:
|
|
self.store.put(key, nxt)
|
|
return nxt
|
|
|
|
# ------------------------------------------------------------------- plan
|
|
def plan_day(self, cal: Calendar, day: date,
|
|
rng: random.Random | None = None) -> list[PlannedCheck]:
|
|
"""Plan del dia. Lista vacia si no toca fichar."""
|
|
if not cal.is_workday(day):
|
|
logger.info("%s no es laborable: %s", day, cal.why_not_workday(day))
|
|
return []
|
|
|
|
rng = rng or random.Random()
|
|
out: list[PlannedCheck] = []
|
|
idx = 0
|
|
|
|
for wi, w in enumerate(cal.windows(day)):
|
|
start = datetime.combine(day, datetime.min.time(), self.tz) + timedelta(minutes=w.start)
|
|
end = datetime.combine(day, datetime.min.time(), self.tz) + timedelta(minutes=w.end)
|
|
|
|
# ENTRADA: siempre antes de la hora teorica
|
|
off = jitter.offset_seconds(
|
|
self.j["entry_before_min"], self.j["entry_before_max"],
|
|
self._drift(wi, ENTRADA, rng), rng,
|
|
# el margen de entrada del propio Timenet acota cuanto se puede adelantar
|
|
outlier_cap=min(self.j["entry_before_max"] * 3, max(w.entry_margin * 60 // 2, 60)),
|
|
)
|
|
off = max(off, self.j["entry_guard"])
|
|
out.append(PlannedCheck(idx, ENTRADA, wi, start, start - timedelta(seconds=off)))
|
|
idx += 1
|
|
|
|
# SALIDA: siempre despues
|
|
off = jitter.offset_seconds(
|
|
self.j["exit_after_min"], self.j["exit_after_max"],
|
|
self._drift(wi, SALIDA, rng), rng,
|
|
outlier_cap=self.j["exit_after_max"] * 4,
|
|
)
|
|
out.append(PlannedCheck(idx, SALIDA, wi, end, end + timedelta(seconds=off)))
|
|
idx += 1
|
|
|
|
self._validate(out, cal, day)
|
|
return out
|
|
|
|
# --------------------------------------------------------------- garantias
|
|
def _validate(self, plan: list[PlannedCheck], cal: Calendar, day: date) -> None:
|
|
"""Invariantes que, si se rompen, cuestan minutos de verdad."""
|
|
for p in plan:
|
|
if p.typ == ENTRADA and p.planned_at >= p.target:
|
|
raise AssertionError(f"entrada planificada tarde: {p}")
|
|
if p.typ == SALIDA and p.planned_at <= p.target:
|
|
raise AssertionError(f"salida planificada pronto: {p}")
|
|
for a, b in zip(plan, plan[1:]):
|
|
if b.planned_at <= a.planned_at:
|
|
raise AssertionError(f"fichajes desordenados: {a} / {b}")
|
|
|
|
def rows_for_store(self, plan: list[PlannedCheck]) -> list[dict]:
|
|
return [{"idx": p.idx, "typ": p.typ, "window_idx": p.window_idx,
|
|
"target": p.target.isoformat(), "planned_at": p.planned_at.isoformat()}
|
|
for p in plan]
|