diff --git a/README.md b/README.md index 77fb902..b25bfa7 100644 --- a/README.md +++ b/README.md @@ -61,14 +61,30 @@ src/ ├── main.py # Entry point, handler registration ├── config.py # Environment & settings ├── handlers/ -│ ├── join_requests.py # Auto-approve/decline logic +│ ├── join_requests.py # Auto-approve/decline logic + audit logging │ ├── moderation.py # Spam detection & enforcement -│ └── admin_commands.py # /start, /stats, /approve_all +│ └── admin_commands.py # /start, /stats, /approve_all, /violations ├── services/ -│ └── spam_checker.py # Heuristic + AI spam assessment +│ ├── spam_checker.py # Heuristic + AI spam assessment +│ └── db.py # Supabase access (members, violations, join_requests) └── models/ # DB models (Phase 2) + +infra/ +└── schema.sql # Supabase tables & migrations ``` +### Database + +Moderation state (warnings, violations) and the join-request audit trail are +persisted in [Supabase](https://supabase.com). Apply the schema once: + +```bash +psql "$SUPABASE_DB_URL" -f infra/schema.sql # or paste into the Supabase SQL editor +``` + +Set `SUPABASE_URL` and `SUPABASE_KEY` in `.env`. If they are unset, the bot +still runs but moderation counts are kept in memory and reset on restart. + ## Deployment Designed for serverless (AWS Lambda + webhook) but runs in polling mode for development. See `infra/` for deployment configs (coming soon). diff --git a/infra/schema.sql b/infra/schema.sql new file mode 100644 index 0000000..8614b47 --- /dev/null +++ b/infra/schema.sql @@ -0,0 +1,39 @@ +-- Schema for TIHelperBot member tracking & moderation logs. +-- +-- Apply with the Supabase SQL editor, or: +-- psql "$SUPABASE_DB_URL" -f infra/schema.sql +-- +-- Tables are created idempotently so this file doubles as the migration. + +-- Members approved into the community. +create table if not exists members ( + user_id bigint primary key, + username text, + join_date timestamptz not null default now(), + approved_by text check (approved_by in ('bot', 'manual')), + spam_score real +); + +-- Moderation violations (warn / mute / ban events). +create table if not exists violations ( + id bigserial primary key, + user_id bigint not null, + message_text text, + violation_type text, + action_taken text, + timestamp timestamptz not null default now() +); + +create index if not exists violations_user_id_idx on violations (user_id); + +-- Audit log of join request decisions. +create table if not exists join_requests ( + id bigserial primary key, + user_id bigint not null, + assessment_score real, + approved boolean not null, + reason text, + timestamp timestamptz not null default now() +); + +create index if not exists join_requests_user_id_idx on join_requests (user_id); diff --git a/src/handlers/admin_commands.py b/src/handlers/admin_commands.py index d2bc037..38577df 100644 --- a/src/handlers/admin_commands.py +++ b/src/handlers/admin_commands.py @@ -1,6 +1,7 @@ from telegram import Update from telegram.ext import ContextTypes from services.help import render_help +from services import db _GROUP_CHAT_TYPES = ("group", "supergroup") @@ -41,6 +42,41 @@ async def stats(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: ) +async def violations(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + chat = update.effective_chat + member_status = "" + if chat.type in _GROUP_CHAT_TYPES: + member = await chat.get_member(update.effective_user.id) + member_status = member.status + + if not _is_admin(chat.type, member_status): + await update.message.reply_text("⛔ Admin only.") + return + + if not context.args: + await update.message.reply_text("Usage: /violations ") + return + + try: + target_id = int(context.args[0]) + except ValueError: + await update.message.reply_text("Please provide a numeric user_id.") + return + + records = await db.get_violations(target_id) + if not records: + await update.message.reply_text(f"No violations recorded for user {target_id}.") + return + + lines = [f"📋 Violations for user {target_id}:"] + for r in records: + lines.append( + f"• [{r.get('action_taken', '?')}] " + f"{r.get('violation_type', '?')} — {r.get('timestamp', '?')}" + ) + await update.message.reply_text("\n".join(lines)) + + async def approve_all(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: chat = update.effective_chat member = await chat.get_member(update.effective_user.id) diff --git a/src/handlers/join_requests.py b/src/handlers/join_requests.py index 019d7f4..dc6bca0 100644 --- a/src/handlers/join_requests.py +++ b/src/handlers/join_requests.py @@ -3,6 +3,7 @@ from telegram.ext import ContextTypes from services.spam_checker import assess_user from services.welcome import render_welcome +from services import db logger = logging.getLogger(__name__) @@ -13,8 +14,21 @@ async def handle_join_request(update: Update, context: ContextTypes.DEFAULT_TYPE assessment = await assess_user(user) + await db.record_join_request( + user_id=user.id, + assessment_score=assessment["score"], + approved=assessment["approved"], + reason=assessment["reason"], + ) + if assessment["approved"]: await request.approve() + await db.record_member( + user_id=user.id, + username=user.username, + approved_by="bot", + spam_score=assessment["score"], + ) logger.info(f"Auto-approved user {user.id} (@{user.username}) — score: {assessment['score']:.2f}") try: diff --git a/src/handlers/moderation.py b/src/handlers/moderation.py index 196a4a2..87788a0 100644 --- a/src/handlers/moderation.py +++ b/src/handlers/moderation.py @@ -2,13 +2,11 @@ from telegram import Update from telegram.ext import ContextTypes from services.spam_checker import assess_message +from services import db from config import MAX_WARNINGS logger = logging.getLogger(__name__) -# In-memory warning tracker (replace with DB in production) -_warnings: dict[int, int] = {} - async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: message = update.effective_message @@ -24,14 +22,29 @@ async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> return user_id = user.id - _warnings[user_id] = _warnings.get(user_id, 0) + 1 - count = _warnings[user_id] + count = (await db.get_warning_count(user_id)) + 1 if count > MAX_WARNINGS: + action = "ban" + elif count == MAX_WARNINGS: + action = "mute" + else: + action = "warn" + + # Record the violation before enforcing, so the warning count advances even + # if a Telegram API call below fails, and to keep the read/record window small. + await db.record_violation( + user_id=user_id, + message_text=message.text, + violation_type=assessment["reason"], + action_taken=action, + ) + + if action == "ban": await chat.ban_member(user_id) await message.delete() logger.info(f"Banned user {user_id} (@{user.username}) — exceeded warning limit") - elif count == MAX_WARNINGS: + elif action == "mute": await message.delete() await chat.restrict_member(user_id, permissions=_muted_permissions()) await message.reply_text( diff --git a/src/main.py b/src/main.py index 5b40aeb..bf9480b 100644 --- a/src/main.py +++ b/src/main.py @@ -10,7 +10,7 @@ from config import TELEGRAM_BOT_TOKEN from handlers.join_requests import handle_join_request from handlers.moderation import handle_message -from handlers.admin_commands import start, stats, approve_all, help_command +from handlers.admin_commands import start, stats, approve_all, help_command, violations logging.basicConfig( format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", @@ -26,6 +26,7 @@ def main() -> None: app.add_handler(CommandHandler("help", help_command)) app.add_handler(CommandHandler("stats", stats)) app.add_handler(CommandHandler("approve_all", approve_all)) + app.add_handler(CommandHandler("violations", violations)) app.add_handler(ChatJoinRequestHandler(handle_join_request)) diff --git a/src/services/db.py b/src/services/db.py new file mode 100644 index 0000000..5593458 --- /dev/null +++ b/src/services/db.py @@ -0,0 +1,176 @@ +"""Supabase data-access layer for member tracking and moderation logs. + +The bot must stay runnable without a database (local development, CI, and the +existing test suite all run without Supabase credentials). When Supabase is not +configured — or a query fails — these helpers fall back to an in-memory warning +counter so moderation still escalates within a single process. Persistence +across restarts only applies when Supabase is configured. + +Supabase's client is synchronous, so the blocking calls are dispatched to a +worker thread via ``asyncio.to_thread`` to avoid stalling the bot's event loop. +""" + +from __future__ import annotations + +import asyncio +import logging +from functools import lru_cache +from typing import Any + +from config import SUPABASE_KEY, SUPABASE_URL + +logger = logging.getLogger(__name__) + +# Per-process fallback warning counts, used only when Supabase is unavailable. +_warning_fallback: dict[int, int] = {} + + +@lru_cache(maxsize=1) +def get_client(): + """Return a cached Supabase client, or ``None`` when not configured.""" + if not SUPABASE_URL or not SUPABASE_KEY: + return None + try: + from supabase import create_client + + return create_client(SUPABASE_URL, SUPABASE_KEY) + except Exception as exc: # pragma: no cover - network/credential failures + logger.warning(f"Failed to initialize Supabase client: {exc}") + return None + + +def is_enabled() -> bool: + """Whether a Supabase connection is available.""" + return get_client() is not None + + +async def get_warning_count(user_id: int) -> int: + """Return how many violations have been recorded for ``user_id``.""" + client = get_client() + if client is None: + return _warning_fallback.get(user_id, 0) + + def _query(): + # count="exact" reports the full match count; limit(1) avoids pulling + # every matching row back just to read that count. + return ( + client.table("violations") + .select("id", count="exact") + .eq("user_id", user_id) + .limit(1) + .execute() + ) + + try: + resp = await asyncio.to_thread(_query) + return resp.count or 0 + except Exception as exc: + logger.warning(f"DB get_warning_count failed for {user_id}: {exc}") + return _warning_fallback.get(user_id, 0) + + +async def record_violation( + user_id: int, + message_text: str, + violation_type: str, + action_taken: str, +) -> None: + """Log a moderation violation (warn / mute / ban).""" + client = get_client() + if client is None: + _warning_fallback[user_id] = _warning_fallback.get(user_id, 0) + 1 + return + + def _insert(): + client.table("violations").insert( + { + "user_id": user_id, + "message_text": message_text, + "violation_type": violation_type, + "action_taken": action_taken, + } + ).execute() + + try: + await asyncio.to_thread(_insert) + except Exception as exc: + logger.warning(f"DB record_violation failed for {user_id}: {exc}") + _warning_fallback[user_id] = _warning_fallback.get(user_id, 0) + 1 + + +async def get_violations(user_id: int, limit: int = 50) -> list[dict[str, Any]]: + """Return recorded violations for ``user_id``, most recent first.""" + client = get_client() + if client is None: + return [] + + def _query(): + return ( + client.table("violations") + .select("*") + .eq("user_id", user_id) + .order("timestamp", desc=True) + .limit(limit) + .execute() + ) + + try: + resp = await asyncio.to_thread(_query) + return resp.data or [] + except Exception as exc: + logger.warning(f"DB get_violations failed for {user_id}: {exc}") + return [] + + +async def record_member( + user_id: int, + username: str | None, + approved_by: str, + spam_score: float, +) -> None: + """Upsert a member record after a join request is approved.""" + client = get_client() + if client is None: + return + + def _upsert(): + client.table("members").upsert( + { + "user_id": user_id, + "username": username, + "approved_by": approved_by, + "spam_score": spam_score, + } + ).execute() + + try: + await asyncio.to_thread(_upsert) + except Exception as exc: + logger.warning(f"DB record_member failed for {user_id}: {exc}") + + +async def record_join_request( + user_id: int, + assessment_score: float, + approved: bool, + reason: str, +) -> None: + """Log a join request decision for the audit trail.""" + client = get_client() + if client is None: + return + + def _insert(): + client.table("join_requests").insert( + { + "user_id": user_id, + "assessment_score": assessment_score, + "approved": approved, + "reason": reason, + } + ).execute() + + try: + await asyncio.to_thread(_insert) + except Exception as exc: + logger.warning(f"DB record_join_request failed for {user_id}: {exc}") diff --git a/tests/test_db.py b/tests/test_db.py new file mode 100644 index 0000000..54729ee --- /dev/null +++ b/tests/test_db.py @@ -0,0 +1,57 @@ +"""Tests for the db service fallback behavior (no Supabase configured). + +conftest.py runs without SUPABASE_URL / SUPABASE_KEY, so get_client() returns +None and the in-memory fallback path is exercised here. +""" + +import pytest + +from services import db + + +@pytest.fixture(autouse=True) +def reset_fallback(): + db._warning_fallback.clear() + db.get_client.cache_clear() + yield + db._warning_fallback.clear() + db.get_client.cache_clear() + + +def test_disabled_without_credentials(): + assert db.is_enabled() is False + assert db.get_client() is None + + +async def test_warning_count_starts_at_zero(): + assert await db.get_warning_count(123) == 0 + + +async def test_record_violation_increments_fallback_count(): + await db.record_violation(123, "buy crypto now", "matched: crypto", "warn") + assert await db.get_warning_count(123) == 1 + + await db.record_violation(123, "still spamming", "matched: crypto", "mute") + assert await db.get_warning_count(123) == 2 + + +async def test_warning_counts_are_per_user(): + await db.record_violation(1, "spam", "promo", "warn") + await db.record_violation(2, "spam", "promo", "warn") + await db.record_violation(2, "spam", "promo", "mute") + + assert await db.get_warning_count(1) == 1 + assert await db.get_warning_count(2) == 2 + assert await db.get_warning_count(999) == 0 + + +async def test_get_violations_empty_without_db(): + await db.record_violation(5, "spam", "promo", "warn") + # Violation history requires Supabase; fallback keeps counts only. + assert await db.get_violations(5) == [] + + +async def test_record_member_and_join_request_are_noops_without_db(): + # Should not raise when Supabase is unavailable. + await db.record_member(7, "user7", "bot", 0.1) + await db.record_join_request(7, 0.1, True, "passed checks") diff --git a/tests/test_spam_checker.py b/tests/test_spam_checker.py index 64c7e57..82efe4c 100644 --- a/tests/test_spam_checker.py +++ b/tests/test_spam_checker.py @@ -54,7 +54,6 @@ async def test_normal_user_is_approved(): result = await assess_user(user) assert result["approved"] is True assert result["score"] == 0.0 - assert "welcome_message" in result async def test_no_username_and_short_name_cumulative_score(): @@ -65,13 +64,6 @@ async def test_no_username_and_short_name_cumulative_score(): assert result["approved"] is True -async def test_welcome_message_contains_user_name(): - user = make_user(username="devguy", first_name="David") - result = await assess_user(user) - assert result["approved"] is True - assert "David" in result["welcome_message"] - - async def test_no_welcome_message_when_rejected(): user = make_user(is_bot=True) result = await assess_user(user)