public/autoficher
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: nucleo nuevo: cliente API, calendario laboral, planificador y jitter

Browse Source 
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>
This commit is contained in:
omar
2026-09-24 18:51:33 +02:00
parent 51976fbe22
commit 9294474066
10 changed files with 1123 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
app/executor.py Normal file
View File

@@ -0,0 +1,123 @@
"""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 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 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,
}

68
app/jitter.py Normal file
View File

@@ -0,0 +1,68 @@
"""Aleatoriedad que imita comportamiento humano, en tiempo y en posicion.
Reglas del juego (ver app/schedule.py y el analisis del historico):
* Timenet solo cuenta el tiempo DENTRO de la ventana teorica. Una entrada tardia se
debita al minuto; una salida tardia no acredita nada. Por tanto las ENTRADAS se
fichan siempre ANTES de la hora y las SALIDAS siempre DESPUES. Nunca al reves.
* Para que no parezca un dado nuevo cada dia, cada hueco del horario tiene su propia
"costumbre" que deriva lentamente (proceso de reversion a la media) y sobre la que
se suma el ruido del dia.
"""
from __future__ import annotations
import math
import random
# reversion a la media de la costumbre diaria: 0 = sin memoria, 1 = no se mueve
DRIFT_MEMORY = 0.82
# probabilidad de un dia atipico (te despistas y llegas bastante antes / te quedas mas)
OUTLIER_P = 0.06
def next_drift(previous: float, sigma: float, clamp: float, rng: random.Random) -> float:
"""Siguiente valor de la costumbre, en segundos. Deriva lenta y acotada."""
nxt = previous * DRIFT_MEMORY + rng.gauss(0.0, sigma)
return max(-clamp, min(clamp, nxt))
def offset_seconds(lo: int, hi: int, drift: float, rng: random.Random,
outlier_cap: int | None = None) -> int:
"""Desplazamiento en segundos respecto de la hora teorica, siempre en el lado seguro.
Devuelve un numero POSITIVO: cuanto antes se ficha una entrada, o cuanto despues
se ficha una salida. La cola larga va hacia el lado seguro (mas margen), nunca
hacia el lado que cuesta dinero.
"""
mid = (lo + hi) / 2.0
spread = max(1.0, (hi - lo) / 4.0)
val = mid + drift + rng.gauss(0.0, spread)
if rng.random() < OUTLIER_P:
cap = outlier_cap if outlier_cap is not None else hi * 3
val += abs(rng.gauss(0.0, (cap - hi) / 2.5)) if cap > hi else 0.0
val = min(val, cap)
return int(round(max(lo, val)))
def jitter_geo(latitude: float, longitude: float, meters: float,
rng: random.Random | None = None) -> tuple[float, float]:
"""Desplaza la posicion dentro de un disco de radio `meters`.
Las coordenadas del historico son identicas al sexto decimal en los 376 fichajes
de 6 meses. Un GPS real nunca hace eso.
"""
rng = rng or random
if not meters:
return round(float(latitude), 6), round(float(longitude), 6)
bearing = rng.uniform(0, 2 * math.pi)
# raiz cuadrada -> reparto uniforme sobre el area del disco, no concentrado al centro
dist = meters * math.sqrt(rng.random())
dlat = (dist * math.cos(bearing)) / 111_320.0
coslat = math.cos(math.radians(float(latitude))) or 1e-9
dlon = (dist * math.sin(bearing)) / (111_320.0 * coslat)
return round(float(latitude) + dlat, 6), round(float(longitude) + dlon, 6)

117
app/planner.py Normal file
View File

@@ -0,0 +1,117 @@
"""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]

149
app/schedule.py Normal file
View File

@@ -0,0 +1,149 @@
"""Modelo de horarios de Timenet, construido desde ``cp/calendar``.
``cp/calendar`` devuelve dos cosas que juntas describen el calendario laboral entero:
* ``DayTypes`` -- catalogo de tipos de jornada. Cada uno lleva su ``schedule``
(tramos entrada/salida), sus minutos previstos y si permite fichar.
* ``Calendar`` -- que tipo de jornada toca cada dia del año.
Con esto no hace falta adivinar festivos ni vacaciones: los da el propio Timenet.
Google Calendar queda solo para ausencias personales puntuales.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from datetime import date, datetime
logger = logging.getLogger(__name__)
# dayMode observados
MODE_WITH_SCHEDULE = "WITH_SCHEDULE"
MODE_NO_WORK = "NO_WORK"
MODE_FREE_HOURS = "FREE_HOURS"
MODE_FLEXIBLE = "FLEXIBLE_SCHEDULE"
@dataclass(frozen=True)
class Window:
"""Un tramo de trabajo, en minutos desde medianoche."""
start: int
end: int
entry_margin: int = 0 # minutos antes de `start` en que ya se admite el fichaje
exit_margin: int = 0
@property
def minutes(self) -> int:
return self.end - self.start
def hhmm(self, m: int) -> str:
return "%02d:%02d" % (m // 60, m % 60)
def __str__(self) -> str:
return f"{self.hhmm(self.start)}-{self.hhmm(self.end)}"
@dataclass
class DayType:
id: int
name: str
mode: str
minutes: int
can_check: bool
windows: list[Window] = field(default_factory=list)
colour: str | None = None
@property
def is_workday(self) -> bool:
return self.mode == MODE_WITH_SCHEDULE and self.minutes > 0 and bool(self.windows)
@property
def consistent(self) -> bool:
"""Los tramos declarados suman los minutos previstos."""
return sum(w.minutes for w in self.windows) == self.minutes
def __str__(self) -> str:
w = ", ".join(str(x) for x in self.windows) or "-"
return f"[{self.id}] {self.name.strip()!r} {self.mode} {self.minutes}min ({w})"
def _windows_from_schedule(sched: list[dict]) -> list[Window]:
out: list[Window] = []
for s in sched or []:
for suf in ("", "2"):
eh, em = s.get(f"EntryHour{suf}"), s.get(f"EntryMinute{suf}")
xh, xm = s.get(f"ExitHour{suf}"), s.get(f"ExitMinute{suf}")
if eh is None or xh is None:
continue
out.append(Window(
start=int(eh) * 60 + int(em or 0),
end=int(xh) * 60 + int(xm or 0),
entry_margin=int(s.get("EntryMargin") or 0),
exit_margin=int(s.get("ExitMargin") or 0),
))
return sorted(out, key=lambda w: w.start)
def _parse_day(s: str) -> date | None:
for fmt in ("%d/%m/%Y", "%Y-%m-%d"):
try:
return datetime.strptime(s[:10], fmt).date()
except (ValueError, TypeError):
continue
return None
class Calendar:
"""Calendario laboral resuelto: dia -> tipo de jornada -> tramos."""
def __init__(self, raw: dict):
self.day_types: dict[int, DayType] = {}
for d in raw.get("DayTypes") or []:
dt = DayType(
id=int(d["idDayType"]),
name=(d.get("name") or "").strip(),
mode=d.get("dayMode") or "",
minutes=int(d.get("minutes") or 0),
can_check=bool(d.get("canCheck")),
windows=_windows_from_schedule(d.get("schedule")),
colour=d.get("colour"),
)
if dt.windows and not dt.consistent:
logger.warning("Tipo de jornada incoherente: %s (tramos suman %d, previstos %d)",
dt, sum(w.minutes for w in dt.windows), dt.minutes)
self.day_types[dt.id] = dt
self.days: dict[date, int] = {}
for bloque in raw.get("Calendar") or []:
for item in bloque or []:
d = _parse_day(item.get("D", ""))
if d is not None:
self.days[d] = int(item.get("T"))
# ------------------------------------------------------------------ query
def day_type(self, d: date) -> DayType | None:
tid = self.days.get(d)
return self.day_types.get(tid) if tid is not None else None
def windows(self, d: date) -> list[Window]:
dt = self.day_type(d)
return list(dt.windows) if dt else []
def is_workday(self, d: date) -> bool:
dt = self.day_type(d)
return bool(dt and dt.is_workday and dt.can_check)
def why_not_workday(self, d: date) -> str | None:
"""Motivo legible por el que hoy no toca fichar, o None si si toca."""
dt = self.day_type(d)
if dt is None:
return f"sin tipo de jornada asignado para {d:%d/%m/%Y}"
if not dt.is_workday:
return f"{dt.name or dt.mode}"
if not dt.can_check:
return f"{dt.name}: el tipo de jornada no permite fichar"
return None
@property
def span(self) -> tuple[date, date] | None:
return (min(self.days), max(self.days)) if self.days else None

66
app/settings.py Normal file
View File

@@ -0,0 +1,66 @@
"""Carga de config.json sin depender de argparse.
El ``config_parser`` original invoca ``ArgParser().parse()`` en tiempo de import, lo
que obliga a que haya argumentos de linea de comandos validos. El daemon y las
herramientas necesitan leer la configuracion sin eso.
"""
from __future__ import annotations
import json
import os
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DEFAULT_CONFIG = os.path.join(ROOT, "config.json")
DEFAULTS = {
"api_url": "https://timenet-wcp.gpisoftware.com/api/",
"timezone": "Europe/Madrid",
"jitter": {
# margen, en segundos, con el que se ficha respecto de la hora teorica.
# Entradas ANTES (Timenet no acredita nada fuera de la ventana, pero si
# debita las entradas tardias). Salidas DESPUES.
# perfil "equilibrado": saldo 0 con ~3,6 min/dia de presencia extra.
# Mas holgado (40/150/30/130) = mas seguro pero ~5,7 min/dia.
# Mas ajustado (20/70/15/45) = ~2,7 min/dia pero solo 15s de colchon.
"entry_before_min": 30,
"entry_before_max": 100,
"exit_after_min": 20,
"exit_after_max": 70,
# deriva lenta de la hora habitual, para no parecer un dado nuevo cada dia
"drift_sigma": 25,
"drift_clamp": 60,
# margen de seguridad: nunca fichar una entrada a menos de X s de la hora
"entry_guard": 20,
},
"geo_jitter_meters": 12.0,
"telegram": {},
}
def _merge(base: dict, over: dict) -> dict:
out = dict(base)
for k, v in (over or {}).items():
out[k] = _merge(base[k], v) if isinstance(v, dict) and isinstance(base.get(k), dict) else v
return out
def load(path: str | None = None) -> dict:
path = path or os.environ.get("AUTOFICHER_CONFIG") or DEFAULT_CONFIG
if not os.path.exists(path):
raise FileNotFoundError(f"No se encuentra el archivo de configuracion: {path}")
with open(path, encoding="utf-8") as f:
cfg = json.load(f)
if not isinstance(cfg, dict):
raise ValueError("config.json no contiene un objeto JSON")
cfg = _merge(DEFAULTS, cfg)
for req in ("api_url", "user"):
if not cfg.get(req):
raise ValueError(f"Falta '{req}' en {path}")
return cfg
def pin(cfg: dict | None = None) -> str:
"""PIN desde el entorno, la config o por teclado. Nunca se escribe a disco."""
import getpass
p = os.environ.get("TIMENET_PIN") or (cfg or {}).get("pin")
return str(p) if p else getpass.getpass("PIN de Timenet: ")

158
app/store.py Normal file
View File

@@ -0,0 +1,158 @@
"""Persistencia en SQLite: plan del dia, overrides manuales y auditoria."""
from __future__ import annotations
import json
import logging
import os
import sqlite3
from datetime import date, datetime
logger = logging.getLogger(__name__)
DEFAULT_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "data", "autoficher.db")
SCHEMA = """
CREATE TABLE IF NOT EXISTS day_plan (
day TEXT NOT NULL, -- YYYY-MM-DD
idx INTEGER NOT NULL, -- orden del fichaje dentro del dia
typ INTEGER NOT NULL, -- 0 entrada, 1 salida
window_idx INTEGER NOT NULL, -- tramo al que pertenece
target TEXT NOT NULL, -- hora teorica ISO local
planned_at TEXT NOT NULL, -- instante planificado ISO local (con jitter)
state TEXT NOT NULL DEFAULT 'pending', -- pending|done|cancelled|skipped|failed
fired_at TEXT,
check_id INTEGER,
note TEXT,
announced_at TEXT,
message_id INTEGER,
PRIMARY KEY (day, idx)
);
CREATE TABLE IF NOT EXISTS overrides (
day TEXT PRIMARY KEY,
action TEXT NOT NULL, -- off|on
reason TEXT,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT NOT NULL,
kind TEXT NOT NULL,
msg TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS events_ts ON events(ts);
CREATE TABLE IF NOT EXISTS kv (k TEXT PRIMARY KEY, v TEXT NOT NULL);
"""
def _d(x) -> str:
return x.isoformat() if isinstance(x, (date, datetime)) else str(x)
class Store:
def __init__(self, path: str = DEFAULT_PATH):
self.path = os.path.abspath(path)
os.makedirs(os.path.dirname(self.path), exist_ok=True)
self.db = sqlite3.connect(self.path, detect_types=0)
self.db.row_factory = sqlite3.Row
self.db.execute("PRAGMA journal_mode=WAL")
self.db.executescript(SCHEMA)
self._migrate()
self.db.commit()
def _migrate(self) -> None:
cols = {r["name"] for r in self.db.execute("PRAGMA table_info(day_plan)")}
for col, decl in (("announced_at", "TEXT"), ("message_id", "INTEGER")):
if col not in cols:
self.db.execute(f"ALTER TABLE day_plan ADD COLUMN {col} {decl}")
logger.info("migracion: day_plan.%s añadida", col)
def close(self) -> None:
self.db.close()
# ------------------------------------------------------------- plan diario
def save_plan(self, day: date, rows: list[dict]) -> None:
"""Reemplaza el plan del dia. No toca fichajes ya realizados."""
with self.db:
done = {r["idx"] for r in self.db.execute(
"SELECT idx FROM day_plan WHERE day=? AND state IN ('done','cancelled')", (_d(day),))}
for r in rows:
if r["idx"] in done:
continue
self.db.execute(
"INSERT OR REPLACE INTO day_plan(day,idx,typ,window_idx,target,planned_at,state)"
" VALUES(?,?,?,?,?,?,COALESCE((SELECT state FROM day_plan WHERE day=? AND idx=?),'pending'))",
(_d(day), r["idx"], r["typ"], r["window_idx"], _d(r["target"]),
_d(r["planned_at"]), _d(day), r["idx"]))
def get_plan(self, day: date) -> list[sqlite3.Row]:
return list(self.db.execute(
"SELECT * FROM day_plan WHERE day=? ORDER BY idx", (_d(day),)))
def pending(self, day: date) -> list[sqlite3.Row]:
return list(self.db.execute(
"SELECT * FROM day_plan WHERE day=? AND state='pending' ORDER BY idx", (_d(day),)))
def set_state(self, day: date, idx: int, state: str,
fired_at: datetime | None = None, check_id: int | None = None,
note: str | None = None) -> None:
with self.db:
self.db.execute(
"UPDATE day_plan SET state=?, fired_at=COALESCE(?,fired_at),"
" check_id=COALESCE(?,check_id), note=COALESCE(?,note) WHERE day=? AND idx=?",
(state, _d(fired_at) if fired_at else None, check_id, note, _d(day), idx))
def mark_announced(self, day: date, idx: int, message_id: int | None) -> None:
with self.db:
self.db.execute(
"UPDATE day_plan SET announced_at=?, message_id=? WHERE day=? AND idx=?",
(datetime.now().isoformat(timespec="seconds"), message_id, _d(day), idx))
def reschedule(self, day: date, idx: int, planned_at: datetime) -> None:
with self.db:
self.db.execute(
"UPDATE day_plan SET planned_at=?, announced_at=NULL, message_id=NULL"
" WHERE day=? AND idx=? AND state='pending'",
(_d(planned_at), _d(day), idx))
def get_check(self, day: date, idx: int):
return self.db.execute(
"SELECT * FROM day_plan WHERE day=? AND idx=?", (_d(day), idx)).fetchone()
# --------------------------------------------------------------- overrides
def set_override(self, day: date, action: str, reason: str = "") -> None:
with self.db:
self.db.execute(
"INSERT OR REPLACE INTO overrides(day,action,reason,created_at) VALUES(?,?,?,?)",
(_d(day), action, reason, datetime.now().isoformat(timespec="seconds")))
def get_override(self, day: date) -> sqlite3.Row | None:
return self.db.execute("SELECT * FROM overrides WHERE day=?", (_d(day),)).fetchone()
def clear_override(self, day: date) -> None:
with self.db:
self.db.execute("DELETE FROM overrides WHERE day=?", (_d(day),))
def list_overrides(self, since: date | None = None) -> list[sqlite3.Row]:
if since:
return list(self.db.execute(
"SELECT * FROM overrides WHERE day>=? ORDER BY day", (_d(since),)))
return list(self.db.execute("SELECT * FROM overrides ORDER BY day"))
# ------------------------------------------------------------------ varios
def log(self, kind: str, msg: str) -> None:
with self.db:
self.db.execute("INSERT INTO events(ts,kind,msg) VALUES(?,?,?)",
(datetime.now().isoformat(timespec="seconds"), kind, msg))
def recent_events(self, limit: int = 30) -> list[sqlite3.Row]:
return list(self.db.execute(
"SELECT * FROM events ORDER BY id DESC LIMIT ?", (limit,)))
def get(self, key: str, default=None):
row = self.db.execute("SELECT v FROM kv WHERE k=?", (key,)).fetchone()
return json.loads(row["v"]) if row else default
def put(self, key: str, value) -> None:
with self.db:
self.db.execute("INSERT OR REPLACE INTO kv(k,v) VALUES(?,?)",
(key, json.dumps(value)))

141
app/timenet_api.py Normal file
View File

@@ -0,0 +1,141 @@
"""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