Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -215,10 +215,11 @@ group=7 # ai_spam_monitor: Runs LAST, classifier.dev AI spam monitoring (last
- Per-group kill switch: `"plugins": {"ai_spam_monitor": false}` / `"plugins": {"ai_spam_callback": false}` in groups.json; no alert chat configured → logfire-only monitoring

### Rich Text (DM Admin Commands)
- Data-dense DM admin outputs use Bot API 10.1+ native rich messages (`context.bot.send_rich_message` with `aiogram.types.InputRichMessage(html=...)`); currently only `/trusted` (native `<table bordered striped>` with columns User | User ID | Added by | Date). Same pattern as the mediadownloadbot metadata table
- Rich is the default with a **Markdown fallback**: any `send_rich_message` failure is caught (broad except, logged with `exc_info=True`) and the handler sends the pre-existing Markdown message instead — the Markdown path is never deleted, only demoted to fallback
- Data-dense DM admin outputs use Bot API 10.1+ native rich messages (`context.bot.send_rich_message` with `aiogram.types.InputRichMessage(html=...)`); currently `/trusted` (native `<table bordered striped>` with columns User | Added by | Date — the numeric ID is folded into the User cell as `name (@user)<br><code>id</code>` since a 4th column forced horizontal scrolling) and `/status` (per-group striped table Grup | Penegakan | Captcha | Probation | Pending | Nonaktif plus a key-value section for uptime/database/job timestamps). Same pattern as the mediadownloadbot metadata table
- Rich is the default with a **Markdown fallback**: any `send_rich_message` failure is caught (broad except, logged with `exc_info=True`) and the handler sends the pre-existing Markdown message instead — the Markdown path is never deleted, only demoted to fallback. `/check` deliberately stays on Markdown: aiogram 3.31 has no edit-to-rich equivalent, so its callback-query edit path could never go rich and a rich-then-Markdown flip mid-flow would be worse than consistent Markdown
- Scope is DM-only admin surface: group-facing messages stay on Markdown v1, so member-visible output never changes and old clients are unaffected where it matters
- Rich HTML content is `html.escape()`d (not the Markdown escaper); headings/column labels live in `constants.py` (`TRUST_LIST_RICH_HEADING`, `TRUST_LIST_RICH_COLUMNS`)
- Rich HTML content is `html.escape()`d (not the Markdown escaper); headings/column labels live in `constants.py` (`TRUST_LIST_RICH_*`, `STATUS_RICH_*`)
- All user-facing dates render in WIB (`constants.WIB = ZoneInfo("Asia/Jakarta")`), in both rich and Markdown paths

### Topic Guard Design
- Handles both `message` and `edited_message` updates (combined filter)
Expand Down
9 changes: 8 additions & 1 deletion src/bot/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,10 @@
"""

from aiogram.types import ChatPermissions
from zoneinfo import ZoneInfo

WIB = ZoneInfo("Asia/Jakarta")
"""Western Indonesian Time — display timezone for all user-facing dates."""

# Permissions applied when restricting a user (effectively mutes them)
RESTRICTED_PERMISSIONS = ChatPermissions(
Expand Down Expand Up @@ -294,7 +298,10 @@ def format_hours_display(hours: int) -> str:
# Rich-message (Bot API 10.1+) rendering of /trusted for DM admins.
# Same data as TRUST_LIST_HEADER, shown as a native striped table.
TRUST_LIST_RICH_HEADING = "📋 Trusted Users"
TRUST_LIST_RICH_COLUMNS = ("User", "User ID", "Added by", "Date")
TRUST_LIST_RICH_COLUMNS = ("User", "Added by", "Date")

STATUS_RICH_HEADING = "📊 Status Bot"
STATUS_RICH_COLUMNS = ("Grup", "Penegakan", "Captcha", "Probation", "Pending", "Nonaktif")

TRUST_DM_ONLY_MESSAGE = (
"❌ Perintah ini hanya bisa digunakan di chat pribadi dengan bot."
Expand Down
219 changes: 164 additions & 55 deletions src/bot/handlers/status.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,16 +10,20 @@

from __future__ import annotations

import html
import logging
import os
import time
from dataclasses import dataclass, field
from datetime import datetime

from aiogram.types import Update
from aiogram.types import InputRichMessage, Update

from bot.dispatch import AppState, HandlerContext, HandlerSpec, command_filter, effective_chat
from bot.services.markdown import escape_markdown

from bot.config import get_settings
from bot.constants import STATUS_RICH_COLUMNS, STATUS_RICH_HEADING, WIB
from bot.database.service import get_database
from bot.group_config import get_group_registry
from bot.services.telegram_utils import get_admin_groups
Expand Down Expand Up @@ -94,90 +98,195 @@ async def handle_status(update: Update, context: HandlerContext) -> None:
admin_user_id = update.message.from_user.id
admin_group_ids = set(get_admin_groups(context, admin_user_id))

lines: list[str] = []
data = _gather_status_data(context, admin_group_ids)
markdown_text = _render_status_markdown(data)
rich_html = _render_status_rich_html(data)

# Rich table is the default; any failure degrades to the Markdown message.
# Broad except is deliberate: the Markdown path must survive whatever
# broke the rich send (API errors, serialization, client issues).
try:
await context.bot.send_rich_message(
chat_id=update.message.chat.id,
rich_message=InputRichMessage(html=rich_html),
reply_parameters=update.message.as_reply_parameters(),
)
except Exception:
logger.warning(
"Rich status send failed, falling back to Markdown",
exc_info=True,
)
await update.message.reply(markdown_text, parse_mode="Markdown")


# --- Uptime ---
@dataclass
class _GroupStatusRow:
"""Per-group status data, scoped to the caller's admin groups."""

group_id: int
enforcement: str
captcha_enabled: bool
probation_count: int
pending_count: int
disabled: list[str] = field(default_factory=list)


@dataclass
class _StatusData:
"""Everything /status displays, gathered once for both renderers."""

uptime: str
rows: list[_GroupStatusRow] = field(default_factory=list)
db_size: str = "N/A"
refresh_display: str = "belum pernah"
restrict_display: str = "belum pernah"


def _gather_status_data(
context: HandlerContext, admin_group_ids: set[int]
) -> _StatusData:
"""Collect /status values once; Markdown and rich renderers share them."""
start = context.state.start_time
if start is not None:
uptime = _format_uptime(time.monotonic() - start)
lines.append(f"*Uptime:* {uptime}")
else:
lines.append("*Uptime:* N/A")
uptime = "N/A"

# --- Per-group summary (scoped to caller's admin groups) ---
lines.append("")
lines.append("*Grup yang kamu admin:*")
registry = get_group_registry()
effective_map = context.state.plugin_effective_map
db = get_database()

all_probations = db.get_all_new_user_probations()
all_pending = db.get_all_pending_captchas()

shown_groups = 0
rows: list[_GroupStatusRow] = []
for gc in registry.all_groups():
if gc.group_id not in admin_group_ids:
continue
shown_groups += 1
gid = gc.group_id
enforcement = "Restriksi" if gc.restrict_failed_users else "Peringatan"
captcha = "CAPTCHA" if gc.captcha_enabled else ""
group_parts = [
f" • `{escape_markdown(str(gid), version=1)}` — _{enforcement}_"
]
if captcha:
group_parts.append(f" _{captcha}_")

# Per-group counts
probation_count = sum(1 for p in all_probations if p.group_id == gid)
pending_count = sum(1 for p in all_pending if p.group_id == gid)
group_parts.append(
f"\n Probation: {probation_count}, Captcha: {pending_count}"
)

if gid not in admin_group_ids:
continue
toggles = effective_map.get(gid, {})
disabled = [k for k, v in toggles.items() if not v]
if disabled:
disabled_str = ", ".join(sorted(disabled))
group_parts.append(
f"\n Plugin nonaktif: {escape_markdown(disabled_str, version=1)}"
disabled = sorted(k for k, v in toggles.items() if not v)
rows.append(
_GroupStatusRow(
group_id=gid,
enforcement="Restriksi" if gc.restrict_failed_users else "Peringatan",
captcha_enabled=gc.captcha_enabled,
probation_count=sum(1 for p in all_probations if p.group_id == gid),
pending_count=sum(1 for p in all_pending if p.group_id == gid),
disabled=disabled,
)
lines.append("".join(group_parts))

if shown_groups == 0:
lines.append(" (Tidak ada grup yang dipantau)")
)

# --- Database size ---
lines.append("")
db_path = get_settings().database_path
size_str = _format_filesize(db_path)
lines.append(f"*Database:* {size_str}")
db_size = _format_filesize(get_settings().database_path)

# --- Last jobs ---
lines.append("")
lines.append("*Jadwal terakhir:*")
refresh_ts = context.state.data.get("last_admin_refresh")
if refresh_ts is not None:
lines.append(
" • Refresh admin: "
f"{time.strftime('%Y-%m-%d %H:%M:%S UTC', time.gmtime(refresh_ts))}"
refresh_display = datetime.fromtimestamp(refresh_ts, tz=WIB).strftime(
"%Y-%m-%d %H:%M:%S WIB"
)
else:
lines.append(" • Refresh admin: belum pernah")
refresh_display = "belum pernah"

restrict_ts = context.state.data.get("last_auto_restrict")
if restrict_ts is not None:
lines.append(
" • Auto-restrict: "
f"{time.strftime('%Y-%m-%d %H:%M:%S UTC', time.gmtime(restrict_ts))}"
restrict_display = datetime.fromtimestamp(restrict_ts, tz=WIB).strftime(
"%Y-%m-%d %H:%M:%S WIB"
)
else:
restrict_display = "belum pernah"

return _StatusData(
uptime=uptime,
rows=rows,
db_size=db_size,
refresh_display=refresh_display,
restrict_display=restrict_display,
)


def _render_status_markdown(data: _StatusData) -> str:
"""Render /status as the pre-existing Markdown message (fallback path).

Kept byte-identical to the original output; the rich renderer is the
default and this only runs when the rich send fails.
"""
lines: list[str] = []

lines.append(f"*Uptime:* {data.uptime}")

lines.append("")
lines.append("*Grup yang kamu admin:*")

if data.rows:
for row in data.rows:
group_parts = [
f" • `{escape_markdown(str(row.group_id), version=1)}`"
f" — _{row.enforcement}_"
]
if row.captcha_enabled:
group_parts.append(" _CAPTCHA_")
group_parts.append(
f"\n Probation: {row.probation_count}, Captcha: {row.pending_count}"
)
if row.disabled:
disabled_str = ", ".join(row.disabled)
group_parts.append(
f"\n Plugin nonaktif: {escape_markdown(disabled_str, version=1)}"
)
lines.append("".join(group_parts))
else:
lines.append(" (Tidak ada grup yang dipantau)")

lines.append("")
lines.append(f"*Database:* {data.db_size}")

lines.append("")
lines.append("*Jadwal terakhir:*")
lines.append(f" • Refresh admin: {data.refresh_display}")
lines.append(f" • Auto-restrict: {data.restrict_display}")

return "\n".join(lines)


def _render_status_rich_html(data: _StatusData) -> str:
"""Render /status as a native rich message (Bot API 10.1+).

Per-group striped table plus a small key-value section. All cell
content is html-escaped.
"""
parts = [f"<b>{html.escape(STATUS_RICH_HEADING)}</b>"]

if data.rows:
header = "".join(
f"<th>{html.escape(col)}</th>" for col in STATUS_RICH_COLUMNS
)
body = "".join(
"<tr>"
+ "".join(
f"<td>{html.escape(cell)}</td>"
for cell in (
str(row.group_id),
row.enforcement,
"Ya" if row.captcha_enabled else "—",
str(row.probation_count),
str(row.pending_count),
", ".join(row.disabled) if row.disabled else "—",
)
)
+ "</tr>"
for row in data.rows
)
parts.append(f"<table bordered striped><tr>{header}</tr>{body}</table>")
else:
lines.append(" • Auto-restrict: belum pernah")
parts.append(f"<i>{html.escape('Tidak ada grup yang dipantau.')}</i>")

await update.message.reply(
"\n".join(lines),
parse_mode="Markdown",
parts.append(
f"<b>Uptime:</b> {html.escape(data.uptime)}<br>"
f"<b>Database:</b> {html.escape(data.db_size)}<br>"
f"<b>Refresh admin:</b> {html.escape(data.refresh_display)}<br>"
f"<b>Auto-restrict:</b> {html.escape(data.restrict_display)}"
)
return "".join(parts)


def get_handlers(state: AppState) -> list[HandlerSpec]:
Expand Down
30 changes: 23 additions & 7 deletions src/bot/handlers/trust.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
TRUST_USER_ID_INVALID_MESSAGE,
TRUST_USER_ID_REQUIRED_MESSAGE,
TRUST_USER_NOT_FOUND_MESSAGE,
WIB,
)
from bot.database.models import TrustedUserData
from bot.database.service import DatabaseService, get_database
Expand Down Expand Up @@ -90,16 +91,32 @@ def _plain_person(full_name: str, username: str | None, user_id: int) -> str:
return display


def _trusted_list_rich_html(rows: list[tuple[str, str, str, str]]) -> str:
def _rich_user_cell(full_name: str, username: str | None, user_id: int) -> str:
"""Render the User cell: display line plus the numeric ID below it.

Returns HTML (already escaped). The standalone "User ID" column was
dropped — four columns forced horizontal scrolling — so the ID is
folded in here, wrapped in <code> to stay visible and copyable.
"""
return (
f"{html.escape(_plain_person(full_name, username, user_id))}"
f"<br><code>{user_id}</code>"
)


def _trusted_list_rich_html(rows: list[tuple[str, str, str]]) -> str:
"""Render trusted-user rows as a native rich table (Bot API 10.1+).

Follows the proven <table bordered striped> pattern: header row plus
one row per trusted user. All cell content is html-escaped.
one row per trusted user. ``rows`` are (user_cell_html, added_by, date);
the user cell is pre-rendered HTML (see _rich_user_cell); the other
cells are plain text and are html-escaped here.
"""
header = "".join(f"<th>{html.escape(col)}</th>" for col in TRUST_LIST_RICH_COLUMNS)
body = "".join(
"<tr>" + "".join(f"<td>{html.escape(cell)}</td>" for cell in row) + "</tr>"
for row in rows
f"<tr><td>{user_cell}</td><td>{html.escape(added_by)}</td>"
f"<td>{html.escape(date)}</td></tr>"
for user_cell, added_by, date in rows
)
return (
f"<b>{html.escape(TRUST_LIST_RICH_HEADING)}</b>"
Expand Down Expand Up @@ -305,7 +322,7 @@ async def handle_trusted_list_command(
trusted_at = record.trusted_at
if trusted_at.tzinfo is None:
trusted_at = trusted_at.replace(tzinfo=UTC)
trusted_at_display = trusted_at.astimezone(UTC).strftime("%Y-%m-%d %H:%M UTC")
trusted_at_display = trusted_at.astimezone(WIB).strftime("%Y-%m-%d %H:%M WIB")

user_display = _format_person_with_username(
record.user_full_name, record.username, record.user_id
Expand All @@ -320,8 +337,7 @@ async def handle_trusted_list_command(
)
rich_rows.append(
(
_plain_person(record.user_full_name, record.username, record.user_id),
str(record.user_id),
_rich_user_cell(record.user_full_name, record.username, record.user_id),
_plain_person(
record.admin_full_name,
record.admin_username,
Expand Down
5 changes: 4 additions & 1 deletion tests/test_properties.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@


import pytest
from hypothesis import HealthCheck, given, settings
from hypothesis import HealthCheck, assume, given, settings
from hypothesis import strategies as st

from bot.constants import format_hours_display, format_threshold_display
Expand Down Expand Up @@ -141,6 +141,9 @@ def test_empty_name_falls_back_to_user_id(self, name: str, uid: int) -> None:
@settings(max_examples=200)
def test_with_username_format(self, name: str, uid: int, username: str | None) -> None:
"""If username given, output contains ' (@<username>)' suffix when truthy."""
# The sentinel "(@" can also come from the name itself (the strategy
# allows punctuation), which says nothing about the username suffix.
assume("(@" not in name)
result = _format_person_with_username(name, username, uid)
if username:
assert f"(@{username})" in result
Expand Down
Loading
Loading