from __future__ import annotations import asyncio import html import logging import os import signal from datetime import datetime, timedelta, timezone from urllib.parse import urlencode from aiohttp import web from telegram import ReplyKeyboardMarkup, ReplyKeyboardRemove, Update from telegram.constants import ParseMode from telegram.ext import Application, CommandHandler, ContextTypes from .config import load_config from .db import Database, parse_iso, utcnow_iso from .reminders import ReminderScheduler from .stats import ( day_range_for, format_duration, format_datetime_local, format_time_local, get_tz, local_date, range_7d, range_month, range_week, range_today, ) from .texts_sk import ( already_running, already_running_elsewhere, admin_only, fix_already_done, fix_end_before_start, fix_end_in_future, fix_not_found, fix_success, fix_usage, fix_wrong_chat, invalid_range, leaderboard_empty, leaderboard_header, leaderboard_line, no_active, menu_hint, menu_hidden, help_text, private_top_unavailable, tasks_empty, tasks_header, tasks_line, started, status_active, status_admin_empty, status_admin_header, status_admin_line, status_inactive, stats_private_groups, stats_report, stats_admin_header, stats_admin_line, stats_admin_empty, stopped, ) logging.basicConfig( level=os.environ.get("LOG_LEVEL", "INFO"), format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) logger = logging.getLogger("timebot") def _command_keyboard() -> ReplyKeyboardMarkup: return ReplyKeyboardMarkup( [ ["/start", "/end"], ["/status", "/stats", "/top"], ["/help"], ], resize_keyboard=True, ) def _display_name(user) -> str: if user.full_name: return user.full_name if user.username: return user.username return str(user.id) def _parse_task_name(args: list[str]) -> str | None: if not args: return None raw = " ".join(args).strip() if raw.startswith('"') and raw.endswith('"') and len(raw) >= 2: raw = raw[1:-1].strip() if raw.startswith("'") and raw.endswith("'") and len(raw) >= 2: raw = raw[1:-1].strip() return raw or None async def _send_dm_with_fallback(bot, user_id: int, chat_id: int, text: str) -> None: try: await bot.send_message( chat_id=user_id, text=text, parse_mode=ParseMode.HTML, disable_web_page_preview=True, ) return except Exception: pass try: await bot.send_message( chat_id=chat_id, text=text, parse_mode=ParseMode.HTML, disable_web_page_preview=True, ) except Exception: pass async def _notify_admins(bot, chat_id: int, text: str) -> None: try: admins = await bot.get_chat_administrators(chat_id) except Exception: admins = [] sent_any = False for admin in admins: if admin.user.is_bot: continue try: await bot.send_message( chat_id=admin.user.id, text=text, parse_mode=ParseMode.HTML, disable_web_page_preview=True, ) sent_any = True except Exception: pass if not sent_any: try: await bot.send_message( chat_id=chat_id, text=text, parse_mode=ParseMode.HTML, disable_web_page_preview=True, ) except Exception: pass async def _reply(update: Update, text: str, reply_markup=None) -> None: await update.message.reply_text( text, reply_markup=reply_markup, parse_mode=ParseMode.HTML, disable_web_page_preview=True, ) def _range_from_arg(arg: str | None, tz_name: str): if arg is None or arg == "today": return range_today(tz_name) if arg == "7d": return range_7d(tz_name) if arg == "month": return range_month(tz_name) return None def _weekday_name_sk(weekday: int) -> str: names = ["Po", "Ut", "St", "Št", "Pi", "So", "Ne"] if 0 <= weekday < len(names): return names[weekday] return "?" def _aggregate_day_totals( sessions, tz_name: str, start_utc: datetime | None = None, end_utc: datetime | None = None, ) -> dict: tz = get_tz(tz_name) now_utc = datetime.now(timezone.utc) day_totals: dict = {} for s in sessions: start_ts = parse_iso(s["start_ts"]).astimezone(timezone.utc) end_ts = parse_iso(s["end_ts"]).astimezone(timezone.utc) if s["end_ts"] else now_utc if start_utc: start_ts = max(start_ts, start_utc) if end_utc: end_ts = min(end_ts, end_utc) if end_ts <= start_ts: continue start_local = start_ts.astimezone(tz) end_local = end_ts.astimezone(tz) current = start_local while current.date() <= end_local.date(): day_start = datetime(current.year, current.month, current.day, tzinfo=tz) day_end = day_start + timedelta(days=1) seg_start = max(start_local, day_start) seg_end = min(end_local, day_end) if seg_end > seg_start: day_key = seg_start.date() day_totals[day_key] = day_totals.get(day_key, 0) + int((seg_end - seg_start).total_seconds()) current = day_end return day_totals def _group_day_totals_by_week(day_totals: dict): groups: dict = {} for day_key, sec in day_totals.items(): week_start = day_key - timedelta(days=day_key.weekday()) groups.setdefault(week_start, []).append((day_key, sec)) ordered = [] for week_start in sorted(groups.keys(), reverse=True): days = sorted(groups[week_start], key=lambda x: x[0]) ordered.append((week_start, days)) return ordered async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not update.effective_chat or not update.effective_user: return db: Database = context.application.bot_data["db"] reminders: ReminderScheduler = context.application.bot_data["reminders"] tz_name: str = context.application.bot_data["tz"] chat_id = update.effective_chat.id user = update.effective_user active_any = await db.get_active_session_any(user.id) if active_any: start_ts = parse_iso(active_any["start_ts"]).astimezone(timezone.utc) if active_any["chat_id"] == chat_id: minutes = int((datetime.now(timezone.utc) - start_ts).total_seconds() // 60) await _reply(update, already_running(minutes)) return since_str = format_datetime_local(start_ts, tz_name) await _reply(update, already_running_elsewhere(since_str, active_any["chat_id"])) return task_name = None if update.effective_chat.type == "private": task_name = _parse_task_name(context.args) now_iso = utcnow_iso() session_id = await db.create_session( chat_id=chat_id, user_id=user.id, username=user.username, display_name=_display_name(user), start_ts=now_iso, task_name=task_name, ) await reminders.schedule_daily_thresholds(session_id, chat_id, user.id, now_iso) await reminders.schedule_hourly_threshold(session_id, now_iso) # Start-time warnings: daily 120+ and weekly 3+ days tz = get_tz(tz_name) local_day = datetime.now(tz).date() day_range = day_range_for(local_day, tz_name) today_stats = await db.get_user_stats(chat_id, user.id, day_range.start_utc.isoformat(), day_range.end_utc.isoformat()) total_today = int(today_stats["total_sec"]) if total_today >= 120 * 60: text = "❗️❗️❗️ Upozornenie: dnes si už prekročil 120 min." await _send_dm_with_fallback(context.application.bot, user.id, chat_id, text) await _notify_admins(context.application.bot, chat_id, text) week_range = range_week(tz_name) week_sessions = await db.list_user_sessions_in_range( chat_id, user.id, week_range.start_utc.isoformat(), week_range.end_utc.isoformat() ) days_played = set() for s in week_sessions: start_ts = parse_iso(s["start_ts"]).astimezone(timezone.utc).astimezone(tz) end_ts = ( parse_iso(s["end_ts"]).astimezone(timezone.utc).astimezone(tz) if s["end_ts"] else datetime.now(timezone.utc).astimezone(tz) ) if end_ts <= start_ts: continue current = start_ts while current.date() <= end_ts.date(): days_played.add(current.date()) current = (current + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) days_without_today = {d for d in days_played if d != local_day} if len(days_without_today) >= 3: text = "❗️❗️❗️ Upozornenie: tento týždeň už máš 3 dni hrania." await _send_dm_with_fallback(context.application.bot, user.id, chat_id, text) await _notify_admins(context.application.bot, chat_id, text) now_local = format_time_local(parse_iso(now_iso), tz_name) await _reply(update, started(now_local, task_name), reply_markup=_command_keyboard()) logger.info("Start session chat_id=%s user_id=%s", chat_id, user.id) async def cmd_end(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not update.effective_chat or not update.effective_user: return db: Database = context.application.bot_data["db"] reminders: ReminderScheduler = context.application.bot_data["reminders"] tz_name: str = context.application.bot_data["tz"] chat_id = update.effective_chat.id user = update.effective_user active = await db.get_active_session(chat_id, user.id) if not active: await _reply(update, no_active()) return end_iso = utcnow_iso() start_ts = parse_iso(active["start_ts"]).astimezone(timezone.utc) end_ts = parse_iso(end_iso).astimezone(timezone.utc) duration_sec = int((end_ts - start_ts).total_seconds()) await db.end_session(active["id"], end_iso, duration_sec) reminders.cancel_session(active["id"]) today = range_today(tz_name) today_row = await db.get_user_stats(chat_id, user.id, today.start_utc.isoformat(), today.end_utc.isoformat()) today_total = format_duration(today_row["total_sec"]) await _reply(update, stopped(format_duration(duration_sec), today_total, active["task_name"])) logger.info("End session chat_id=%s user_id=%s duration_sec=%s", chat_id, user.id, duration_sec) async def cmd_fix(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not update.effective_chat or not update.effective_user: return db: Database = context.application.bot_data["db"] reminders: ReminderScheduler = context.application.bot_data["reminders"] tz_name: str = context.application.bot_data["tz"] chat = update.effective_chat user = update.effective_user if chat.type not in {"group", "supergroup"}: await _reply(update, admin_only()) return try: member = await context.application.bot.get_chat_member(chat.id, user.id) if member.status not in {"administrator", "creator"}: await _reply(update, admin_only()) return except Exception: await _reply(update, admin_only()) return args = context.args if len(args) < 2 or len(args) > 3: await _reply(update, fix_usage()) return try: session_id = int(args[0]) except ValueError: await _reply(update, fix_usage()) return tz = get_tz(tz_name) if len(args) == 2: date_obj = datetime.now(tz).date() time_part = args[1] else: try: date_obj = datetime.strptime(args[1], "%Y-%m-%d").date() except ValueError: await _reply(update, fix_usage()) return time_part = args[2] try: time_obj = datetime.strptime(time_part, "%H:%M").time() except ValueError: await _reply(update, fix_usage()) return local_dt = datetime( date_obj.year, date_obj.month, date_obj.day, time_obj.hour, time_obj.minute, tzinfo=tz, ) end_utc = local_dt.astimezone(timezone.utc) now_utc = datetime.now(timezone.utc) if end_utc > now_utc: await _reply(update, fix_end_in_future()) return session = await db.get_session_by_id(session_id) if not session: await _reply(update, fix_not_found()) return if session["chat_id"] != chat.id: await _reply(update, fix_wrong_chat()) return if session["end_ts"] is not None: await _reply(update, fix_already_done()) return start_ts = parse_iso(session["start_ts"]).astimezone(timezone.utc) if end_utc <= start_ts: await _reply(update, fix_end_before_start()) return duration_sec = int((end_utc - start_ts).total_seconds()) await db.end_session(session_id, end_utc.isoformat(), duration_sec) reminders.cancel_session(session_id) await _reply(update, fix_success(session_id, format_datetime_local(end_utc, tz_name), format_duration(duration_sec)) ) logger.info( "Fix session chat_id=%s admin_id=%s session_id=%s end_ts=%s duration_sec=%s", chat.id, user.id, session_id, end_utc.isoformat(), duration_sec, ) async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not update.effective_chat or not update.effective_user: return db: Database = context.application.bot_data["db"] tz_name: str = context.application.bot_data["tz"] chat_id = update.effective_chat.id user = update.effective_user if update.effective_chat.type in {"group", "supergroup"}: try: member = await context.application.bot.get_chat_member(chat_id, user.id) if member.status in {"administrator", "creator"}: sessions = await db.list_active_sessions_by_chat(chat_id) if not sessions: await _reply(update, status_admin_empty()) return lines = [status_admin_header()] now_utc = datetime.now(timezone.utc) for s in sessions: start_ts = parse_iso(s["start_ts"]).astimezone(timezone.utc) duration_sec = int((now_utc - start_ts).total_seconds()) since_str = format_time_local(start_ts, tz_name) name = s["display_name"] or s["username"] or str(s["user_id"]) line = status_admin_line(name, format_duration(duration_sec), since_str) if s["task_name"]: line = f"{line} — {s['task_name']}" lines.append(line) await _reply(update, "\n".join(lines)) return except Exception: pass active = await db.get_active_session(chat_id, user.id) today = range_today(tz_name) week_range = range_week(tz_name) today_row = await db.get_user_stats(chat_id, user.id, today.start_utc.isoformat(), today.end_utc.isoformat()) today_total_sec = int(today_row["total_sec"]) today_sessions = int(today_row["sessions_count"]) tz = get_tz(tz_name) week_sessions = await db.list_user_sessions_in_range( chat_id, user.id, week_range.start_utc.isoformat(), week_range.end_utc.isoformat() ) days_played = set() for s in week_sessions: start_ts = parse_iso(s["start_ts"]).astimezone(timezone.utc).astimezone(tz) end_ts = ( parse_iso(s["end_ts"]).astimezone(timezone.utc).astimezone(tz) if s["end_ts"] else datetime.now(timezone.utc).astimezone(tz) ) if end_ts <= start_ts: continue current = start_ts while current.date() <= end_ts.date(): days_played.add(current.date()) current = (current + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) if not active: summary = ( f"Tento týždeň dni: {len(days_played)}\n" f"Dnes spolu: {format_duration(today_total_sec)}\n" f"Dnes relácií: {today_sessions}" ) await _reply(update, status_inactive() + "\n" + summary) return start_ts = parse_iso(active["start_ts"]).astimezone(timezone.utc) duration_sec = int((datetime.now(timezone.utc) - start_ts).total_seconds()) since_str = format_time_local(start_ts, tz_name) # include current session in today's total if it overlaps today if start_ts >= today.start_utc: today_total_sec += duration_sec today_sessions += 1 summary = ( f"Tento týždeň dni: {len(days_played)}\n" f"Dnes spolu: {format_duration(today_total_sec)}\n" f"Dnes relácií: {today_sessions}" ) await _reply(update, status_active(format_duration(duration_sec), since_str, active["task_name"]) + "\n" + summary ) async def cmd_stats(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not update.effective_chat or not update.effective_user: return db: Database = context.application.bot_data["db"] tz_name: str = context.application.bot_data["tz"] user = update.effective_user chat = update.effective_chat args = context.args selected_arg = args[0] if args else None target_chat_id = None if chat.type == "private" and selected_arg and selected_arg.startswith("chat:"): try: target_chat_id = int(selected_arg.split(":", 1)[1]) selected_arg = None except ValueError: await _reply(update, invalid_range()) return today = range_today(tz_name) week = range_7d(tz_name) month = range_month(tz_name) week_range = range_week(tz_name) range_selected = _range_from_arg(selected_arg, tz_name) if range_selected is None and selected_arg is not None: await _reply(update, invalid_range()) return range_selected = range_selected or today if chat.type in {"group", "supergroup"}: try: member = await context.application.bot.get_chat_member(chat.id, user.id) if member.status in {"administrator", "creator"}: admins = await context.application.bot.get_chat_administrators(chat.id) admin_ids = {admin.user.id for admin in admins if not admin.user.is_bot} admin_range = range_selected admin_label = selected_arg or "today" if selected_arg is None: admin_range = week_range admin_label = "week" rows = await db.get_group_user_stats_all_chats( chat.id, admin_range.start_utc.isoformat(), admin_range.end_utc.isoformat() ) if not rows: await _reply(update, stats_admin_empty()) return lines = [stats_admin_header(admin_label)] per_user_name: dict[int, str] = {} for idx, row in enumerate(rows, start=1): name = row["display_name"] or str(row["user_id"]) per_user_name[row["user_id"]] = name lines.append( stats_admin_line(idx, name, row["total_sec"], row["sessions_count"]) ) per_user_sessions: dict[int, list] = {} week_sessions = await db.list_sessions_in_range( week_range.start_utc.isoformat(), week_range.end_utc.isoformat() ) for s in week_sessions: user_id = s["user_id"] if user_id not in per_user_name: continue per_user_sessions.setdefault(user_id, []).append(s) if not per_user_name.get(user_id): per_user_name[user_id] = s["display_name"] or s["username"] or str(user_id) if per_user_sessions: lines.append("") lines.append("🗓️ Rozpis týždňa (Po–Ne) — podľa hráča:") first_user = True for user_id in sorted(per_user_name.keys(), key=lambda uid: per_user_name[uid].lower()): sessions_for_user = per_user_sessions.get(user_id, []) day_totals = _aggregate_day_totals( sessions_for_user, tz_name, week_range.start_utc, week_range.end_utc ) if not day_totals: continue if not first_user: lines.append("") first_user = False admin_tag = " (admin)" if user_id in admin_ids else "" lines.append(f"{html.escape(per_user_name[user_id])}{admin_tag}:") for day_key in sorted(day_totals.keys()): sec = day_totals[day_key] if sec <= 0: continue label = f"{_weekday_name_sk(day_key.weekday())} {day_key.strftime('%d.%m.')}" lines.append(f"• {html.escape(label)}: {format_duration(sec)}") total_week_sec = sum(day_totals.values()) lines.append( f"Spolu týždeň: {format_duration(total_week_sec)}, dni: {len(day_totals)}" ) await _reply(update, "\n".join(lines)) return except Exception: pass if chat.type == "private": chat_id_for_query = target_chat_id else: chat_id_for_query = None today_row = await db.get_user_stats(chat_id_for_query, user.id, today.start_utc.isoformat(), today.end_utc.isoformat()) week_row = await db.get_user_stats(chat_id_for_query, user.id, week.start_utc.isoformat(), week.end_utc.isoformat()) month_row = await db.get_user_stats(chat_id_for_query, user.id, month.start_utc.isoformat(), month.end_utc.isoformat()) range_row = await db.get_user_stats( chat_id_for_query, user.id, range_selected.start_utc.isoformat(), range_selected.end_utc.isoformat() ) count = int(range_row["sessions_count"]) total = int(range_row["total_sec"]) avg_sec = int(total / count) if count else 0 report = stats_report( today_row["total_sec"], week_row["total_sec"], month_row["total_sec"], count, avg_sec, ) if chat.type == "private" and target_chat_id is None: groups = await db.get_group_totals_for_user( user.id, range_selected.start_utc.isoformat(), range_selected.end_utc.isoformat(), limit=3 ) top_groups = [(row["chat_id"], format_duration(row["total_sec"])) for row in groups] report = report + "\n" + stats_private_groups(top_groups) if chat_id_for_query is None: week_sessions = await db.list_user_sessions_in_range_all_chats( user.id, week_range.start_utc.isoformat(), week_range.end_utc.isoformat() ) else: week_sessions = await db.list_user_sessions_in_range( chat_id_for_query, user.id, week_range.start_utc.isoformat(), week_range.end_utc.isoformat() ) day_totals = _aggregate_day_totals( week_sessions, tz_name, week_range.start_utc, week_range.end_utc ) if day_totals: lines = ["🗓️ Rozpis týždňa (Po–Ne):"] for day_key in sorted(day_totals.keys()): sec = day_totals[day_key] if sec <= 0: continue label = f"{_weekday_name_sk(day_key.weekday())} {day_key.strftime('%d.%m.')}" lines.append(f"• {label}: {format_duration(sec)}") total_week_sec = sum(day_totals.values()) lines.append( f"Spolu týždeň: {format_duration(total_week_sec)}, dni: {len(day_totals)}" ) report = report + "\n" + "\n".join(lines) await _reply(update, report) async def cmd_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not update.effective_chat or not update.effective_user: return await _reply(update, menu_hint(), reply_markup=_command_keyboard()) async def cmd_hide(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not update.effective_chat or not update.effective_user: return await _reply(update, menu_hidden(), reply_markup=ReplyKeyboardRemove()) async def cmd_top(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not update.effective_chat or not update.effective_user: return db: Database = context.application.bot_data["db"] tz_name: str = context.application.bot_data["tz"] chat = update.effective_chat arg = context.args[0] if context.args else "7d" time_range = _range_from_arg(arg, tz_name) if time_range is None: await _reply(update, invalid_range()) return if chat.type == "private": rows = await db.get_user_task_totals( update.effective_user.id, time_range.start_utc.isoformat(), time_range.end_utc.isoformat(), ) if not rows: await _reply(update, tasks_empty()) return lines = [tasks_header(arg)] for idx, row in enumerate(rows, start=1): lines.append(tasks_line(idx, row["task_name"], row["total_sec"], row["sessions_count"])) await _reply(update, "\n".join(lines)) return rows = await db.get_leaderboard_all_chats( chat.id, time_range.start_utc.isoformat(), time_range.end_utc.isoformat() ) if not rows: await _reply(update, leaderboard_empty()) return lines = [leaderboard_header(arg)] for idx, row in enumerate(rows, start=1): name = row["display_name"] or str(row["user_id"]) lines.append(leaderboard_line(idx, name, row["total_sec"])) await _reply(update, "\n".join(lines)) async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not update.effective_chat or not update.effective_user: return await _reply(update, help_text(), reply_markup=_command_keyboard()) async def handle_health(request: web.Request) -> web.Response: return web.Response(text="ok") async def handle_web_ui(request: web.Request) -> web.Response: app = request.app application: Application = app["ptb_app"] db: Database = application.bot_data["db"] tz_name: str = application.bot_data["tz"] web_ui_secret: str | None = application.bot_data.get("web_ui_secret") # Use relative links in HTML so proxy paths like /timebot/ work without special headers. if web_ui_secret: token = request.query.get("token", "") if token != web_ui_secret: return web.Response(status=403, text="forbidden") limit = 200 try: limit = min(500, max(1, int(request.query.get("limit", "200")))) except ValueError: limit = 200 chat_id = None chat_id_param = request.query.get("chat_id") if chat_id_param: try: chat_id = int(chat_id_param) except ValueError: chat_id = None name_param = request.query.get("name") if name_param: sessions_all = await db.list_sessions_by_name(name_param) now_utc = datetime.now(timezone.utc) def _overlap_seconds(s, start_utc: datetime | None, end_utc: datetime | None) -> int: start_ts = parse_iso(s["start_ts"]).astimezone(timezone.utc) end_ts = parse_iso(s["end_ts"]).astimezone(timezone.utc) if s["end_ts"] else now_utc if start_utc: start_ts = max(start_ts, start_utc) if end_utc: end_ts = min(end_ts, end_utc) if end_ts <= start_ts: return 0 return int((end_ts - start_ts).total_seconds()) def _aggregate_tasks(start_utc: datetime | None, end_utc: datetime | None): totals: dict[str, int] = {} for s in sessions_all: sec = _overlap_seconds(s, start_utc, end_utc) if sec <= 0: continue task = s["task_name"] or "Bez názvu" totals[task] = totals.get(task, 0) + sec rows = sorted(totals.items(), key=lambda x: x[1], reverse=True) return rows today = range_today(tz_name) week = range_week(tz_name) month = range_month(tz_name) today_rows = _aggregate_tasks(today.start_utc, today.end_utc) week_rows = _aggregate_tasks(week.start_utc, week.end_utc) month_rows = _aggregate_tasks(month.start_utc, month.end_utc) all_rows = _aggregate_tasks(None, None) query_params = {} if web_ui_secret: query_params["token"] = request.query.get("token", "") if chat_id is not None: query_params["chat_id"] = str(chat_id) if limit: query_params["limit"] = str(limit) back_url = "." if query_params: back_url = f"?{urlencode(query_params)}" def _table(title: str, rows: list[tuple[str, int]]) -> str: if not rows: return f"

{title}

(žiadne dáta)
" body = "".join( f"{html.escape(name)}{format_duration(sec)}" for name, sec in rows ) return ( f"

{title}

" "" f"{body}
ÚlohaČas
" ) day_totals = _aggregate_day_totals(sessions_all, tz_name) weekly_groups = _group_day_totals_by_week(day_totals) def _weekly_html() -> str: if not weekly_groups: return "

Týždne

(žiadne dáta)
" parts = ["

Týždne

"] for week_start, days in weekly_groups: if not days: continue week_end = week_start + timedelta(days=6) label = f"Týždeň {week_start.strftime('%d.%m.%Y')} – {week_end.strftime('%d.%m.%Y')}" parts.append(f"

{label}

") rows = "".join( f"{_weekday_name_sk(day_key.weekday())} {day_key.strftime('%d.%m.%Y')}" f"{format_duration(sec)}" for day_key, sec in days ) parts.append("") parts.append(f"{rows}
DeňČas
") total_week_sec = sum(sec for _, sec in days) parts.append( f"
Spolu týždeň: {format_duration(total_week_sec)}, dni: {len(days)}
" ) return "".join(parts) page = f""" Timebot – Detail používateľa

Timebot – Detail používateľa

Meno: {html.escape(name_param)}
← Späť
{_table("Dnes", today_rows)} {_table("Tento týždeň", week_rows)} {_table("Tento mesiac", month_rows)} {_table("Celkom", all_rows)} {_weekly_html()} """ return web.Response(text=page, content_type="text/html") sessions = await db.list_sessions(limit=limit, chat_id=chat_id) today = range_today(tz_name) week = range_7d(tz_name) month = range_month(tz_name) today_row = await db.get_summary(today.start_utc.isoformat(), today.end_utc.isoformat()) week_row = await db.get_summary(week.start_utc.isoformat(), week.end_utc.isoformat()) month_row = await db.get_summary(month.start_utc.isoformat(), month.end_utc.isoformat()) allowance_days = 3 allowance_day_seconds = 2 * 60 * 60 week_range = range_week(tz_name) week_sessions = await db.list_sessions_in_range( week_range.start_utc.isoformat(), week_range.end_utc.isoformat() ) per_user_day: dict[int, dict[str, int]] = {} per_user_name: dict[int, str] = {} tz = get_tz(tz_name) for s in week_sessions: user_id = s["user_id"] per_user_name[user_id] = s["display_name"] or s["username"] or str(user_id) start_ts = parse_iso(s["start_ts"]).astimezone(timezone.utc) end_ts = parse_iso(s["end_ts"]).astimezone(timezone.utc) if s["end_ts"] else datetime.now(timezone.utc) # Clip to the week window start_ts = max(start_ts, week_range.start_utc) end_ts = min(end_ts, week_range.end_utc) if end_ts <= start_ts: continue start_local = start_ts.astimezone(tz) end_local = end_ts.astimezone(tz) current = start_local while current.date() <= end_local.date(): day_start = datetime(current.year, current.month, current.day, tzinfo=tz) day_end = day_start + timedelta(days=1) seg_start = max(start_local, day_start) seg_end = min(end_local, day_end) if seg_end > seg_start: day_key = str(seg_start.date()) per_user_day.setdefault(user_id, {}).setdefault(day_key, 0) per_user_day[user_id][day_key] += int((seg_end - seg_start).total_seconds()) current = day_end allowance_rows = [] for user_id, day_totals in per_user_day.items(): days_played = [d for d, sec in day_totals.items() if sec > 0] days_over = [d for d, sec in day_totals.items() if sec > allowance_day_seconds] total_week_sec = sum(day_totals.values()) over_days = max(0, len(days_played) - allowance_days) day_list = [] for d in sorted(day_totals.keys()): sec = day_totals[d] mark = "!" if sec > allowance_day_seconds else "" day_list.append(f"{d}: {format_duration(sec)}{mark}") allowance_rows.append( ( per_user_name.get(user_id, str(user_id)), len(days_played), len(days_over), over_days, format_duration(total_week_sec), ", ".join(day_list), ) ) allowance_rows.sort(key=lambda r: (r[3], r[2], r[1], r[0]), reverse=True) query_params = {} if web_ui_secret: token = request.query.get("token", "") if token: query_params["token"] = token if chat_id is not None: query_params["chat_id"] = str(chat_id) if limit: query_params["limit"] = str(limit) rows_html = [] for s in sessions: start_ts = parse_iso(s["start_ts"]).astimezone(timezone.utc) end_ts = parse_iso(s["end_ts"]).astimezone(timezone.utc) if s["end_ts"] else None duration_sec = s["duration_sec"] if duration_sec is None: duration_sec = int((datetime.now(timezone.utc) - start_ts).total_seconds()) status = "active" if s["end_ts"] is None else "done" display_name = s["display_name"] or s["username"] or str(s["user_id"]) name_link = display_name if display_name: name_query = dict(query_params) name_query["name"] = display_name name_link = f'{html.escape(display_name)}' rows_html.append( "" f"{s['id']}" f"{s['chat_id']}" f"{s['user_id']}" f"{name_link}" f"{html.escape(s['username'] or '')}" f"{html.escape(s['task_name'] or '')}" f"{format_datetime_local(start_ts, tz_name)}" f"{format_datetime_local(end_ts, tz_name) if end_ts else '-'}" f"{format_duration(duration_sec)}" f"{status}" "" ) page = f""" Timebot – Relácie

Timebot – Relácie

Dnes
{format_duration(today_row['total_sec'])}
relácie: {today_row['sessions_count']}
7 dní
{format_duration(week_row['total_sec'])}
relácie: {week_row['sessions_count']}
Tento mesiac
{format_duration(month_row['total_sec'])}
relácie: {month_row['sessions_count']}

Týždenné limity

Limit: max {allowance_days} dni / týždeň, max 02:00 na deň. Známka ! = prekročenie denného limitu.
{''.join( '' f'' f'' f'' f'' f'' f'' '' for r in allowance_rows )}
Meno Dni hrania Dni nad 02:00 Nad limit dní Spolu týždeň Detaily dní
{html.escape(r[0])}{r[1]}{r[2]}{r[3]}{r[4]}{html.escape(r[5])}
Zobrazené relácie: {len(sessions)} (limit {limit})
{''.join(rows_html)}
ID Chat User Meno Username Úloha Štart Koniec Trvanie Stav
""" return web.Response(text=page, content_type="text/html") async def handle_webhook(request: web.Request) -> web.Response: application: Application = request.app["ptb_app"] if request.method != "POST": return web.Response(status=405) try: data = await request.json() except Exception: return web.Response(status=400, text="invalid json") update = Update.de_json(data, application.bot) await application.process_update(update) return web.Response(text="ok") async def main() -> None: config = load_config() db = Database(config.sqlite_path) await db.connect() application = Application.builder().token(config.token).build() application.add_handler(CommandHandler("start", cmd_start)) application.add_handler(CommandHandler("end", cmd_end)) application.add_handler(CommandHandler("fix", cmd_fix)) application.add_handler(CommandHandler("status", cmd_status)) application.add_handler(CommandHandler("stats", cmd_stats)) application.add_handler(CommandHandler("menu", cmd_menu)) application.add_handler(CommandHandler("hide", cmd_hide)) application.add_handler(CommandHandler("top", cmd_top)) application.add_handler(CommandHandler("help", cmd_help)) application.bot_data["db"] = db application.bot_data["tz"] = config.tz application.bot_data["web_ui_secret"] = config.web_ui_secret await application.initialize() await application.start() reminders = ReminderScheduler(db, application.bot, config.reminder_dm, config.tz) reminders.start() application.bot_data["reminders"] = reminders await application.bot.set_webhook(url=config.webhook_url) logger.info("Webhook set to %s", config.webhook_url) await reminders.rehydrate() app = web.Application() app["ptb_app"] = application app.router.add_get("/health", handle_health) app.router.add_get("/web", handle_web_ui) app.router.add_post(config.webhook_path, handle_webhook) runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, config.listen_host, config.listen_port) await site.start() logger.info("Server listening on %s:%s", config.listen_host, config.listen_port) stop_event = asyncio.Event() def _stop() -> None: stop_event.set() loop = asyncio.get_running_loop() for sig in (signal.SIGINT, signal.SIGTERM): try: loop.add_signal_handler(sig, _stop) except NotImplementedError: pass await stop_event.wait() logger.info("Shutting down...") await runner.cleanup() await reminders.shutdown() await application.stop() await application.shutdown() await db.close() if __name__ == "__main__": asyncio.run(main())