diff --git a/app/executor.py b/app/executor.py new file mode 100644 index 0000000..441daf6 --- /dev/null +++ b/app/executor.py @@ -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, + } diff --git a/app/jitter.py b/app/jitter.py new file mode 100644 index 0000000..8c528b2 --- /dev/null +++ b/app/jitter.py @@ -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) diff --git a/app/planner.py b/app/planner.py new file mode 100644 index 0000000..e766d40 --- /dev/null +++ b/app/planner.py @@ -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] diff --git a/app/schedule.py b/app/schedule.py new file mode 100644 index 0000000..dba9632 --- /dev/null +++ b/app/schedule.py @@ -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 diff --git a/app/settings.py b/app/settings.py new file mode 100644 index 0000000..8827a83 --- /dev/null +++ b/app/settings.py @@ -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: ") diff --git a/app/store.py b/app/store.py new file mode 100644 index 0000000..096a781 --- /dev/null +++ b/app/store.py @@ -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))) diff --git a/app/timenet_api.py b/app/timenet_api.py new file mode 100644 index 0000000..e8e0693 --- /dev/null +++ b/app/timenet_api.py @@ -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 diff --git a/tools/dryrun.py b/tools/dryrun.py new file mode 100644 index 0000000..38a1d55 --- /dev/null +++ b/tools/dryrun.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +"""Simula el plan de todo el año y comprueba que Timenet contaria 0 de saldo. No ficha nada.""" +import json, os, random, sys +from collections import Counter +from datetime import timedelta + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from app import settings, schedule, jitter +from app.planner import Planner, ENTRADA + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +CACHE = sys.argv[1] if len(sys.argv) > 1 else os.path.join(ROOT, "data", "calendar_cache.json") + + +def load_calendar(cfg): + if CACHE and os.path.exists(CACHE): + print("calendario desde cache: %s" % CACHE) + return schedule.Calendar(json.load(open(CACHE))) + from app.timenet_api import TimenetClient + cli = TimenetClient(cfg["api_url"], cfg["user"]) + cli.login(settings.pin(cfg)) + return schedule.Calendar(cli.raw_calendar()) + + +def main(): + cfg = settings.load() + cal = load_calendar(cfg) + rng = random.Random(int(os.environ.get("SEED", "1234"))) + pl = Planner(cfg, store=None) + + lo, hi = cal.span + dias = 0 + tot_P = tot_W = tot_pres = 0 + offs = {"ENTRADA": [], "SALIDA": []} + muestra = [] + minutos = {"ENTRADA": Counter(), "SALIDA": Counter()} + + d = lo + while d <= hi: + if cal.is_workday(d): + plan = pl.plan_day(cal, d, rng) + dias += 1 + for wi, w in enumerate(cal.windows(d)): + ent = [p for p in plan if p.window_idx == wi and p.typ == ENTRADA][0] + sal = [p for p in plan if p.window_idx == wi and p.typ != ENTRADA][0] + # Timenet trabaja a granularidad de MINUTO: trunca los segundos + em = ent.planned_at.hour * 60 + ent.planned_at.minute + sm = sal.planned_at.hour * 60 + sal.planned_at.minute + tot_W += min(sm, w.end) - max(em, w.start) + tot_P += w.minutes + tot_pres += sm - em + for p in plan: + offs[p.label].append(p.offset_seconds) + minutos[p.label][p.planned_at.strftime("%H:%M")] += 1 + if len(muestra) < 5: + muestra.append((d, plan)) + d += timedelta(days=1) + + saldo = tot_W - tot_P + print("\ndias laborables simulados: %d (%s -> %s)" % (dias, lo, hi)) + print(" previsto P = %6d min" % tot_P) + print(" contado W = %6d min" % tot_W) + print(" SALDO = %+d min <-- objetivo: 0" % saldo) + print(" presencia = %6d min (%+d vs previsto = %.2f min/dia de mas)" + % (tot_pres, tot_pres - tot_P, (tot_pres - tot_P) / dias)) + + print("\n=== margen respecto de la hora teorica (segundos, lado seguro) ===") + for k, v in offs.items(): + v = sorted(v) + print(" %-7s n=%-4d min %3d p25 %3d mediana %3d p75 %3d max %3d media %.0f" + % (k, len(v), v[0], v[len(v) // 4], v[len(v) // 2], v[3 * len(v) // 4], v[-1], sum(v) / len(v))) + + print("\n=== minuto en que lo registra Timenet ===") + for k in ("ENTRADA", "SALIDA"): + top = sorted(minutos[k].items())[:10] + print(" %-7s %s" % (k, " ".join("%s x%d" % t for t in top))) + + print("\n=== muestra de dias ===") + for d, plan in muestra: + print(" %s %s (%s)" % (d, d.strftime("%a"), cal.day_type(d).name.strip())) + for p in plan: + print(" %s" % p) + + print("\n=== jitter de posicion (8 muestras, radio %s m) ===" % cfg["geo_jitter_meters"]) + g = cfg["geo"] + for _ in range(8): + la, ln = jitter.jitter_geo(g["latitude"], g["longitude"], cfg["geo_jitter_meters"], rng) + print(" %.6f,%.6f" % (la, ln)) + + if saldo < 0: + print("\n!! SALDO NEGATIVO: el plan pierde minutos") + sys.exit(1) + print("\nOK: ningun minuto perdido.") + + +if __name__ == "__main__": + main() diff --git a/tools/test_daemon.py b/tools/test_daemon.py new file mode 100644 index 0000000..b5077d1 --- /dev/null +++ b/tools/test_daemon.py @@ -0,0 +1,151 @@ +#!/usr/bin/env python3 +"""Ejercita el servicio con reloj simulado, sin red ni Telegram. + +Comprueba el ciclo completo: planificar -> avisar -> fichar, el killswitch, +el retraso de 15 min, la pausa y la recuperacion de fichajes perdidos. +""" +import asyncio, json, os, sys, tempfile +from datetime import date, datetime, timedelta + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from app import settings, schedule +from app.daemon import ANNOUNCE_LEAD, Service +from app.planner import ENTRADA + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +CAL = json.load(open(os.path.join(ROOT, "data", "calendar_cache.json"))) +OK = FAIL = 0 + + +def check(cond, label): + global OK, FAIL + print((" ✅ " if cond else " ❌ ") + label) + OK, FAIL = OK + bool(cond), FAIL + (not cond) + + +class FakeExecutor: + """Sustituye la red por respuestas deterministas.""" + def __init__(self): + self.cal = schedule.Calendar(CAL) + self.checks = [] + self.fail_next = False + + def ensure_login(self, force=False): pass + def calendar(self, force=False): return self.cal + def balance(self, day=None): return {"P": 9750, "W": 7810, "X": -50, "XAnt": -726, "total": -776} + def today_summary(self, day=None): return {"checks": [], "P": 510, "W": 0, "X": 0, "tipo": "H 8"} + + def do_check(self, typ): + if self.fail_next: + self.fail_next = False + raise RuntimeError("timeout simulado") + self.checks.append(typ) + return {"typ": typ, "repeated": False, "check_id": 1000 + len(self.checks), + "recorded": "09:00", "real": "08:58"} + + +class FakeService(Service): + def __init__(self, cfg, db): + super().__init__(cfg, pin="0000", db_path=db) + self.executor = FakeExecutor() + self.clock = datetime(2026, 9, 28, 3, 0, tzinfo=self.tz) # lunes laborable + self.sent = [] + + def now(self): return self.clock + async def notify(self, text, keyboard=None): + self.sent.append((text, keyboard)) + class M: message_id = len(self.sent) + return M() + + async def advance_to(self, when): + """Avanza el reloj disparando tick() en cada evento intermedio.""" + while self.clock < when: + nxt = self._next_event() + self.clock = min(when, max(nxt, self.clock + timedelta(seconds=1))) + await self.tick() + + +async def main(): + cfg = settings.load() + cfg["use_google_calendar"] = False # el test no debe salir a la red + db = os.path.join(tempfile.mkdtemp(), "t.db") + svc = FakeService(cfg, db) + day = date(2026, 9, 28) + + print("\n--- plan del dia ---") + await svc.ensure_plan() + plan = svc.store.get_plan(day) + check(len(plan) == 4, "se planifican 4 fichajes (lunes, 2 tramos)") + for r in plan: + print(" idx=%s %-7s %s" % (r["idx"], "ENTRADA" if r["typ"] == ENTRADA else "SALIDA", + datetime.fromisoformat(r["planned_at"]).strftime("%H:%M:%S"))) + p0 = datetime.fromisoformat(plan[0]["planned_at"]) + check(p0.hour == 8 and p0.minute >= 55, "la primera entrada cae antes de las 09:00") + + print("\n--- aviso con antelacion (killswitch) ---") + await svc.advance_to(p0 - ANNOUNCE_LEAD + timedelta(seconds=5)) + check(any("ENTRADA" in t and "programada" in t for t, _ in svc.sent), "llega el aviso previo") + check(any(k is not None for _, k in svc.sent), "el aviso trae teclado inline") + check(svc.store.get_check(day, 0)["announced_at"] is not None, "queda marcado como anunciado") + check(not svc.executor.checks, "todavia NO se ha fichado") + + print("\n--- el fichaje se ejecuta a su hora ---") + await svc.advance_to(p0 + timedelta(seconds=2)) + check(svc.executor.checks == [ENTRADA], "se ficha la entrada y solo esa") + check(svc.store.get_check(day, 0)["state"] == "done", "queda en estado done") + check(svc.store.get_check(day, 0)["check_id"] == 1001, "guarda el id devuelto por Timenet") + + print("\n--- killswitch: cancelar el siguiente ---") + svc.store.set_state(day, 1, "cancelled", note="prueba") + p2 = datetime.fromisoformat(svc.store.get_check(day, 2)["planned_at"]) + await svc.advance_to(p2 + timedelta(seconds=2)) + check(svc.executor.checks == [ENTRADA, ENTRADA], "el cancelado NO se ficha, el siguiente si") + + print("\n--- retraso de 15 min ---") + svc2 = FakeService(cfg, os.path.join(tempfile.mkdtemp(), "t2.db")) + await svc2.ensure_plan() + orig = datetime.fromisoformat(svc2.store.get_check(day, 0)["planned_at"]) + svc2.store.reschedule(day, 0, orig + timedelta(minutes=15)) + await svc2.advance_to(orig + timedelta(seconds=30)) + check(not svc2.executor.checks, "tras retrasarlo no ficha a la hora vieja") + check(svc2.store.get_check(day, 0)["announced_at"] is None, "se vuelve a anunciar") + await svc2.advance_to(orig + timedelta(minutes=15, seconds=5)) + check(svc2.executor.checks == [ENTRADA], "ficha a la hora nueva") + + print("\n--- pausa global ---") + svc3 = FakeService(cfg, os.path.join(tempfile.mkdtemp(), "t3.db")) + await svc3.ensure_plan() + svc3.set_paused(True) + last = datetime.fromisoformat(svc3.store.get_plan(day)[-1]["planned_at"]) + await svc3.advance_to(last + timedelta(minutes=1)) + check(not svc3.executor.checks, "en pausa no ficha nada") + + print("\n--- recuperacion: el equipo estaba apagado ---") + svc4 = FakeService(cfg, os.path.join(tempfile.mkdtemp(), "t4.db")) + await svc4.ensure_plan() + svc4.clock = datetime(2026, 9, 28, 16, 0, tzinfo=svc4.tz) # arranca tardisimo + await svc4.tick() + check(len(svc4.executor.checks) == 3, "ficha los 3 que se habian perdido, en orden") + check(any("tarde respecto del plan" in t for t, _ in svc4.sent), "avisa de que van tarde") + + print("\n--- fallo de red al fichar ---") + svc5 = FakeService(cfg, os.path.join(tempfile.mkdtemp(), "t5.db")) + await svc5.ensure_plan() + svc5.executor.fail_next = True + f0 = datetime.fromisoformat(svc5.store.get_check(day, 0)["planned_at"]) + await svc5.advance_to(f0 + timedelta(seconds=2)) + check(svc5.store.get_check(day, 0)["state"] == "failed", "el fichaje queda marcado failed") + check(any("No se pudo fichar" in t for t, _ in svc5.sent), "avisa del fallo por Telegram") + + print("\n--- dia no laborable ---") + svc6 = FakeService(cfg, os.path.join(tempfile.mkdtemp(), "t6.db")) + svc6.clock = datetime(2026, 9, 24, 3, 0, tzinfo=svc6.tz) # Festiu Local + await svc6.ensure_plan() + check(not svc6.store.get_plan(date(2026, 9, 24)), "no planifica en festivo local") + + print("\n%d OK, %d fallos" % (OK, FAIL)) + sys.exit(1 if FAIL else 0) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tools/tn_inspect.py b/tools/tn_inspect.py new file mode 100644 index 0000000..bfb7d62 --- /dev/null +++ b/tools/tn_inspect.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 +"""Inspecciona el calendario laboral y el saldo tal y como los ve Timenet. Solo lectura.""" +import os, sys +from datetime import date, timedelta + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from app import settings, schedule +from app.timenet_api import TimenetClient + + +def main(): + cfg = settings.load() + cli = TimenetClient(cfg["api_url"], cfg["user"]) + print("servidor Timenet: %s (cliente manda %s)" % (cli.version(), cli.wcp_version)) + cli.login(settings.pin(cfg)) + print("login OK\n") + + cal = schedule.Calendar(cli.raw_calendar()) + lo, hi = cal.span + print("=== TIPOS DE JORNADA (%d) ===" % len(cal.day_types)) + for dt in sorted(cal.day_types.values(), key=lambda d: d.id): + flag = " <- laborable" if dt.is_workday else "" + warn = "" if (not dt.windows or dt.consistent) else " !! tramos no cuadran con minutos" + print(" %s%s%s" % (dt, flag, warn)) + print("\ncalendario cargado: %s -> %s (%d dias)" % (lo, hi, len(cal.days))) + + print("\n=== PROXIMOS 14 DIAS ===") + hoy = date.today() + for i in range(14): + d = hoy + timedelta(days=i) + dt = cal.day_type(d) + if cal.is_workday(d): + tramos = " ".join(str(w) for w in cal.windows(d)) + margenes = ", ".join("ent-%d/sal+%d" % (w.entry_margin, w.exit_margin) for w in cal.windows(d)) + print(" %s %-3s LABORABLE %-24s %4dmin margenes: %s" % ( + d, d.strftime("%a"), tramos, dt.minutes, margenes)) + else: + print(" %s %-3s --- %s" % (d, d.strftime("%a"), cal.why_not_workday(d))) + + print("\n=== SALDO (cp/checks del mes en curso) ===") + first = hoy.replace(day=1) + nxt = (first + timedelta(days=32)).replace(day=1) - timedelta(days=1) + ck = cli.raw_checks(first.strftime("%d/%m/%Y"), nxt.strftime("%d/%m/%Y")) + print(" previsto mes P = %s min" % ck.get("P")) + print(" trabajado W = %s min" % ck.get("W")) + print(" saldo del mes X = %+s min" % ck.get("X")) + print(" saldo anterior XAnt= %+s min" % ck.get("XAnt")) + tot = (ck.get("XAnt") or 0) + (ck.get("X") or 0) + print(" SALDO ACUMULADO = %+d min (%+.2f h)" % (tot, tot / 60)) + + +if __name__ == "__main__": + main()