timka123
77b26757bf
- Refactored TimerPage component to improve layout and user experience. - Updated theme configuration to utilize design tokens for better maintainability. - Introduced new tokens for colors and typography to align with design principles. - Simplified state management and improved timer functionality. - Enhanced accessibility and responsiveness of UI elements.
343 lines
13 KiB
Python
343 lines
13 KiB
Python
"""A long-polling Telegram bot, so the chat is a remote control, not just an inbox.
|
||
|
||
Runs in a daemon thread next to the API. Commands do exactly what the buttons in
|
||
the UI do — they call the same services — so starting a pomodoro from your phone
|
||
and finishing it on the desktop is one unbroken interval.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import threading
|
||
import time as time_module
|
||
from typing import Any, Callable
|
||
|
||
import httpx
|
||
|
||
from app.config import env_settings
|
||
from app.db import session_scope
|
||
from app.services import (
|
||
blocker_service,
|
||
pomodoro,
|
||
report,
|
||
settings_service,
|
||
stats,
|
||
tasks_service,
|
||
)
|
||
from app.settings_schema import AppSettings
|
||
from app.timeutil import humanize_minutes, local_date
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
_thread: threading.Thread | None = None
|
||
_stop = threading.Event()
|
||
_POLL_TIMEOUT = 25
|
||
|
||
HELP_EN = """<b>Kadence</b>
|
||
|
||
/start_work [minutes] — start a pomodoro
|
||
/break — start a break
|
||
/stop — abandon the current interval
|
||
/pause · /resume — hold the clock
|
||
/status — what is running right now
|
||
/today — your todo list, grouped
|
||
/report — today's report, now
|
||
/week — the weekly review
|
||
/streak — streak and all-time totals
|
||
/block — what the blocker is doing
|
||
/help — this message"""
|
||
|
||
HELP_RU = """<b>Kadence</b>
|
||
|
||
/start_work [минуты] — запустить помидорку
|
||
/break — перерыв
|
||
/stop — бросить текущий интервал
|
||
/pause · /resume — пауза и продолжение
|
||
/status — что идёт сейчас
|
||
/today — список задач по группам
|
||
/report — отчёт за сегодня
|
||
/week — итоги недели
|
||
/streak — серия и общая статистика
|
||
/block — что делает блокировщик
|
||
/help — это сообщение"""
|
||
|
||
|
||
def _api(token: str, method: str, payload: dict[str, Any], timeout: int = 30) -> dict[str, Any]:
|
||
url = f"https://api.telegram.org/bot{token}/{method}"
|
||
with httpx.Client(timeout=timeout) as client:
|
||
resp = client.post(url, json=payload)
|
||
if not resp.is_success:
|
||
raise RuntimeError(f"{method} -> {resp.status_code}: {resp.text[:200]}")
|
||
return resp.json() or {}
|
||
|
||
|
||
def _reply(settings: AppSettings, chat_id: str, text: str, buttons: list[list[str]] | None = None) -> None:
|
||
telegram = settings.notifications.telegram
|
||
payload: dict[str, Any] = {
|
||
"chat_id": chat_id,
|
||
"text": text,
|
||
"parse_mode": "HTML",
|
||
"disable_web_page_preview": True,
|
||
}
|
||
if buttons:
|
||
payload["reply_markup"] = {
|
||
"inline_keyboard": [
|
||
[{"text": label, "callback_data": label} for label in row] for row in buttons
|
||
]
|
||
}
|
||
try:
|
||
_api(telegram.bot_token, "sendMessage", payload)
|
||
except Exception as exc: # pragma: no cover
|
||
log.warning("telegram reply failed: %s", exc)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Command handlers
|
||
# --------------------------------------------------------------------------
|
||
def _cmd_status(session, settings: AppSettings, _args: str) -> str:
|
||
ru = settings.notifications.report.language == "ru"
|
||
state = pomodoro.state(session, settings)
|
||
current = state["current"]
|
||
done, goal = state["completed_today"], state["goal"]
|
||
header = (
|
||
f"🍅 {done}/{goal} " + ("сегодня" if ru else "today")
|
||
+ f" · {report.progress_bar(done, goal)}"
|
||
)
|
||
if current is None:
|
||
return header + "\n\n" + ("Таймер стоит." if ru else "The timer is idle.")
|
||
remaining = humanize_minutes(current["remaining_seconds"] / 60)
|
||
kind = {
|
||
"work": "работа" if ru else "work",
|
||
"short_break": "короткий перерыв" if ru else "short break",
|
||
"long_break": "длинный перерыв" if ru else "long break",
|
||
}[current["kind"]]
|
||
task = f"\n📌 {report._esc(current['task']['title'])}" if current.get("task") else ""
|
||
status = ("на паузе" if ru else "paused") if current["status"] == "paused" else ("идёт" if ru else "running")
|
||
return f"{header}\n\n▶️ <b>{kind}</b> ({status}) — {remaining} " + ("осталось" if ru else "left") + task
|
||
|
||
|
||
def _cmd_start_work(session, settings: AppSettings, args: str) -> str:
|
||
ru = settings.notifications.report.language == "ru"
|
||
minutes: int | None = None
|
||
if args.strip().isdigit():
|
||
minutes = max(1, min(240, int(args.strip())))
|
||
row = pomodoro.start(session, settings, kind="work", minutes=minutes)
|
||
length = humanize_minutes(row.planned_seconds / 60)
|
||
return (
|
||
f"▶️ Помидорка на {length} пошла. Удачи!" if ru else f"▶️ {length} pomodoro started. Go."
|
||
)
|
||
|
||
|
||
def _cmd_break(session, settings: AppSettings, _args: str) -> str:
|
||
ru = settings.notifications.report.language == "ru"
|
||
kind = pomodoro.next_kind(session, settings)
|
||
if kind == "work":
|
||
kind = "short_break"
|
||
row = pomodoro.start(session, settings, kind=kind)
|
||
return (
|
||
f"☕ Перерыв {humanize_minutes(row.planned_seconds / 60)}."
|
||
if ru
|
||
else f"☕ {humanize_minutes(row.planned_seconds / 60)} break."
|
||
)
|
||
|
||
|
||
def _cmd_stop(session, settings: AppSettings, args: str) -> str:
|
||
ru = settings.notifications.report.language == "ru"
|
||
try:
|
||
pomodoro.abandon(session, settings, reason=args.strip() or "stopped from Telegram")
|
||
except pomodoro.TimerError as exc:
|
||
return str(exc)
|
||
return "⏹ Остановлено." if ru else "⏹ Stopped."
|
||
|
||
|
||
def _cmd_pause(session, settings: AppSettings, _args: str) -> str:
|
||
ru = settings.notifications.report.language == "ru"
|
||
try:
|
||
pomodoro.pause(session, settings)
|
||
except pomodoro.TimerError as exc:
|
||
return str(exc)
|
||
return "⏸ Пауза." if ru else "⏸ Paused."
|
||
|
||
|
||
def _cmd_resume(session, settings: AppSettings, _args: str) -> str:
|
||
ru = settings.notifications.report.language == "ru"
|
||
try:
|
||
pomodoro.resume(session)
|
||
except pomodoro.TimerError as exc:
|
||
return str(exc)
|
||
return "▶️ Продолжаем." if ru else "▶️ Resumed."
|
||
|
||
|
||
def _cmd_today(session, settings: AppSettings, _args: str) -> str:
|
||
"""The todo list, grouped by deadline and numbered for reading."""
|
||
ru = settings.notifications.report.language == "ru"
|
||
today = local_date(settings.timezone)
|
||
tasks = tasks_service.list_tasks(session)
|
||
buckets = tasks_service.bucket_tasks(tasks, today)
|
||
ordered = buckets["overdue"] + buckets["today"] + buckets["upcoming"] + buckets["someday"]
|
||
if not ordered:
|
||
return "✨ Задач нет." if ru else "✨ Nothing on the list."
|
||
|
||
limit = settings.notifications.report.max_tasks_listed
|
||
lines = [f"<b>{'📋 Задачи' if ru else '📋 Your tasks'}</b>"]
|
||
marks = {"overdue": "🔴", "today": "🟡", "upcoming": "🔵", "someday": "⚪"}
|
||
position = 0
|
||
for bucket_name in ("overdue", "today", "upcoming", "someday"):
|
||
for task in buckets[bucket_name]:
|
||
position += 1
|
||
if position > limit:
|
||
continue
|
||
due = f" ({task.due_date.isoformat()})" if task.due_date else ""
|
||
lines.append(
|
||
f"{marks[bucket_name]} <b>{position}.</b> {report._esc(task.title)}{due}"
|
||
)
|
||
if position > limit:
|
||
lines.append(("… и ещё " if ru else "… and ") + f"{position - limit}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _cmd_report(session, settings: AppSettings, _args: str) -> str:
|
||
today = local_date(settings.timezone)
|
||
_, payload = report.get_or_build_daily(session, settings, today, rebuild=True)
|
||
return report.render_daily_telegram(payload, settings)
|
||
|
||
|
||
def _cmd_week(session, settings: AppSettings, _args: str) -> str:
|
||
today = local_date(settings.timezone)
|
||
payload = report.build_weekly(session, settings, today)
|
||
return report.render_weekly_telegram(payload, settings)
|
||
|
||
|
||
def _cmd_streak(session, settings: AppSettings, _args: str) -> str:
|
||
ru = settings.notifications.report.language == "ru"
|
||
streak = stats.streak(session, settings)
|
||
lifetime = stats.lifetime(session, settings)
|
||
return "\n".join(
|
||
[
|
||
f"🔥 <b>{'Серия' if ru else 'Streak'}</b>: {streak['current']} "
|
||
+ ("дн." if ru else "days"),
|
||
f"🏆 {'Рекорд' if ru else 'Best'}: {streak['best']} · "
|
||
f"{'дней с целью' if ru else 'goal streak'}: {streak['goal_streak']}",
|
||
"",
|
||
f"🍅 {'Всего помидорок' if ru else 'All-time pomodoros'}: {lifetime['pomodoros']}",
|
||
f"⏱ {'Всего фокуса' if ru else 'All-time focus'}: {lifetime['focus_hours']}h",
|
||
f"✅ {'Закрыто задач' if ru else 'Tasks closed'}: {lifetime['tasks_completed']}",
|
||
f"🚫 {'Заблокировано попыток' if ru else 'Blocked attempts'}: {lifetime['blocked_attempts']}",
|
||
]
|
||
)
|
||
|
||
|
||
def _cmd_block(session, settings: AppSettings, _args: str) -> str:
|
||
ru = settings.notifications.report.language == "ru"
|
||
policy = blocker_service.policy(session, settings)
|
||
events = blocker_service.events_for_day(session, settings)
|
||
state = ("включён" if ru else "ON") if policy["active"] else ("выключен" if ru else "OFF")
|
||
lines = [
|
||
f"🛡 <b>{'Блокировщик' if ru else 'Blocker'}</b>: {state} ({report._esc(policy['reason'])})",
|
||
f"{'Сайтов' if ru else 'Domains'}: {len(policy['all_domains'])} · "
|
||
f"{'приложений' if ru else 'apps'}: {len(policy['all_apps'])}",
|
||
f"{'Попыток сегодня' if ru else 'Attempts today'}: {events['total']}",
|
||
]
|
||
for entry in events["by_target"][:5]:
|
||
lines.append(f"• {report._esc(entry['target'])} × {entry['count']}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _cmd_help(_session, settings: AppSettings, _args: str) -> str:
|
||
return HELP_RU if settings.notifications.report.language == "ru" else HELP_EN
|
||
|
||
|
||
COMMANDS: dict[str, Callable] = {
|
||
"start": _cmd_help,
|
||
"help": _cmd_help,
|
||
"status": _cmd_status,
|
||
"start_work": _cmd_start_work,
|
||
"work": _cmd_start_work,
|
||
"break": _cmd_break,
|
||
"stop": _cmd_stop,
|
||
"pause": _cmd_pause,
|
||
"resume": _cmd_resume,
|
||
"today": _cmd_today,
|
||
"tasks": _cmd_today,
|
||
"report": _cmd_report,
|
||
"week": _cmd_week,
|
||
"streak": _cmd_streak,
|
||
"stats": _cmd_streak,
|
||
"block": _cmd_block,
|
||
}
|
||
|
||
|
||
def handle_text(text: str) -> str | None:
|
||
"""Run one command and return the reply, or None when it is not for us."""
|
||
text = (text or "").strip()
|
||
if not text.startswith("/"):
|
||
return None
|
||
head, _, args = text[1:].partition(" ")
|
||
name = head.split("@", 1)[0].lower()
|
||
handler = COMMANDS.get(name)
|
||
if handler is None:
|
||
return None
|
||
|
||
with session_scope() as session:
|
||
settings = settings_service.load_settings(session)
|
||
try:
|
||
return handler(session, settings, args)
|
||
except Exception as exc: # pragma: no cover - a bad command must not kill the bot
|
||
log.exception("command /%s failed", name)
|
||
return f"⚠️ {exc}"
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Polling loop
|
||
# --------------------------------------------------------------------------
|
||
def _loop() -> None:
|
||
offset: int | None = None
|
||
while not _stop.is_set():
|
||
try:
|
||
with session_scope() as session:
|
||
settings = settings_service.load_settings(session)
|
||
telegram = settings.notifications.telegram
|
||
if not (telegram.enabled and telegram.commands_enabled and telegram.bot_token):
|
||
_stop.wait(30)
|
||
continue
|
||
|
||
payload: dict[str, Any] = {"timeout": _POLL_TIMEOUT}
|
||
if offset is not None:
|
||
payload["offset"] = offset
|
||
data = _api(
|
||
telegram.bot_token, "getUpdates", payload, timeout=_POLL_TIMEOUT + 10
|
||
)
|
||
for update in data.get("result", []):
|
||
offset = update["update_id"] + 1
|
||
message = update.get("message") or update.get("edited_message") or {}
|
||
chat_id = str((message.get("chat") or {}).get("id", ""))
|
||
if telegram.chat_id and chat_id and chat_id != str(telegram.chat_id):
|
||
continue # only the configured chat may drive the timer
|
||
reply = handle_text(message.get("text", ""))
|
||
if reply:
|
||
with session_scope() as session:
|
||
fresh = settings_service.load_settings(session)
|
||
_reply(fresh, chat_id or telegram.chat_id, reply)
|
||
except Exception as exc: # pragma: no cover
|
||
log.warning("telegram polling hiccup: %s", exc)
|
||
_stop.wait(10)
|
||
else:
|
||
time_module.sleep(0.3)
|
||
|
||
|
||
def start() -> None:
|
||
"""Start the polling thread if the environment allows it."""
|
||
global _thread
|
||
if not env_settings().telegram_bot_enabled or _thread is not None:
|
||
return
|
||
_stop.clear()
|
||
_thread = threading.Thread(target=_loop, name="telegram-bot", daemon=True)
|
||
_thread.start()
|
||
log.info("telegram command bot started")
|
||
|
||
|
||
def shutdown() -> None:
|
||
global _thread
|
||
_stop.set()
|
||
_thread = None
|