"""Interfaz de Telegram: comandos, teclados inline y killswitch. Solo atiende al ``chat_id`` configurado. Cualquier otro chat se ignora en silencio. """ from __future__ import annotations import asyncio import logging from datetime import date, datetime, timedelta from telegram import (BotCommand, BotCommandScopeChat, InlineKeyboardButton, InlineKeyboardMarkup, Update) from telegram.constants import ParseMode from telegram.ext import Application, CallbackQueryHandler, CommandHandler, ContextTypes logger = logging.getLogger(__name__) ENTRADA = 0 CB = "af" # prefijo de los callback_data # --------------------------------------------------------------------- formato DIAS = ("lunes", "martes", "miércoles", "jueves", "viernes", "sábado", "domingo") DIAS_CORTO = ("lun", "mar", "mié", "jue", "vie", "sáb", "dom") def fecha_larga(d: date) -> str: return f"{DIAS[d.weekday()]} {d:%d/%m/%Y}" def fecha_corta(d: date) -> str: return f"{DIAS_CORTO[d.weekday()]} {d:%d/%m}" def hm(minutes: int) -> str: """Minutos -> '+1h 23m' / '-4m' / '0m'.""" if not minutes: return "0m" sign = "-" if minutes < 0 else "+" m = abs(int(minutes)) return f"{sign}{m // 60}h {m % 60:02d}m" if m >= 60 else f"{sign}{m}m" def parse_day(txt: str, today: date) -> date | None: t = (txt or "").strip().lower() if t in ("", "hoy"): return today if t in ("mañana", "manana"): return today + timedelta(days=1) if t == "ayer": return today - timedelta(days=1) for fmt in ("%d/%m/%Y", "%d/%m/%y", "%d/%m", "%Y-%m-%d"): try: d = datetime.strptime(t, fmt).date() return d.replace(year=today.year) if fmt == "%d/%m" else d except ValueError: continue return None def parse_range(args: list[str], today: date) -> list[date]: """'hoy', '05/10', '05/10-09/10' -> lista de fechas.""" txt = " ".join(args).strip() if "-" in txt and not txt.startswith("-"): a, _, b = txt.partition("-") da, db = parse_day(a, today), parse_day(b, today) if da and db and da <= db: return [da + timedelta(days=i) for i in range((db - da).days + 1)] return [] d = parse_day(txt, today) return [d] if d else [] # ------------------------------------------------------------------------ teclado def check_result_text(lab: str, res: dict, manual: bool = False) -> str: """Resultado de un fichaje, contando si Timenet lo ha recortado.""" if res.get("repeated"): return f"🕐 {lab}: Timenet dice que ya estaba fichado." rec, real = res.get("recorded"), res.get("real") head = f"{'✅ (manual)' if manual else '✅'} {lab} registrada" if rec and real and rec != real: return f"{head}\nFichado {real} → cuenta como {rec}" if rec: return f"{head} a las {rec}" return f"{head} (sin confirmación de hora)" def pending_keyboard(day: date, idx: int) -> InlineKeyboardMarkup: d = day.isoformat() return InlineKeyboardMarkup([[ InlineKeyboardButton("🛑 Cancelar hoy", callback_data=f"{CB}:cancel:{d}:{idx}"), InlineKeyboardButton("⏭ +15 min", callback_data=f"{CB}:delay:{d}:{idx}"), InlineKeyboardButton("✅ Fichar ya", callback_data=f"{CB}:now:{d}:{idx}"), ]]) # Lo que Telegram mostrara en el menu de comandos. Orden = orden en el menu. MENU: list[tuple[str, str]] = [ ("hoy", "Plan y estado de hoy"), ("saldo", "Saldo del mes y acumulado"), ("plan", "Ver el plan y re-tirar los dados"), ("proximos", "Previsión de los próximos días [nº de días]"), ("fichar", "Fichar ENTRADA ahora"), ("desfichar", "Fichar SALIDA ahora"), ("off", "No fichar un dia [hoy|dd/mm|dd/mm-dd/mm]"), ("on", "Volver a fichar ese dia"), ("pausa", "Parar el servicio"), ("reanudar", "Retomar el servicio"), ("log", "Ultimos eventos"), ("ayuda", "Ayuda"), ] class BotUI: def __init__(self, service): self.svc = service self.chat_id = int((service.cfg.get("telegram") or {}).get("chat_id")) # -------------------------------------------------------------- utilidades def _mine(self, update: Update) -> bool: chat = update.effective_chat if chat is None or chat.id != self.chat_id: logger.warning("mensaje ignorado de chat no autorizado: %s", chat.id if chat else "?") return False return True def register(self, app: Application) -> None: cmds = { "start": self.cmd_hoy, "help": self.cmd_help, "ayuda": self.cmd_help, "hoy": self.cmd_hoy, "saldo": self.cmd_saldo, "plan": self.cmd_plan, "proximos": self.cmd_proximos, "off": self.cmd_off, "on": self.cmd_on, "fichar": self.cmd_fichar, "desfichar": self.cmd_desfichar, "pausa": self.cmd_pausa, "reanudar": self.cmd_reanudar, "log": self.cmd_log, } for name, fn in cmds.items(): app.add_handler(CommandHandler(name, self._guard(fn))) app.add_handler(CallbackQueryHandler(self.on_button, pattern=rf"^{CB}:")) def _guard(self, fn): async def wrapper(update: Update, ctx: ContextTypes.DEFAULT_TYPE): if not self._mine(update): return try: await fn(update, ctx) except Exception as e: logger.exception("fallo en %s", fn.__name__) await update.effective_message.reply_text(f"🟥 Error en el comando: {e}") return wrapper async def send(self, text: str, keyboard=None): return await self.svc.app.bot.send_message( chat_id=self.chat_id, text=text, reply_markup=keyboard, parse_mode=ParseMode.HTML, disable_notification=False) # ---------------------------------------------------------------- comandos async def cmd_help(self, update: Update, ctx) -> None: cuerpo = "\n".join(f"/{c} — {d}" for c, d in MENU) await update.effective_message.reply_text( f"Autoficher\n\n{cuerpo}", parse_mode=ParseMode.HTML) async def publish_menu(self) -> None: """Registra los comandos en Telegram para que salgan en el menu del chat.""" cmds = [BotCommand(c, d) for c, d in MENU] await self.svc.app.bot.set_my_commands( cmds, scope=BotCommandScopeChat(chat_id=self.chat_id)) logger.info("menu de comandos publicado en Telegram (%d comandos)", len(cmds)) async def cmd_hoy(self, update: Update, ctx) -> None: await update.effective_message.reply_text(await self._status_text(), parse_mode=ParseMode.HTML) async def _status_text(self) -> str: today = self.svc.today() lines = [f"{fecha_larga(today)}"] cal = await asyncio.to_thread(self.svc.executor.calendar) if not cal.is_workday(today): lines.append(f"🌴 No laborable: {cal.why_not_workday(today)}") else: dt = cal.day_type(today) lines.append(f"🗓 {dt.name} · previsto {dt.minutes} min") ov = self.svc.store.get_override(today) if ov: lines.append(f"🚫 Desactivado a mano: {ov['reason'] or 'sin motivo'}") if self.svc.is_paused(): lines.append("⏸ Servicio en pausa") plan = self.svc.store.get_plan(today) if plan: lines.append("\nPlan de hoy") icon = {"pending": "⏳", "done": "✅", "cancelled": "🛑", "skipped": "⤵️", "failed": "🟥"} for r in plan: t = datetime.fromisoformat(r["planned_at"]) lab = "ENTRADA" if r["typ"] == ENTRADA else "SALIDA " extra = "" if r["state"] == "done" and r["fired_at"]: extra = f" → fichado {datetime.fromisoformat(r['fired_at']):%H:%M:%S}" lines.append(f"{icon.get(r['state'], '·')} {t:%H:%M:%S} {lab}{extra}") try: s = await asyncio.to_thread(self.svc.executor.today_summary) if s["checks"]: lines.append("\nSegún Timenet") for c in s["checks"]: lab = "entrada" if c["typ"] == ENTRADA else "salida " mark = "" if c["recorded"] == c["real"] else f" (fichado {c['real']}, recortado)" lines.append(f"· {lab} {c['recorded']}{mark}") lines.append(f"trabajado {s['W']}/{s['P']} min · saldo del día {hm(s['X'])}") except Exception as e: lines.append(f"\n⚠️ No se pudo leer Timenet: {e}") return "\n".join(lines) async def cmd_saldo(self, update: Update, ctx) -> None: b = await asyncio.to_thread(self.svc.executor.balance) await update.effective_message.reply_text( f"Saldo\n" f"Mes en curso: previsto {b['P']} min · trabajado {b['W']} min\n" f"Saldo del mes: {hm(b['X'])}\n" f"Arrastrado: {hm(b['XAnt'])}\n" f"━━━━━━━━━━━━━\n" f"Acumulado: {hm(b['total'])}", parse_mode=ParseMode.HTML) async def cmd_plan(self, update: Update, ctx) -> None: kb = InlineKeyboardMarkup([[InlineKeyboardButton( "🎲 Re-tirar el plan de hoy", callback_data=f"{CB}:replan:{self.svc.today()}:0")]]) await update.effective_message.reply_text(await self._status_text(), parse_mode=ParseMode.HTML, reply_markup=kb) async def cmd_proximos(self, update: Update, ctx) -> None: try: ndias = max(1, min(60, int(ctx.args[0]))) if ctx.args else 14 except ValueError: ndias = 14 cal = await asyncio.to_thread(self.svc.executor.calendar) j = self.svc.cfg["jitter"] hoy = self.svc.today() lines = [f"Previsión · próximos {ndias} días"] laborables = 0 for i in range(ndias): d = hoy + timedelta(days=i) cabecera = f"{fecha_corta(d)}" ov = self.svc.store.get_override(d) if ov and ov["action"] == "off": lines.append(f"{cabecera} 🚫 {ov['reason'] or 'desactivado a mano'}") continue if not cal.is_workday(d): lines.append(f"{cabecera} 🌴 {cal.why_not_workday(d)}") continue laborables += 1 if d == hoy and self.svc.store.get_plan(d): icon = {"pending": "⏳", "done": "✅", "cancelled": "🛑", "skipped": "⤵️", "failed": "🟥"} trozos = [] for r in self.svc.store.get_plan(d): t = datetime.fromisoformat(r["planned_at"]) trozos.append(f"{icon.get(r['state'], '·')}{t:%H:%M:%S}") lines.append(f"{cabecera} {' '.join(trozos)}") continue # dias futuros: el instante exacto se sortea esa madrugada, pero el # rango en que caera si es exacto trozos = [] for w in cal.windows(d): ini = datetime.combine(d, datetime.min.time()) ent = ini + timedelta(minutes=w.start) sal = ini + timedelta(minutes=w.end) trozos.append( f"{ent - timedelta(seconds=j['entry_before_max']):%H:%M}" f"-{ent - timedelta(seconds=j['entry_before_min']):%H:%M}" f" → {sal + timedelta(seconds=j['exit_after_min']):%H:%M}" f"-{sal + timedelta(seconds=j['exit_after_max']):%H:%M}") lines.append(f"{cabecera} {' · '.join(trozos)}") b = await asyncio.to_thread(self.svc.executor.balance) lines.append(f"\n{laborables} días laborables · saldo acumulado {hm(b['total'])}") lines.append("Las horas futuras son el rango en que caerá el fichaje; " "el segundo exacto se sortea esa madrugada.") await update.effective_message.reply_text("\n".join(lines), parse_mode=ParseMode.HTML) async def cmd_off(self, update: Update, ctx) -> None: days = parse_range(ctx.args, self.svc.today()) if not days: await update.effective_message.reply_text( "Uso: /off [hoy | mañana | dd/mm | dd/mm-dd/mm]") return for d in days: self.svc.store.set_override(d, "off", "desde Telegram") self.svc.store.log("override", f"off {days[0]}..{days[-1]}") if self.svc.today() in days: await self.svc.cancel_pending_today("desactivado a mano") await update.effective_message.reply_text( f"🚫 No se fichará {'el ' + days[0].strftime('%d/%m') if len(days) == 1 else f'del {days[0]:%d/%m} al {days[-1]:%d/%m} ({len(days)} días)'}") async def cmd_on(self, update: Update, ctx) -> None: days = parse_range(ctx.args, self.svc.today()) if not days: await update.effective_message.reply_text("Uso: /on [hoy | dd/mm | dd/mm-dd/mm]") return for d in days: self.svc.store.clear_override(d) if self.svc.today() in days: await self.svc.ensure_plan(force=True) await update.effective_message.reply_text( f"✅ Reactivado ({len(days)} día{'s' if len(days) > 1 else ''})") async def cmd_fichar(self, update: Update, ctx) -> None: await self._manual(update, ENTRADA) async def cmd_desfichar(self, update: Update, ctx) -> None: await self._manual(update, 1) async def _manual(self, update: Update, typ: int) -> None: lab = "ENTRADA" if typ == ENTRADA else "SALIDA" msg = await update.effective_message.reply_text(f"⏳ Fichando {lab} a mano…") try: res = await asyncio.to_thread(self.svc.executor.do_check, typ) self.svc.store.log("manual", f"{lab} -> {res.get('recorded')}") await msg.edit_text(check_result_text(lab, res, manual=True), parse_mode=ParseMode.HTML) except Exception as e: logger.exception("fallo en fichaje manual") await msg.edit_text(f"🟥 No se pudo fichar {lab}: {e}") async def cmd_pausa(self, update: Update, ctx) -> None: self.svc.set_paused(True) await self.svc.cancel_pending_today("servicio en pausa") await update.effective_message.reply_text("⏸ Servicio en pausa. /reanudar para volver.") async def cmd_reanudar(self, update: Update, ctx) -> None: self.svc.set_paused(False) await self.svc.ensure_plan(force=True) await update.effective_message.reply_text("▶️ Servicio reanudado.\n\n" + await self._status_text(), parse_mode=ParseMode.HTML) async def cmd_log(self, update: Update, ctx) -> None: ev = self.svc.store.recent_events(15) if not ev: await update.effective_message.reply_text("Sin eventos todavía.") return txt = "\n".join(f"{e['ts'][5:16]} {e['kind']}: {e['msg'][:80]}" for e in ev) await update.effective_message.reply_text(f"Últimos eventos\n{txt}", parse_mode=ParseMode.HTML) # --------------------------------------------------------------- callbacks async def on_button(self, update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: q = update.callback_query if not self._mine(update): await q.answer() return try: _, action, day_s, idx_s = q.data.split(":") day, idx = date.fromisoformat(day_s), int(idx_s) except ValueError: await q.answer("Botón no reconocido") return try: if action == "cancel": self.svc.store.set_state(day, idx, "cancelled", note="cancelado desde Telegram") self.svc.store.log("killswitch", f"cancelado {day} idx={idx}") await q.answer("Cancelado") await q.edit_message_text("🛑 Fichaje cancelado desde Telegram.", parse_mode=ParseMode.HTML) elif action == "delay": row = self.svc.store.get_check(day, idx) if not row or row["state"] != "pending": await q.answer("Ya no está pendiente") return nuevo = datetime.fromisoformat(row["planned_at"]) + timedelta(minutes=15) self.svc.store.reschedule(day, idx, nuevo) self.svc.wake() await q.answer("Retrasado 15 min") await q.edit_message_text( f"⏭ Fichaje retrasado a las {nuevo:%H:%M:%S}.", parse_mode=ParseMode.HTML) elif action == "now": await q.answer("Fichando…") await q.edit_message_text("⏳ Fichando ahora…") ok = await self.svc.fire(day, idx, reason="botón «fichar ya»") await q.edit_message_text(ok, parse_mode=ParseMode.HTML) elif action == "replan": await self.svc.ensure_plan(force=True) await q.answer("Plan regenerado") await q.edit_message_text(await self._status_text(), parse_mode=ParseMode.HTML) else: await q.answer("Acción desconocida") except Exception as e: logger.exception("fallo en callback %s", q.data) await q.answer("Error") await self.send(f"🟥 Error procesando el botón: {e}")