omar
ba0d6b5f08
El cambio de perfil funcionaba, pero con 13 dias planificados por delante la diferencia entre perfiles no se apreciaba: la variedad es una propiedad estadistica y con esa muestra un 51% y un 29% se ven casi igual. - Horizonte de 21 a 35 dias naturales (~22 laborables). Con esa muestra el histograma ya se corresponde con lo anunciado: disperso da 6 horas distintas con la mas repetida al 27%, frente al 29% medido sobre el año entero. - /perfil enseña ahora el histograma REAL de lo que hay sorteado, no solo las cifras teoricas del perfil. Asi se ve el efecto del cambio en el momento, que era justo lo que faltaba para poder decidir. Verificado que los cinco perfiles siguen dando exactamente las cifras documentadas tras el refactor a perfiles por tramo. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
323 lines
13 KiB
Python
323 lines
13 KiB
Python
"""Servicio en segundo plano: planifica, avisa, ficha y atiende Telegram.
|
||
|
||
No usa ``JobQueue`` de PTB (APScheduler 3.6 no funciona en Python 3.13 y ademas
|
||
necesitamos precision al segundo). El bucle propio duerme exactamente hasta el
|
||
siguiente evento, asi que un fichaje no se desvia mas de unas decimas.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import random
|
||
import signal
|
||
from datetime import date, datetime, timedelta
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from telegram.constants import ParseMode
|
||
from telegram.ext import Application
|
||
|
||
from app import settings
|
||
from app.bot import BotUI, check_result_text, hm, pending_keyboard
|
||
from app.executor import Executor
|
||
from app.planner import ENTRADA, Planner
|
||
from app.store import Store
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
ANNOUNCE_LEAD = timedelta(minutes=12) # cuanta antelacion tiene el killswitch
|
||
MAX_SLEEP = 60.0
|
||
REPLAN_HOUR = 4 # hora a la que se planifica el dia
|
||
HORIZON_DAYS = 35 # dias que se planifican por adelantado
|
||
|
||
|
||
class Service:
|
||
def __init__(self, cfg: dict, pin: str, db_path: str | None = None):
|
||
self.cfg = cfg
|
||
self.tz = ZoneInfo(cfg.get("timezone", "Europe/Madrid"))
|
||
self.store = Store(db_path) if db_path else Store()
|
||
self.executor = Executor(cfg, self.store, pin)
|
||
self.planner = Planner(cfg, self.store)
|
||
self.app: Application | None = None
|
||
self.ui: BotUI | None = None
|
||
self._wake = asyncio.Event()
|
||
self._stop = asyncio.Event()
|
||
self._planned_for: date | None = None
|
||
self._gcal = None
|
||
|
||
# ------------------------------------------------------------------ basico
|
||
def now(self) -> datetime:
|
||
return datetime.now(self.tz)
|
||
|
||
def today(self) -> date:
|
||
return self.now().date()
|
||
|
||
def is_paused(self) -> bool:
|
||
return bool(self.store.get("paused", False))
|
||
|
||
def set_paused(self, value: bool) -> None:
|
||
self.store.put("paused", bool(value))
|
||
self.store.log("pausa", "activada" if value else "desactivada")
|
||
self.wake()
|
||
|
||
def wake(self) -> None:
|
||
self._wake.set()
|
||
|
||
async def notify(self, text: str, keyboard=None):
|
||
if self.ui is None:
|
||
logger.info("(sin bot todavia) %s", text)
|
||
return None
|
||
try:
|
||
return await self.ui.send(text, keyboard)
|
||
except Exception:
|
||
logger.exception("no se pudo enviar el mensaje por Telegram")
|
||
return None
|
||
|
||
# --------------------------------------------------------------- calendario
|
||
async def _personal_absence(self, day: date) -> str | None:
|
||
"""Ausencias personales desde Google Calendar. Solo eso: festivos y
|
||
vacaciones los da Timenet. Si Google falla, se sigue fichando."""
|
||
if not self.cfg.get("use_google_calendar", True):
|
||
return None
|
||
try:
|
||
from app import google_calendar
|
||
if self._gcal is None: # una sola autenticacion para todo el horizonte
|
||
self._gcal = await asyncio.to_thread(google_calendar.GoogleCalendar)
|
||
ev = await asyncio.to_thread(self._gcal.getEvent, day)
|
||
if ev and isinstance(ev, str) and not ev.startswith("Error"):
|
||
return ev
|
||
if ev and str(ev).startswith("Error"):
|
||
await self.notify(f"⚠️ Google Calendar no responde: <i>{ev}</i>\nSe sigue con el plan.")
|
||
except Exception as e:
|
||
logger.warning("Google Calendar fallo: %s", e)
|
||
await self.notify(f"⚠️ Google Calendar falló ({e}). Se sigue con el plan.")
|
||
return None
|
||
|
||
# --------------------------------------------------------------------- plan
|
||
def _plan_al_dia(self, day: date, cal) -> bool:
|
||
"""El plan guardado sigue cuadrando con el calendario de Timenet."""
|
||
rows = self.store.get_plan(day)
|
||
if not rows:
|
||
return False
|
||
esperado = [m for w in cal.windows(day) for m in (w.start, w.end)]
|
||
real = []
|
||
for r in rows:
|
||
t = datetime.fromisoformat(r["target"])
|
||
real.append(t.hour * 60 + t.minute)
|
||
return real == esperado
|
||
|
||
async def _plan_un_dia(self, day: date, cal, force: bool = False) -> list:
|
||
"""Planifica un dia concreto. Devuelve las filas guardadas."""
|
||
if self.is_paused():
|
||
return []
|
||
|
||
ov = self.store.get_override(day)
|
||
if ov and ov["action"] == "off":
|
||
return []
|
||
|
||
if not cal.is_workday(day):
|
||
if day == self.today():
|
||
self.store.log("plan", f"{day} no laborable: {cal.why_not_workday(day)}")
|
||
return []
|
||
|
||
# ya planificado y el calendario no ha cambiado
|
||
if not force and self._plan_al_dia(day, cal):
|
||
return self.store.get_plan(day)
|
||
|
||
ausencia = await self._personal_absence(day)
|
||
if ausencia:
|
||
self.store.set_override(day, "off", f"Google Calendar: {ausencia}")
|
||
if day == self.today():
|
||
await self.notify(f"🌴 Hoy no se ficha: <i>{ausencia}</i> (Google Calendar)")
|
||
return []
|
||
|
||
plan = self.planner.plan_day(cal, day, random.Random())
|
||
self.store.save_plan(day, self.planner.rows_for_store(plan))
|
||
logger.info("plan de %s: %s", day, [str(p) for p in plan])
|
||
return self.store.get_plan(day)
|
||
|
||
async def ensure_horizon(self, dias: int = HORIZON_DAYS, force: bool = False,
|
||
force_all: bool = False) -> None:
|
||
"""Planifica hoy y los proximos dias, para que la prevision sea exacta."""
|
||
cal = await asyncio.to_thread(self.executor.calendar)
|
||
hoy = self.today()
|
||
nuevos = 0
|
||
for i in range(dias):
|
||
d = hoy + timedelta(days=i)
|
||
antes = bool(self.store.get_plan(d))
|
||
await self._plan_un_dia(d, cal, force=force_all or (force and i == 0))
|
||
if not antes and self.store.get_plan(d):
|
||
nuevos += 1
|
||
self._planned_for = hoy
|
||
if nuevos:
|
||
self.store.log("plan", f"planificados {nuevos} dias nuevos hasta "
|
||
f"{hoy + timedelta(days=dias - 1)}")
|
||
self.wake()
|
||
|
||
async def ensure_plan(self, force: bool = False) -> list:
|
||
"""Plan de hoy. `force` lo vuelve a tirar aunque ya exista."""
|
||
await self.ensure_horizon(force=force)
|
||
return self.store.get_plan(self.today())
|
||
|
||
async def cambiar_perfil(self, nombre: str) -> None:
|
||
"""Cambia el perfil y vuelve a tirar los dados de todo lo que no se ha fichado."""
|
||
self.store.put("perfil", nombre)
|
||
self.store.log("perfil", f"cambiado a {nombre}")
|
||
logger.info("perfil de jitter cambiado a %s", nombre)
|
||
await self.ensure_horizon(force_all=True)
|
||
|
||
async def cancel_pending_today(self, reason: str) -> None:
|
||
day = self.today()
|
||
for r in self.store.pending(day):
|
||
self.store.set_state(day, r["idx"], "cancelled", note=reason)
|
||
self.wake()
|
||
|
||
# ------------------------------------------------------------------- fichar
|
||
async def fire(self, day: date, idx: int, reason: str = "") -> str:
|
||
row = self.store.get_check(day, idx)
|
||
if row is None:
|
||
return "🟥 Ese fichaje ya no existe en el plan."
|
||
if row["state"] != "pending":
|
||
return f"ℹ️ Ese fichaje ya estaba en estado <b>{row['state']}</b>."
|
||
|
||
typ = row["typ"]
|
||
lab = "ENTRADA" if typ == ENTRADA else "SALIDA"
|
||
planned = datetime.fromisoformat(row["planned_at"])
|
||
retraso = (self.now() - planned).total_seconds()
|
||
|
||
try:
|
||
res = await asyncio.to_thread(self.executor.do_check, typ)
|
||
except Exception as e:
|
||
logger.exception("fallo al fichar %s idx=%s", day, idx)
|
||
self.store.set_state(day, idx, "failed", note=str(e)[:200])
|
||
self.store.log("error", f"fichaje {lab} fallido: {e}")
|
||
txt = f"🟥 <b>No se pudo fichar {lab}</b>\n{e}"
|
||
await self.notify(txt)
|
||
return txt
|
||
|
||
self.store.set_state(day, idx, "done", fired_at=self.now(),
|
||
check_id=res.get("check_id"), note=reason or None)
|
||
self.store.log("check", f"{lab} -> {res.get('recorded')} (real {res.get('real')})")
|
||
|
||
txt = check_result_text(lab, res)
|
||
if retraso > 45:
|
||
txt += f"\n⚠️ Se ejecutó {int(retraso)}s tarde respecto del plan."
|
||
if reason:
|
||
txt += f"\n<i>{reason}</i>"
|
||
return txt
|
||
|
||
# -------------------------------------------------------------------- bucle
|
||
def _next_event(self) -> datetime | None:
|
||
"""Instante del proximo evento accionable."""
|
||
day = self.today()
|
||
best: datetime | None = None
|
||
for r in self.store.pending(day):
|
||
t = datetime.fromisoformat(r["planned_at"])
|
||
cand = t if r["announced_at"] else t - ANNOUNCE_LEAD
|
||
cand = max(cand, self.now())
|
||
if best is None or cand < best:
|
||
best = cand
|
||
# replanificacion del dia siguiente
|
||
manana = (self.now() + timedelta(days=1)).replace(
|
||
hour=REPLAN_HOUR, minute=0, second=0, microsecond=0)
|
||
return min(best, manana) if best else manana
|
||
|
||
async def tick(self) -> None:
|
||
now = self.now()
|
||
if self._planned_for != now.date():
|
||
await self.ensure_horizon()
|
||
if self.is_paused():
|
||
return
|
||
|
||
for r in self.store.pending(now.date()):
|
||
planned = datetime.fromisoformat(r["planned_at"])
|
||
|
||
if now >= planned:
|
||
txt = await self.fire(now.date(), r["idx"])
|
||
await self.notify(txt)
|
||
continue
|
||
|
||
if not r["announced_at"] and now >= planned - ANNOUNCE_LEAD:
|
||
lab = "ENTRADA" if r["typ"] == ENTRADA else "SALIDA"
|
||
falta = int((planned - now).total_seconds() // 60)
|
||
b = await asyncio.to_thread(self.executor.balance)
|
||
msg = await self.notify(
|
||
f"⏳ <b>{lab}</b> programada a las <code>{planned:%H:%M:%S}</code>"
|
||
f" (en {falta} min)\nSaldo acumulado: {hm(b['total'])}",
|
||
pending_keyboard(now.date(), r["idx"]))
|
||
self.store.mark_announced(now.date(), r["idx"],
|
||
msg.message_id if msg else None)
|
||
|
||
async def _loop(self) -> None:
|
||
while not self._stop.is_set():
|
||
try:
|
||
await self.tick()
|
||
except Exception as e:
|
||
logger.exception("fallo en el tick")
|
||
self.store.log("error", f"tick: {e}")
|
||
await self.notify(f"🟥 Error interno del servicio: {e}")
|
||
|
||
nxt = self._next_event()
|
||
delay = max(0.5, min(MAX_SLEEP, (nxt - self.now()).total_seconds()))
|
||
self._wake.clear()
|
||
try:
|
||
await asyncio.wait_for(self._wake.wait(), timeout=delay)
|
||
except asyncio.TimeoutError:
|
||
pass
|
||
|
||
# ------------------------------------------------------------------ arranque
|
||
async def run(self) -> None:
|
||
token = (self.cfg.get("telegram") or {}).get("token")
|
||
if not token:
|
||
raise ValueError("falta telegram.token en config.json")
|
||
|
||
self.app = Application.builder().token(token).job_queue(None).build()
|
||
self.ui = BotUI(self)
|
||
self.ui.register(self.app)
|
||
|
||
await asyncio.to_thread(self.executor.ensure_login)
|
||
await self.app.initialize()
|
||
await self.app.start()
|
||
await self.app.updater.start_polling(drop_pending_updates=True)
|
||
logger.info("bot de Telegram escuchando")
|
||
try:
|
||
await self.ui.publish_menu()
|
||
except Exception:
|
||
logger.exception("no se pudo publicar el menu de comandos")
|
||
|
||
await self.ensure_plan()
|
||
plan = self.store.get_plan(self.today())
|
||
resumen = ("\n".join(
|
||
f"· <code>{datetime.fromisoformat(r['planned_at']):%H:%M:%S}</code> "
|
||
f"{'ENTRADA' if r['typ'] == ENTRADA else 'SALIDA'}" for r in plan)
|
||
or "<i>hoy no toca fichar</i>")
|
||
await self.notify(f"🟢 <b>Autoficher arrancado</b>\n{resumen}")
|
||
self.store.log("arranque", f"servicio iniciado, {len(plan)} fichajes planificados")
|
||
|
||
loop = asyncio.get_running_loop()
|
||
for sig in (signal.SIGINT, signal.SIGTERM):
|
||
loop.add_signal_handler(sig, self._stop.set)
|
||
|
||
task = asyncio.create_task(self._loop())
|
||
await self._stop.wait()
|
||
logger.info("parando…")
|
||
task.cancel()
|
||
await self.notify("🔴 <b>Autoficher detenido</b>")
|
||
await self.app.updater.stop()
|
||
await self.app.stop()
|
||
await self.app.shutdown()
|
||
self.store.close()
|
||
|
||
|
||
def main() -> None:
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
|
||
datefmt="%Y-%m-%d %H:%M:%S")
|
||
logging.getLogger("httpx").setLevel(logging.WARNING)
|
||
cfg = settings.load()
|
||
asyncio.run(Service(cfg, settings.pin(cfg)).run())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|