diff --git a/AGENTS.md b/AGENTS.md
index 53202e5..420ace9 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -68,7 +68,8 @@ PythonID/
│ │ ├── dm.py # DM unrestriction flow
│ │ ├── topic_guard.py # Warning topic protection (group=-1)
│ │ ├── trust.py # /trust, /untrust, /trusted admin commands
-│ │ └── duplicate_spam.py # Duplicate message detection
+│ │ ├── duplicate_spam.py # Duplicate message detection
+│ │ └── bio_bait.py # Bio-bait spam (bait phrases + suspicious profile bio links)
│ ├── services/
│ │ ├── user_checker.py # Profile validation (photo + username)
│ │ ├── scheduler.py # JobQueue auto-restriction (every 5 min)
@@ -90,7 +91,7 @@ PythonID/
| Task | Location | Notes |
|------|----------|-------|
-| Add new handler | `main.py` | Register with appropriate group (-1, 0, 1-5) |
+| Add new handler | `plugins/definitions.py` + `plugins/builtin/X.py` | Add to `MANIFEST_ORDER` with a `handler_group` (-1, 0-6); handlers register via `PluginManager`, not directly in `main.py` |
| Modify messages | `constants.py` | All Indonesian templates centralized |
| Add DB table | `database/models.py` → `database/service.py` | Add model, then service methods |
| Change config | `config.py` | Pydantic BaseSettings with env vars |
@@ -102,44 +103,65 @@ PythonID/
| File | Lines | Role |
|------|-------|------|
-| `group_config.py` | 250 | Multi-group config, registry, JSON loading, .env fallback |
-| `database/service.py` | 671 | **Complexity hotspot** - handles warnings, captcha, probation state |
-| `constants.py` | 530 | Templates + massive whitelists (Indonesian tech community) |
-| `handlers/captcha.py` | 375 | New member join → restrict → verify (with profile check) → unrestrict lifecycle |
-| `handlers/verify.py` | 358 | Admin verification commands + inline button callbacks |
-| `handlers/anti_spam.py` | 420 | Anti-spam: contact cards, inline keyboards, probation enforcement |
-| `handlers/trust.py` | 350 | /trust, /untrust, /trusted admin commands + cache |
-| `main.py` | 315 | Entry point, logging, handler registration, JobQueue setup |
-| `plugins/manager.py` | 250 | PluginManager — discovers + registers all built-in plugins |
-| `plugins/builtin/captcha.py` | 50 | Wraps captcha handler + applies guard_plugin gating |
+| `database/service.py` | 813 | **Complexity hotspot** - handles warnings, captcha, probation state |
+| `constants.py` | 666 | Templates + massive whitelists (Indonesian tech community) |
+| `handlers/anti_spam.py` | 494 | Anti-spam: contact cards, inline keyboards, probation enforcement |
+| `handlers/bio_bait.py` | 441 | Bio-bait spam: obfuscated bait phrases + suspicious profile bio links |
+| `handlers/captcha.py` | 412 | New member join → restrict → verify (with profile check) → unrestrict lifecycle |
+| `handlers/trust.py` | 381 | /trust, /untrust, /trusted admin commands + cache |
+| `handlers/verify.py` | 320 | Admin verification commands + inline button callbacks |
+| `group_config.py` | 255 | Multi-group config, registry, JSON loading, .env fallback |
+| `handlers/duplicate_spam.py` | 216 | Duplicate message detection |
+| `main.py` | 191 | Entry point, logging, post_init, PluginManager bootstrap |
+| `plugins/manager.py` | 188 | PluginManager — static registry + deterministic registration order |
+| `plugins/config.py` | 156 | `guard_plugin` runtime gate + toggle resolution |
+| `plugins/definitions.py` | 69 | `MANIFEST_ORDER` / `PLUGIN_NAMES` — single source of truth for plugin names + groups |
+| `plugins/builtin/spam.py` | 93 | Wraps all 5 anti-spam handlers with `guard_plugin` |
+| `plugins/builtin/captcha.py` | 43 | Wraps captcha handler + applies guard_plugin gating |
## Architecture Patterns
### Modular Plugin System
- Built-in plugins live in `src/bot/plugins/builtin/`, one per handler domain (captcha, spam, topic_guard, profile_monitor, commands, dm, jobs)
-- Each plugin's `register(application)` is called by `PluginManager` in `main.py:post_init`
+- `plugins/definitions.py` holds `MANIFEST_ORDER` — a static, hand-maintained tuple of 23 plugin names (topic_guard first, job plugins last) that is the single source of truth for registration order and for the group number each plugin runs in
+- `PluginManager.register_all()` (called from `main.py:main`, not `post_init`) walks `MANIFEST_ORDER` against a static `_REGISTRY` dict (name → registrar function) and stores results in `application.bot_data["plugin_handlers"]`
- The plugin wrapper pattern: `bot.plugins.builtin.X` imports from `bot.handlers.X`, clones the handler list, and applies `guard_plugin("X")` for per-group runtime gating
-- To add a new plugin: create a class with `name` + `register(application)` in `builtin/`, register it in `manager.py`
+- To add a new plugin: add a `register_*(application) -> list[BaseHandler]` function in `builtin/`, add its name + group to `_PLUGIN_DEFINITIONS` in `definitions.py`, wire it into `_REGISTRY` in `manager.py`
- Handler modules stay decoupled from plugin internals — changes to `bot/handlers/X.py` flow through transparently
### Handler Priority Groups
```python
-# main.py - Order matters!
+# Registration order comes from MANIFEST_ORDER (plugins/definitions.py), not main.py directly
group=-1 # topic_guard: Runs FIRST
-group=0 # Commands, DM, captcha
+group=0 # commands, verify/unverify/check/trust callbacks, captcha, dm (14 plugins, order-independent)
group=1 # inline_keyboard_spam: Catches inline keyboard URL spam
group=2 # contact_spam: Blocks contact card sharing
group=3 # new_user_spam: Probation enforcement (links/forwards)
-group=4 # duplicate_spam: Repeated message detection
-group=5 # message_handler: Runs LAST, profile compliance check
+group=4 # duplicate_spam + bio_bait_spam: Repeated messages / bio-bait detection
+group=5 # profile_monitor: Runs LAST, profile compliance check
+group=6 # JobQueue only (not a handler group): auto_restrict_job, refresh_admin_ids_job
```
+### Plugin Gating (`guard_plugin`)
+- `guard_plugin("name")` (`plugins/config.py`) wraps a handler callback; on each update it looks up `context.bot_data["plugin_effective_map"][group_id]["name"]` and no-ops the callback if disabled
+- Only gates **group/supergroup** chats — private and channel updates always pass through unchanged
+- Fail-open by design: unknown group, missing plugin key, or missing effective map all resolve to enabled
+- Per-group toggle resolution (`resolve_plugin_toggles`), first match wins: (1) `GroupConfig.plugins` override in groups.json, (2) `Settings.plugins_default` (`PLUGINS_DEFAULT` env, JSON object), (3) `True`
+- `PluginManager.compute_effective_map()` runs once at startup (after `register_all`) and caches the resolved per-group map in `bot_data["plugin_effective_map"]` — it is not recomputed per-update
+- There is no bypass mechanism for admin commands beyond simply not wrapping them in `guard_plugin`
+
### Captcha Profile Check
- New members must have a public profile photo AND username before captcha verification completes
- `check_user_profile()` in `services/user_checker.py` queries both via Bot API
- Profile-incomplete path: alert shown, captcha record preserved, timeout still armed, user can fix profile and retry
- DB finalization (remove_pending_captcha + start_new_user_probation) runs **before** `unrestrict_user` — the irreversible Telegram side effect goes last, so a failure leaves the user still restricted + DB consistent
+### Bio Bait Detection
+- `handlers/bio_bait.py` catches two vectors: (1) a bait phrase in the message itself ("cek bio aku"), (2) the sender's Telegram **profile bio** containing a promo/invite link — bio is fetched via `get_chat` and cached per-user for 1 hour (5 min on fetch failure) to avoid hammering the API
+- Bait-phrase matching normalizes obfuscation first: NFKC + lowercase, strips zero-width characters, canonicalizes obfuscated "bio"/"byo" spellings (leetspeak, separators, Cyrillic look-alikes) before running the bait regexes
+- `bio_bait_monitor_only` (per-group) skips delete/restrict and only logs + optionally alerts `bio_bait_alert_chat_id` — use this to tune detection before enforcing
+- Admins and trusted users are exempt (`is_user_admin_or_trusted`)
+
### Topic Guard Design
- Handles both `message` and `edited_message` updates (combined filter)
- Raises `ApplicationHandlerStop` after handling ANY warning-topic message (allows or deletes)
@@ -153,15 +175,17 @@ group=5 # message_handler: Runs LAST, profile compliance check
- `BotInfoCache` — Class-level cache for bot username/ID
### Admin Cache
-- Fetched at startup in `post_init()` and stored in `bot_data["group_admin_ids"]` (per-group) and `bot_data["admin_ids"]` (union)
-- Refreshed every 10 minutes via `refresh_admin_ids` JobQueue job
+- `post_init()` order: `preload_admin_ids()` (fail-soft per group) → load all `TrustedUser` IDs into `bot_data["trusted_user_ids"]` → conditionally `recover_pending_captchas()` if any group has `captcha_enabled`
+- Admin IDs stored in `bot_data["group_admin_ids"]` (per-group) and `bot_data["admin_ids"]` (union)
+- Refreshed every 10 minutes via `refresh_admin_ids_job` JobQueue job
- On refresh failure for a group, falls back to existing cached data (not empty list)
- Spam handlers use cached admin IDs; topic_guard uses live `get_chat_member` API call
+- Handler + JobQueue registration (`PluginManager.register_all()`) and effective-plugin-map computation happen later, in `main()` after `post_init` is wired up but before `run_polling` — not inside `post_init` itself
### Multi-Group Support
-- `GroupConfig` — Pydantic model for per-group settings (warning thresholds, captcha, probation)
+- `GroupConfig` — Pydantic model with 20 per-group settings: warning thresholds, captcha, probation, contact/duplicate/bio-bait spam tuning, `rules_link`, and a `plugins: dict[str, bool] | None` override
- `GroupRegistry` — O(1) lookup by group_id, manages all monitored groups
-- `groups.json` — Per-group config file; falls back to `.env` for single-group mode
+- `groups.json` — Per-group config file; falls back to `.env` for single-group mode (missing fields default from `GroupConfig.model_fields`)
- `get_group_config_for_update()` — Helper to resolve config for incoming Telegram updates
- Exception-isolated loops — Per-group API calls wrapped in try/except to prevent cross-group failures
@@ -174,11 +198,12 @@ Time threshold → Auto-restrict via scheduler (parallel path)
```
### Database Conventions
-- SQLite with **WAL mode** for concurrency
+- SQLite with **WAL mode + `synchronous=NORMAL`** for write concurrency under a single-process bot
- `session.exec(select(Model).where(...)).first()` syntax
-- Atomic updates for violation counts (prevents race conditions)
-- No Alembic — use `SQLModel.metadata.create_all` + `_migrate_trusted_users` for column adds
-- New tables: `TrustedUser` (5th table) for the /trust admin bypass feature
+- Atomic updates for violation counts via raw `UPDATE ... SET x = x + 1` (prevents read-modify-write races)
+- No Alembic — use `SQLModel.metadata.create_all` + `_migrate_trusted_users` (`ALTER TABLE`) for column adds
+- Registers a datetime SQLite adapter to isoformat strings (avoids the Python 3.12+ default-adapter deprecation)
+- New tables: `TrustedUser` (5th table) for the /trust admin bypass feature — `group_id` defaults to `0` (global scope); per-group trust is modeled in the schema but not currently exercised anywhere
## Code Style
@@ -251,11 +276,14 @@ if user.id not in admin_ids:
| Add a new built-in plugin | `src/bot/plugins/builtin/X.py` (class with `name` + `register(application)`) |
| Register an existing handler as a plugin | Wrap `bot.handlers.X.get_handlers()` in `bot/plugins/builtin/X.py` |
| Add per-group runtime gating | `guard_plugin("X")` decorator in `src/bot/plugins/config.py` |
-| Disable a plugin for one group | Set `enabled_plugins: list[str]` in that group's config (groups.json) |
-| Bypass per-group gating for a single call | `guard_plugin.bypass_for(group_id)` context manager |
+| Disable a plugin for one group | Set `plugins: {"name": false}` in that group's entry in groups.json (`GroupConfig.plugins`) |
+| Set a bot-wide plugin default | `PLUGINS_DEFAULT` env var (JSON object, `Settings.plugins_default`) |
+| Exempt a handler from gating entirely | Don't wrap it in `guard_plugin(...)` (this is how admin commands stay ungated) |
## Notes
+- Registration order for all 23 built-in plugins lives in `MANIFEST_ORDER` (`plugins/definitions.py`), not scattered across `main.py`
+- `duplicate_spam` and `bio_bait_spam` both run at `group=4`; `auto_restrict_job` / `refresh_admin_ids_job` run as JobQueue jobs tagged `group=6` (not a PTB handler group)
- Topic guard runs at `group=-1` to intercept unauthorized messages BEFORE other handlers
- Topic guard handles both messages and edited messages, raises `ApplicationHandlerStop` to block downstream handlers
- JobQueue auto-restriction job runs every 5 minutes (first run after 5 min delay)
diff --git a/README.md b/README.md
index 01e1c33..a537ac5 100644
--- a/README.md
+++ b/README.md
@@ -23,6 +23,8 @@ A comprehensive Telegram bot for managing group members with profile verificatio
- **Captcha timeout recovery**: Automatically recovers pending verifications after bot restart
- **New user probation**: New members restricted from sending links/forwarded messages for 3 days (configurable)
- **Contact card blocking**: Prevents all non-admin members from sharing contact cards/phone numbers (delete + restrict)
+- **Duplicate message detection**: Flags repeated near-identical messages within a configurable window
+- **Bio-bait detection**: Catches obfuscated "check my bio" bait phrases and suspicious promo links in a sender's Telegram profile bio (monitor-only mode available)
- **Anti-spam enforcement**: Tracks violations and restricts spammers after threshold
- **Trusted users**: Admin-managed trusted list to bypass anti-spam + duplicate-spam checks
@@ -146,7 +148,7 @@ Add `GROUPS_CONFIG_PATH=groups.json` to your `.env` file, then edit `groups.json
]
```
-When `groups.json` is present, per-group settings override the `.env` defaults. Each group can have its own warning thresholds, captcha settings, probation rules, and rules link.
+When `groups.json` is present, per-group settings override the `.env` defaults. Each group can have its own warning thresholds, captcha settings, probation rules, and rules link. Each group entry can also add a `"plugins": {"bio_bait_spam": false}`-style object to disable specific built-in plugins just for that group, overriding the bot-wide `PLUGINS_DEFAULT`.
**Backward compatibility**: If no `groups.json` is configured (i.e., `GROUPS_CONFIG_PATH` is not set), the bot falls back to single-group mode using `GROUP_ID`, `WARNING_TOPIC_ID`, and other settings from `.env`.
@@ -287,7 +289,8 @@ PythonID/
│ ├── topic_guard.py # Warning topic protection
│ ├── trust.py # /trust, /untrust, /trusted admin commands
│ ├── verify.py # /verify and /unverify command handlers
- │ └── duplicate_spam.py # Duplicate message detection
+ │ ├── duplicate_spam.py # Duplicate message detection
+ │ └── bio_bait.py # Bio-bait spam (bait phrases + suspicious profile bio links)
├── database/
│ ├── models.py # SQLModel schemas (5 tables)
│ └── service.py # Database operations
@@ -306,175 +309,259 @@ The following diagram illustrates the complete bot workflow including captcha ve
```mermaid
flowchart TD
- Start([Bot Starts]) --> Init[Initialize Database & Config]
- Init --> FetchAdmins[Fetch Group Admin IDs]
- FetchAdmins --> RecoverCaptcha{Captcha
Enabled?}
+ Start([Bot Starts]) --> InitDB[Init Database + Group Registry
from groups.json or .env]
+ InitDB --> RegisterHandlers[PluginManager.register_all:
register handlers + jobs
in MANIFEST_ORDER]
+ RegisterHandlers --> ComputeMap[Compute Per-Group
Effective Plugin Map]
+ ComputeMap --> RunPolling[run_polling starts]
+ RunPolling --> FetchAdmins[post_init: Fetch
Group Admin IDs]
+ FetchAdmins --> LoadTrusted[(Load Trusted User IDs
into bot_data cache)]
+ LoadTrusted --> RecoverCaptcha{Any Group Has
Captcha Enabled?}
RecoverCaptcha -->|Yes| RecoverPending[Recover Pending Captchas]
- RecoverCaptcha -->|No| StartJobs
- RecoverPending --> StartJobs[Start JobQueue Scheduler
5-minute interval]
- StartJobs --> Poll[Poll for Updates]
-
+ RecoverCaptcha -->|No| Poll[Poll for Updates]
+ RecoverPending --> Poll
+
Poll --> UpdateType{Update Type?}
-
- %% New Member Flow
- UpdateType -->|New Member| CheckCaptchaEnabled{Captcha
Enabled?}
- CheckCaptchaEnabled -->|No| StartProbation[Start Probation Only]
- CheckCaptchaEnabled -->|Yes| RestrictAndChallenge[Restrict & Send Captcha]
- RestrictAndChallenge --> StorePending[(Store Pending Validation)]
+
+ %% ===================== New Member Flow (captcha.py, group=0) =====================
+ UpdateType -->|New Member Joins| EntryPoint{Entry Point}
+ EntryPoint -->|ChatMemberHandler
works with Hide Join| MaybeStart[_maybe_start_captcha]
+ EntryPoint -->|MessageHandler
NEW_CHAT_MEMBERS fallback| MaybeStart
+ MaybeStart --> StartProbationInit[(Start New User Probation
unconditional)]
+ StartProbationInit --> CheckCaptchaEnabled{Captcha Enabled
for this Group?}
+ CheckCaptchaEnabled -->|No| EndNM1([Done - probation only,
not restricted])
+ CheckCaptchaEnabled -->|Yes| RestrictAndChallenge[Restrict Member]
+ RestrictAndChallenge --> SendChallenge[Send Verify Button
captcha_verify_group_user]
+ SendChallenge --> StorePending[(Store Pending
Captcha Validation)]
StorePending --> ScheduleTimeout[Schedule Timeout Job]
- ScheduleTimeout --> WaitCaptcha[Wait for Verification]
-
- WaitCaptcha --> CaptchaAnswer{User
Action?}
- CaptchaAnswer -->|Correct Button| CancelTimeout[Cancel Timeout Job]
- CancelTimeout --> UnrestrictMember[Unrestrict Member]
- UnrestrictMember --> StartProbationAfter[Start Probation]
- CaptchaAnswer -->|Wrong User| ShowError[Show Error Message]
- CaptchaAnswer -->|Timeout| KickMember[Keep Restricted]
- KickMember --> UpdateMessage[Update Challenge Message]
-
- %% Anti-Spam Flow (Contact Card + New User Probation)
- UpdateType -->|Group Message| CheckContact{Has Contact
Card?}
- CheckContact -->|Yes| CheckContactAdmin{Is Admin?}
- CheckContactAdmin -->|Yes| CheckProbation
- CheckContactAdmin -->|No| DeleteContact[Delete Contact Message]
- DeleteContact --> RestrictContact[Restrict User]
- RestrictContact --> SendContactNotify[Send Contact
Spam Notification]
- CheckContact -->|No| CheckProbation
- CheckProbation{User On
Probation?} -->|No| CheckBot
- CheckProbation -->|Yes| CheckExpired{Probation
Expired?}
- CheckExpired -->|Yes| ClearProbation[(Clear Probation)]
- CheckExpired -->|No| CheckViolation{Forward/Link/
External Reply?}
-
- CheckViolation -->|No| End1([Continue])
- CheckViolation -->|Yes| CheckWhitelisted{URL
Whitelisted?}
- CheckWhitelisted -->|Yes| End1
- CheckWhitelisted -->|No| DeleteSpam[Delete Message]
- DeleteSpam --> IncrementViolation[(Increment Violation)]
- IncrementViolation --> ViolationCount{Violation
Count?}
-
- ViolationCount -->|First| SendSpamWarning[Send Probation Warning]
- ViolationCount -->|< Threshold| End2([Done])
- ViolationCount -->|>= Threshold| RestrictSpammer[Restrict User]
- RestrictSpammer --> SendSpamRestriction[Send Restriction Notice]
-
- %% Group Message Flow - Topic Guard
- CheckBot{From Bot?}
- CheckBot -->|Yes| End3([Ignore])
- CheckBot -->|No| TopicGuard{In Warning
Topic?}
- TopicGuard -->|Yes| IsAdmin{Is Admin
or Bot?}
- IsAdmin -->|No| DeleteMsg[Delete Message]
- IsAdmin -->|Yes| End4([Allow])
-
- %% Group Message Flow - Profile Check
- TopicGuard -->|No| CheckWhitelist{User
Whitelisted?}
- CheckWhitelist -->|Yes| End5([Allow])
- CheckWhitelist -->|No| CheckProfile[Check User Profile:
Photo + Username]
-
- CheckProfile --> ProfileComplete{Profile
Complete?}
- ProfileComplete -->|Yes| End6([Allow])
- ProfileComplete -->|No| CheckMode{Restriction
Mode?}
-
- %% Warning Mode
- CheckMode -->|Warning Only| SendWarning[Send Warning to Topic
Time threshold mentioned]
- SendWarning --> End7([Done])
-
- %% Progressive Restriction Mode
- CheckMode -->|Progressive| CheckCount{Message
Count?}
- CheckCount -->|First Message| SendFirstWarning[Send Warning with
Message & Time Thresholds]
- SendFirstWarning --> IncrementDB[(Store Warning in DB
with timestamp)]
- IncrementDB --> End8([Done])
-
- CheckCount -->|2 to N-1| SilentIncrement[(Silent: Increment Count)]
- SilentIncrement --> End9([Done])
-
- CheckCount -->|>= Threshold| RestrictUser[Apply Restriction
Mute Permissions]
- RestrictUser --> MarkRestricted[(Mark as Restricted
in Database)]
- MarkRestricted --> SendRestrictionMsg[Send Restriction Notice
with DM Link]
- SendRestrictionMsg --> End10([Done])
-
- %% DM Flow
- UpdateType -->|Private Message| CheckInGroup{User in
Group?}
- CheckInGroup -->|No| SendNotInGroup[Send: Not in Group]
- CheckInGroup -->|Yes| CheckPendingCaptcha{Has Pending
Captcha?}
-
- CheckPendingCaptcha -->|Yes| SendCaptchaRedirect[Send: Complete Captcha
in Group First]
- CheckPendingCaptcha -->|No| CheckDMProfile[Check Profile]
-
- CheckDMProfile --> DMProfileComplete{Profile
Complete?}
- DMProfileComplete -->|No| SendMissing[Send: Missing Items]
- DMProfileComplete -->|Yes| CheckBotRestricted{Restricted
by Bot?}
-
- CheckBotRestricted -->|No| SendNoRestriction[Send: No Bot Restriction]
- CheckBotRestricted -->|Yes| CheckCurrentStatus{Currently
Restricted?}
-
- CheckCurrentStatus -->|No| ClearRecord[(Clear Database Record)]
- ClearRecord --> SendAlreadyUnrestricted[Send: Already Unrestricted]
-
- CheckCurrentStatus -->|Yes| UnrestrictUser[Remove Restriction]
- UnrestrictUser --> ClearRecord2[(Clear Database Record)]
- ClearRecord2 --> SendSuccess[Send: Success Message]
-
- %% Scheduler Job (Background)
- StartJobs -.->|Every 5 min| SchedulerJob[Auto-Restriction Job]
- SchedulerJob --> QueryDB[(Query Warnings Past
Time Threshold)]
+ ScheduleTimeout --> WaitCaptcha[Wait for Callback
or Timeout]
+
+ WaitCaptcha --> CaptchaAnswer{Which Fires
First?}
+ CaptchaAnswer -->|Wrong user tapped button| ShowError[Alert: Not Your Button]
+ ShowError --> WaitCaptcha
+ CaptchaAnswer -->|Timeout job fires| CaptchaTimeout[captcha_timeout_callback]
+ CaptchaTimeout --> HandleExpiration[handle_captcha_expiration
captcha_recovery.py]
+ HandleExpiration --> EndNM2([User stays restricted
pending row cleared])
+
+ CaptchaAnswer -->|Correct user tapped| CheckProfileCaptcha[check_user_profile:
photo + username]
+ CheckProfileCaptcha --> ProfileOkCaptcha{Profile
Complete?}
+ ProfileOkCaptcha -->|No| ShowIncomplete[Alert: Missing Items
pending row + timeout kept armed]
+ ShowIncomplete --> WaitCaptcha
+ ProfileOkCaptcha -->|Yes| RemovePending[(Remove Pending Captcha)]
+ RemovePending --> RestartProbation[(Restart Probation Clock)]
+ RestartProbation --> UnrestrictMember[Unrestrict Member
irreversible call, done last]
+ UnrestrictMember --> CancelTimeout[Cancel Timeout Job]
+ CancelTimeout --> UpdateMessage[Edit Message: Verified]
+ UpdateMessage --> EndNM3([Done])
+
+ %% ===================== Group Message Pipeline (real PTB group order) =====================
+ UpdateType -->|Group Message| G_TopicGuard{group=-1 topic_guard:
In Warning Topic?}
+ G_TopicGuard -->|No, or lookup error| G_InlineGate
+ G_TopicGuard -->|Yes, incl. API error
fail-closed| G_TopicIsBotAdmin{Sender is
Bot or Admin?}
+ G_TopicIsBotAdmin -->|Yes| StopTopic1([ApplicationHandlerStop
message allowed])
+ G_TopicIsBotAdmin -->|No| G_TopicDelete[Delete Message]
+ G_TopicDelete --> StopTopic2([ApplicationHandlerStop])
+
+ G_InlineGate{group=1 inline_keyboard_spam:
Bot or Admin/Trusted?}
+ G_InlineGate -->|Yes| G_ContactGate
+ G_InlineGate -->|No| G_InlineCheck{Inline Button URL
Not Whitelisted?}
+ G_InlineCheck -->|No match| G_ContactGate
+ G_InlineCheck -->|Yes| G_InlineDelete[Delete Message]
+ G_InlineDelete --> G_InlineRestrict[Restrict User
always, no config gate]
+ G_InlineRestrict --> G_InlineNotify[Notify Warning Topic]
+ G_InlineNotify --> StopInline([ApplicationHandlerStop])
+
+ G_ContactGate{group=2 contact_spam:
Bot or Admin/Trusted?}
+ G_ContactGate -->|Yes| G_NewUserGate
+ G_ContactGate -->|No| G_ContactCheck{Message Has
Contact Card?}
+ G_ContactCheck -->|No| G_NewUserGate
+ G_ContactCheck -->|Yes| G_ContactDelete[Delete Message]
+ G_ContactDelete --> G_ContactRestrictGate{contact_spam_restrict
Enabled for Group?}
+ G_ContactRestrictGate -->|Yes| G_ContactRestrict[Restrict User]
+ G_ContactRestrict --> G_ContactNotify[Notify Warning Topic]
+ G_ContactRestrictGate -->|No, delete only| G_ContactNotify
+ G_ContactNotify --> StopContact([ApplicationHandlerStop])
+
+ G_NewUserGate{group=3 new_user_spam:
Bot or Admin/Trusted?}
+ G_NewUserGate -->|Yes| G_Group4Note
+ G_NewUserGate -->|No| G_OnProbation{User On
Probation?}
+ G_OnProbation -->|No| G_Group4Note
+ G_OnProbation -->|Yes| G_ProbationExpired{Probation Window
Elapsed?}
+ G_ProbationExpired -->|Yes| G_ClearProbation[(Clear Probation Record)]
+ G_ClearProbation --> G_Group4Note
+ G_ProbationExpired -->|No| G_Violation{Forwarded, non-whitelisted
link, external reply,
story, or media?}
+ G_Violation -->|No| G_Group4Note
+ G_Violation -->|Yes| G_ViolationDelete[Delete Message]
+ G_ViolationDelete --> G_ViolationIncrement[(Increment Violation Count)]
+ G_ViolationIncrement --> G_ViolationCount{Violation
Count?}
+ G_ViolationCount -->|1st| G_ViolationWarn[Send Probation Warning]
+ G_ViolationWarn --> StopNewUser1([ApplicationHandlerStop])
+ G_ViolationCount -->|2 to N-1| StopNewUser2([ApplicationHandlerStop
silent, delete only])
+ G_ViolationCount -->|">= new_user_violation_threshold"| G_ViolationRestrict[Restrict User]
+ G_ViolationRestrict --> G_ViolationNotify[Send Restriction Notice]
+ G_ViolationNotify --> StopNewUser3([ApplicationHandlerStop])
+
+ G_Group4Note[/"group=4: two handlers share the SAME filter
(GROUPS & not COMMAND). duplicate_spam registers
first in MANIFEST_ORDER and PTB block defaults to
True, so it always claims the update -
bio_bait_spam's handler never runs in practice."/]
+ G_Group4Note --> G_DupGate{duplicate_spam:
Enabled, Bot,
or Admin/Trusted?}
+ G_DupGate -->|Disabled, bot,
or admin/trusted| G_ProfileGate
+ G_DupGate -->|No| G_DupLength{Message Long Enough?
duplicate_spam_min_length}
+ G_DupLength -->|No| G_ProfileGate
+ G_DupLength -->|Yes| G_DupSimilar{Similar Messages in Window
>= threshold-1?
SequenceMatcher.ratio >= similarity}
+ G_DupSimilar -->|No| G_ProfileGate
+ G_DupSimilar -->|Yes| G_DupDelete[Delete Matching Messages
in Window]
+ G_DupDelete --> G_DupRestrict[Restrict User]
+ G_DupRestrict --> G_DupNotify[Notify Warning Topic]
+ G_DupNotify --> StopDup([ApplicationHandlerStop])
+
+ G_ProfileGate{group=5 profile_monitor:
Sender is Bot?
no admin bypass here}
+ G_ProfileGate -->|Yes| EndG0([Ignore])
+ G_ProfileGate -->|No| G_Whitelist{User in Photo
Verification Whitelist?}
+ G_Whitelist -->|Yes| EndG1([Allow])
+ G_Whitelist -->|No| G_CheckProfile[check_user_profile:
photo + username]
+ G_CheckProfile --> G_ProfileComplete{Profile
Complete?}
+ G_ProfileComplete -->|Yes| EndG2([Allow])
+ G_ProfileComplete -->|No| G_Mode{restrict_failed_users
for this Group?}
+
+ G_Mode -->|False: Warning Mode| G_WarnOnly[Send Warning to Topic
time threshold mentioned, no tracking]
+ G_WarnOnly --> EndG3([Done])
+
+ G_Mode -->|True: Progressive Mode| G_MsgCount{Message Count
for this User?}
+ G_MsgCount -->|1st message| G_FirstWarn[Send Warning with
Message + Time Thresholds]
+ G_FirstWarn --> G_StoreWarn[(Store Warning in DB
with timestamp)]
+ G_StoreWarn --> EndG4([Done])
+ G_MsgCount -->|"2 to N-1"| G_SilentInc[(Silent: Increment Count)]
+ G_SilentInc --> EndG5([Done])
+ G_MsgCount -->|">= warning_threshold"| G_RestrictUser[Apply Restriction
Mute Permissions]
+ G_RestrictUser --> G_MarkRestricted[(Mark Restricted in DB)]
+ G_MarkRestricted --> G_RestrictMsg[Send Restriction Notice
with DM Link]
+ G_RestrictMsg --> EndG6([Done])
+
+ %% ===================== DM Flow (dm.py, group=0, PRIVATE & TEXT) =====================
+ UpdateType -->|Private Message| DM_FindGroups[Scan Every Monitored Group
for This User's Membership]
+ DM_FindGroups --> DM_InGroup{Member of
Any Group?}
+ DM_InGroup -->|No| DM_NotInGroup[Send: Not in Group]
+ DM_InGroup -->|Yes| DM_PendingCaptcha{Pending Captcha in
Any Member Group?}
+ DM_PendingCaptcha -->|Yes| DM_CaptchaRedirect[Send: Complete Captcha
in Group First]
+ DM_PendingCaptcha -->|No| DM_CheckProfile[check_user_profile]
+ DM_CheckProfile --> DM_ProfileOk{Profile
Complete?}
+ DM_ProfileOk -->|No| DM_Missing[Send: Missing Items]
+ DM_ProfileOk -->|Yes| DM_FindRestricted[Find Groups Where
Bot Restricted This User]
+ DM_FindRestricted --> DM_AnyRestricted{Any Bot-Restricted
Group Found?}
+ DM_AnyRestricted -->|No| DM_NoRestriction[Send: No Bot Restriction
e.g. admin-restricted instead]
+ DM_AnyRestricted -->|Yes| DM_StatusCheck{Per Group: Still
Restricted on Telegram?}
+ DM_StatusCheck -->|No| DM_ClearOnly[(Clear DB Record Only)]
+ DM_StatusCheck -->|Yes| DM_Unrestrict[unrestrict_user API call]
+ DM_Unrestrict --> DM_ClearAndNotify[(Clear DB Record
+ Notify Warning Topic)]
+ DM_ClearOnly --> DM_Result{Any Group
Actually Unrestricted?}
+ DM_ClearAndNotify --> DM_Result
+ DM_Result -->|Yes, at least one| DM_Success[Send: Success Message]
+ DM_Result -->|No, all already clear| DM_AlreadyDone[Send: Already Unrestricted]
+
+ %% ===================== Scheduler Jobs (JobQueue, group=6, background) =====================
+ ComputeMap -.->|Every 5 min| SchedulerJob[auto_restrict_job]
+ SchedulerJob --> QueryDB[(Query Warnings Past
Time Threshold, All Groups)]
QueryDB --> HasExpired{Expired
Warnings?}
-
- HasExpired -->|No| EndJob([Wait Next Cycle])
- HasExpired -->|Yes| CheckKicked{User
Kicked?}
-
+ HasExpired -->|No| EndJob1([Wait Next Cycle])
+ HasExpired -->|Yes| CheckKicked{User
Kicked/Left?}
CheckKicked -->|Yes| ClearKicked[(Clear Record)]
- ClearKicked --> NextUser{More
Users?}
-
CheckKicked -->|No| ApplyTimeRestriction[Apply Restriction
Mute Permissions]
ApplyTimeRestriction --> MarkTimeRestricted[(Mark as Restricted)]
MarkTimeRestricted --> SendTimeNotice[Send Time-Based
Restriction Notice]
- SendTimeNotice --> NextUser
-
- NextUser -->|Yes| CheckKicked
- NextUser -->|No| EndJob
-
- %% Command Handlers - Verify/Unverify
- UpdateType -->|/verify Command| CheckAdminVerify{Is Admin?}
- CheckAdminVerify -->|No| DenyVerify[Send: Admin Only]
- CheckAdminVerify -->|Yes| AddWhitelist[(Add User to
Photo Whitelist)]
- AddWhitelist --> UnrestrictVerified[Unrestrict User]
- UnrestrictVerified --> DeleteWarnings[(Delete Warning Records)]
- DeleteWarnings --> CheckWarningsExist{Had
Warnings?}
- CheckWarningsExist -->|Yes| SendClearance[Send Clearance Notification
to Warning Topic]
- CheckWarningsExist -->|No| SendVerifySuccess[Send: User Verified]
- SendClearance --> SendVerifySuccess
-
- UpdateType -->|/unverify Command| CheckAdminUnverify{Is Admin?}
- CheckAdminUnverify -->|No| DenyUnverify[Send: Admin Only]
- CheckAdminUnverify -->|Yes| RemoveWhitelist[(Remove from Whitelist)]
- RemoveWhitelist --> SendUnverifySuccess[Send: User Unverified]
-
- %% Forwarded Message Handler
- UpdateType -->|Forwarded Message
in DM| CheckAdminForward{Is Admin?}
- CheckAdminForward -->|No| DenyForward[Send: Admin Only]
- CheckAdminForward -->|Yes| ExtractUser{Extract
User Info?}
- ExtractUser -->|Success| SendButtons[Send Verify/Unverify Buttons]
- ExtractUser -->|Failed| SendExtractError[Send: Cannot Extract User]
-
- %% Callback Handlers
- UpdateType -->|Verify Button| ProcessVerify[Process Verify Callback]
- ProcessVerify --> AddWhitelist
- UpdateType -->|Unverify Button| ProcessUnverify[Process Unverify Callback]
- ProcessUnverify --> RemoveWhitelist
-
+ ClearKicked --> EndJob1
+ SendTimeNotice --> EndJob1
+
+ ComputeMap -.->|Every 10 min| AdminRefreshJob[refresh_admin_ids_job]
+ AdminRefreshJob --> AdminRefreshLoop{Per Group:
API Call Succeeded?}
+ AdminRefreshLoop -->|Yes| AdminRefreshUpdate[(Update Cached
Admin IDs)]
+ AdminRefreshLoop -->|No| AdminRefreshKeep[(Keep Existing Cache
fail-soft, never empties)]
+
+ %% ===================== Admin Commands: verify / unverify / check (group=0, DM-only) =====================
+ UpdateType -->|"/verify user_id"| Cmd_VerifyAdmin{Admin +
DM?}
+ Cmd_VerifyAdmin -->|No| AdminDeny[Reply: Admin Only]
+ Cmd_VerifyAdmin -->|Yes| Do_Verify[verify_user]
+
+ UpdateType -->|"/unverify user_id"| Cmd_UnverifyAdmin{Admin +
DM?}
+ Cmd_UnverifyAdmin -->|No| AdminDeny
+ Cmd_UnverifyAdmin -->|Yes| Do_Unverify[unverify_user]
+
+ UpdateType -->|"/check user_id"| Cmd_CheckAdmin{Admin +
DM?}
+ Cmd_CheckAdmin -->|No| AdminDeny
+ Cmd_CheckAdmin -->|Yes| Do_CheckProfile[check_user_profile
_build_check_response]
+
+ UpdateType -->|Forwarded Message
in DM| Cmd_ForwardAdmin{Admin +
DM?}
+ Cmd_ForwardAdmin -->|No| AdminDeny
+ Cmd_ForwardAdmin -->|Yes| Cmd_ExtractUser{Extract User
Info from Forward?}
+ Cmd_ExtractUser -->|Failed| Cmd_ExtractError[Send: Cannot Extract User]
+ Cmd_ExtractUser -->|Success| Do_CheckProfile
+
+ Do_CheckProfile --> Cmd_ProfileState{Profile
Complete?}
+ Cmd_ProfileState -->|Complete| Cmd_ButtonsComplete[Show Unverify
+ Trust/Untrust Buttons]
+ Cmd_ProfileState -->|Incomplete| Cmd_ButtonsIncomplete[Show Warn + Verify
+ Trust/Untrust Buttons]
+
+ %% Callback: Verify / Unverify (also reachable from the buttons above)
+ UpdateType -->|"Verify Button
verify:user_id"| CB_VerifyAdmin{Admin?}
+ CB_VerifyAdmin -->|No| AdminDeny
+ CB_VerifyAdmin -->|Yes| Do_Verify
+ Do_Verify --> Verify_Whitelist[(Add to Photo
Verification Whitelist)]
+ Verify_Whitelist --> Verify_PerGroup[Per Group: Unrestrict
+ Delete Warning Records]
+ Verify_PerGroup --> Verify_HadWarnings{Had
Warnings?}
+ Verify_HadWarnings -->|Yes| Verify_Clearance[Send Clearance
to Warning Topic]
+ Verify_Clearance --> Verify_Success[Reply/Edit: User Verified]
+ Verify_HadWarnings -->|No| Verify_Success
+
+ UpdateType -->|"Unverify Button
unverify:user_id"| CB_UnverifyAdmin{Admin?}
+ CB_UnverifyAdmin -->|No| AdminDeny
+ CB_UnverifyAdmin -->|Yes| Do_Unverify
+ Do_Unverify --> Unverify_Remove[(Remove from Whitelist)]
+ Unverify_Remove --> Unverify_Success[Reply/Edit: User Unverified]
+
+ %% Callback: Warn (from check.py's incomplete-profile button)
+ UpdateType -->|"Warn Button
warn:user_id:code"| CB_WarnAdmin{Admin?}
+ CB_WarnAdmin -->|No| AdminDeny
+ CB_WarnAdmin -->|Yes| Warn_Build[Build Warning Text
from Missing Photo/Username Code]
+ Warn_Build --> Warn_Broadcast[Send to Every Monitored
Group's Warning Topic]
+ Warn_Broadcast --> Warn_Success[Edit Message: Sent]
+
+ %% ===================== Admin Commands: trust / untrust / trusted (group=0, DM-only) =====================
+ UpdateType -->|"/trust user_id or forward"| Cmd_TrustAdmin{Admin +
DM?}
+ Cmd_TrustAdmin -->|No| AdminDeny
+ Cmd_TrustAdmin -->|Yes| Do_Trust[trust_user]
+ UpdateType -->|"Trust Button
trust:user_id"| CB_TrustAdmin{Admin?}
+ CB_TrustAdmin -->|No| AdminDeny
+ CB_TrustAdmin -->|Yes| Do_Trust
+ Do_Trust --> Trust_DB[(Add TrustedUser Row
caches admin+user names
+ Clear Probation
+ Unrestrict, per group)]
+ Trust_DB --> Trust_Cache[(Update In-Memory
Trusted ID Cache)]
+ Trust_Cache --> Trust_Success[Reply: Trusted]
+
+ UpdateType -->|"/untrust user_id"| Cmd_UntrustAdmin{Admin +
DM?}
+ Cmd_UntrustAdmin -->|No| AdminDeny
+ Cmd_UntrustAdmin -->|Yes| Do_Untrust[untrust_user]
+ UpdateType -->|"Untrust Button
untrust:user_id"| CB_UntrustAdmin{Admin?}
+ CB_UntrustAdmin -->|No| AdminDeny
+ CB_UntrustAdmin -->|Yes| Do_Untrust
+ Do_Untrust --> Untrust_DB[(Remove TrustedUser Row)]
+ Untrust_DB --> Untrust_Cache[(Update Cache)]
+ Untrust_Cache --> Untrust_Success[Reply: Untrusted
or Not Found]
+
+ UpdateType -->|"/trusted"| Cmd_TrustedAdmin{Admin +
DM?}
+ Cmd_TrustedAdmin -->|No| AdminDeny
+ Cmd_TrustedAdmin -->|Yes| Trusted_Read[(Read All Trusted Users)]
+ Trusted_Read --> Trusted_List[Reply: Formatted List
name, username, trusted-by, date]
+
classDef processNode fill:#1a1a2e,stroke:#16213e,color:#eee
classDef decisionNode fill:#0f3460,stroke:#16213e,color:#eee
classDef dataNode fill:#16213e,stroke:#0f3460,color:#eee
classDef actionNode fill:#533483,stroke:#16213e,color:#eee
classDef endNode fill:#e94560,stroke:#16213e,color:#eee
classDef startNode fill:#1a5f7a,stroke:#16213e,color:#eee
-
- class Init,FetchAdmins,RecoverPending,StartJobs,Poll,CheckProfile,CheckDMProfile,RestrictAndChallenge,StorePending,ScheduleTimeout,WaitCaptcha,StartProbation,StartProbationAfter processNode
- class UpdateType,RecoverCaptcha,TopicGuard,IsAdmin,CheckBot,CheckWhitelist,ProfileComplete,CheckMode,CheckCount,CheckInGroup,CheckPendingCaptcha,DMProfileComplete,CheckBotRestricted,CheckCurrentStatus,HasExpired,CheckKicked,NextUser,CheckAdminVerify,CheckAdminUnverify,CaptchaAnswer,CheckCaptchaEnabled,CheckProbation,CheckExpired,CheckViolation,CheckWhitelisted,ViolationCount,CheckWarningsExist,CheckAdminForward,ExtractUser,CheckContact,CheckContactAdmin decisionNode
- class IncrementDB,SilentIncrement,MarkRestricted,ClearRecord,ClearRecord2,QueryDB,ClearKicked,MarkTimeRestricted,AddWhitelist,RemoveWhitelist,IncrementViolation,ClearProbation,DeleteWarnings dataNode
- class DeleteMsg,SendWarning,SendFirstWarning,RestrictUser,SendRestrictionMsg,SendNotInGroup,SendCaptchaRedirect,SendMissing,SendNoRestriction,SendAlreadyUnrestricted,UnrestrictUser,SendSuccess,ApplyTimeRestriction,SendTimeNotice,SchedulerJob,SendVerifySuccess,SendUnverifySuccess,DenyVerify,DenyUnverify,UnrestrictMember,KickMember,UpdateMessage,CancelTimeout,ShowError,DeleteSpam,SendSpamWarning,RestrictSpammer,SendSpamRestriction,UnrestrictVerified,SendClearance,DenyForward,SendButtons,SendExtractError,ProcessVerify,ProcessUnverify,DeleteContact,RestrictContact,SendContactNotify actionNode
- class End1,End2,End3,End4,End5,End6,End7,End8,End9,End10,EndJob,StartProbation endNode
+ classDef noteNode fill:#5c3a00,stroke:#e9c46a,color:#ffe8b3
+
+ class InitDB,RegisterHandlers,ComputeMap,RunPolling,FetchAdmins,RecoverPending,Poll,MaybeStart,CheckProfileCaptcha,DM_CheckProfile,DM_FindGroups,DM_FindRestricted,RestrictAndChallenge,SendChallenge,ScheduleTimeout,WaitCaptcha,G_CheckProfile,Do_CheckProfile processNode
+ class UpdateType,EntryPoint,RecoverCaptcha,CheckCaptchaEnabled,CaptchaAnswer,ProfileOkCaptcha,G_TopicGuard,G_TopicIsBotAdmin,G_InlineGate,G_InlineCheck,G_ContactGate,G_ContactCheck,G_ContactRestrictGate,G_NewUserGate,G_OnProbation,G_ProbationExpired,G_Violation,G_ViolationCount,G_DupGate,G_DupLength,G_DupSimilar,G_ProfileGate,G_Whitelist,G_ProfileComplete,G_Mode,G_MsgCount,DM_InGroup,DM_PendingCaptcha,DM_ProfileOk,DM_AnyRestricted,DM_StatusCheck,DM_Result,HasExpired,CheckKicked,AdminRefreshLoop,Cmd_VerifyAdmin,Cmd_UnverifyAdmin,Cmd_CheckAdmin,Cmd_ForwardAdmin,Cmd_ExtractUser,Cmd_ProfileState,CB_VerifyAdmin,CB_UnverifyAdmin,Verify_HadWarnings,CB_WarnAdmin,Cmd_TrustAdmin,CB_TrustAdmin,Cmd_UntrustAdmin,CB_UntrustAdmin,Cmd_TrustedAdmin decisionNode
+ class StartProbationInit,StorePending,RemovePending,RestartProbation,G_ClearProbation,G_ViolationIncrement,G_StoreWarn,G_SilentInc,G_MarkRestricted,DM_ClearOnly,DM_ClearAndNotify,QueryDB,ClearKicked,MarkTimeRestricted,AdminRefreshUpdate,AdminRefreshKeep,Verify_Whitelist,Unverify_Remove,Trust_DB,Trust_Cache,Untrust_DB,Untrust_Cache,Trusted_Read,LoadTrusted dataNode
+ class ShowError,CaptchaTimeout,HandleExpiration,ShowIncomplete,UnrestrictMember,CancelTimeout,UpdateMessage,G_TopicDelete,G_InlineDelete,G_InlineRestrict,G_InlineNotify,G_ContactDelete,G_ContactRestrict,G_ContactNotify,G_ViolationDelete,G_ViolationWarn,G_ViolationRestrict,G_ViolationNotify,G_DupDelete,G_DupRestrict,G_DupNotify,G_WarnOnly,G_FirstWarn,G_RestrictUser,G_RestrictMsg,DM_NotInGroup,DM_CaptchaRedirect,DM_Missing,DM_NoRestriction,DM_Unrestrict,DM_Success,DM_AlreadyDone,SchedulerJob,ApplyTimeRestriction,SendTimeNotice,AdminRefreshJob,AdminDeny,Cmd_ExtractError,Cmd_ButtonsComplete,Cmd_ButtonsIncomplete,Do_Verify,Do_Unverify,Do_Trust,Do_Untrust,Verify_PerGroup,Verify_Clearance,Verify_Success,Unverify_Success,Warn_Build,Warn_Broadcast,Warn_Success,Trust_Success,Untrust_Success,Trusted_List actionNode
+ class EndNM1,EndNM2,EndNM3,StopTopic1,StopTopic2,StopInline,StopContact,StopNewUser1,StopNewUser2,StopNewUser3,StopDup,EndG0,EndG1,EndG2,EndG3,EndG4,EndG5,EndG6,EndJob1 endNode
class Start startNode
+ class G_Group4Note noteNode
```
## How It Works
@@ -492,6 +579,7 @@ The bot is organized into clear modules for maintainability:
- `captcha.py`: Captcha verification for new members, including profile photo + username check
- `anti_spam.py`: Inline keyboard spam (group=1) + contact card spam (group=2) + new user probation enforcement (group=3)
- `duplicate_spam.py`: Repeated message detection (group=4)
+ - `bio_bait.py`: Obfuscated bait-phrase + suspicious profile-bio link detection (group=4, monitor-only mode available)
- `verify.py`: /verify and /unverify command handlers
- `check.py`: /check command + forwarded message handling
- `trust.py`: /trust, /untrust, /trusted admin commands (TrustedUser table caches names at trust time so /trusted renders without API calls)
@@ -593,12 +681,21 @@ When a restricted user DMs the bot (or sends `/start`):
| `NEW_USER_PROBATION_HOURS` | Hours new users can't send links/forwards | `72` (3 days) |
| `NEW_USER_VIOLATION_THRESHOLD` | Spam violations before restriction | `3` |
| `CONTACT_SPAM_RESTRICT` | Restrict users who share contact cards | `true` |
+| `DUPLICATE_SPAM_ENABLED` | Enable duplicate-message detection | `true` |
+| `DUPLICATE_SPAM_WINDOW_SECONDS` | Window to compare messages for duplicates | `120` |
+| `DUPLICATE_SPAM_THRESHOLD` | Repeats within window before flagging | `2` |
+| `DUPLICATE_SPAM_MIN_LENGTH` | Minimum message length considered | `20` |
+| `DUPLICATE_SPAM_SIMILARITY` | Similarity ratio (0-1) to count as duplicate | `0.95` |
+| `BIO_BAIT_ENABLED` | Enable bio-bait phrase/link detection | `true` |
+| `BIO_BAIT_MONITOR_ONLY` | Log/alert only, skip delete + restrict | `false` |
+| `BIO_BAIT_ALERT_CHAT_ID` | Chat ID to receive monitor-only detection alerts | None |
| `DATABASE_PATH` | SQLite database path | `data/bot.db` |
| `RULES_LINK` | Link to group rules message | `https://t.me/pythonID/290029/321799` |
| `LOGFIRE_ENABLED` | Enable Logfire logging integration | `true` |
| `LOGFIRE_TOKEN` | Logfire API token (optional) | None |
| `LOG_LEVEL` | Logging level (DEBUG/INFO/WARNING/ERROR) | `INFO` |
| `GROUPS_CONFIG_PATH` | Path to `groups.json` for multi-group support | None (single-group mode from `.env`) |
+| `PLUGINS_DEFAULT` | Bot-wide plugin enable/disable defaults (JSON object, e.g. `{"bio_bait_spam": false}`) | `{}` (all enabled) |
### Restriction Modes
diff --git a/src/bot/config.py b/src/bot/config.py
index fb8f15c..f7454f8 100644
--- a/src/bot/config.py
+++ b/src/bot/config.py
@@ -9,7 +9,6 @@
import json
import logging
import os
-from datetime import timedelta
from functools import lru_cache
from pathlib import Path
@@ -28,11 +27,7 @@ def get_env_file() -> str | None:
- "staging" -> ".env.staging" (if exists)
"""
env = os.getenv("BOT_ENV", "production")
- env_files = {
- "production": ".env",
- "staging": ".env.staging",
- }
- env_file = env_files.get(env, ".env")
+ env_file = ".env.staging" if env == "staging" else ".env"
# Return path only if file exists, otherwise return None
# Pydantic will load from environment variables if no .env file
@@ -143,35 +138,27 @@ def model_post_init(self, __context):
self.logfire_environment = "staging"
logger.info("Configuration loaded successfully")
- logger.debug(f"group_id: {self.group_id}")
- logger.debug(f"warning_topic_id: {self.warning_topic_id}")
- logger.debug(f"restrict_failed_users: {self.restrict_failed_users}")
- logger.debug(f"warning_threshold: {self.warning_threshold}")
- logger.debug(f"warning_time_threshold_minutes: {self.warning_time_threshold_minutes}")
- logger.debug(f"database_path: {self.database_path}")
- logger.debug(f"captcha_enabled: {self.captcha_enabled}")
- logger.debug(f"captcha_timeout_seconds: {self.captcha_timeout_seconds}")
- logger.debug(f"new_user_probation_hours: {self.new_user_probation_hours}")
- logger.debug(f"new_user_violation_threshold: {self.new_user_violation_threshold}")
- logger.debug(f"bio_bait_enabled: {self.bio_bait_enabled}")
- logger.debug(f"bio_bait_monitor_only: {self.bio_bait_monitor_only}")
- logger.debug(f"bio_bait_alert_chat_id: {self.bio_bait_alert_chat_id}")
- logger.debug(f"telegram_bot_token: {'***' + self.telegram_bot_token[-4:]}") # Mask sensitive token
+ for field in (
+ "group_id",
+ "warning_topic_id",
+ "restrict_failed_users",
+ "warning_threshold",
+ "warning_time_threshold_minutes",
+ "database_path",
+ "captcha_enabled",
+ "captcha_timeout_seconds",
+ "new_user_probation_hours",
+ "new_user_violation_threshold",
+ "bio_bait_enabled",
+ "bio_bait_monitor_only",
+ "bio_bait_alert_chat_id",
+ ):
+ logger.debug(f"{field}: {getattr(self, field)}")
+ logger.debug(f"telegram_bot_token: {'***' + self.telegram_bot_token[-4:]}")
logger.debug(f"logfire_enabled: {self.logfire_enabled}")
logger.debug(f"logfire_environment: {self.logfire_environment}")
logger.debug(f"plugins_default: {self.plugins_default}")
- @property
- def probation_timedelta(self) -> timedelta:
- return timedelta(hours=self.new_user_probation_hours)
-
- @property
- def warning_time_threshold_timedelta(self) -> timedelta:
- return timedelta(minutes=self.warning_time_threshold_minutes)
-
- @property
- def captcha_timeout_timedelta(self) -> timedelta:
- return timedelta(seconds=self.captcha_timeout_seconds)
@lru_cache
def get_settings() -> Settings:
diff --git a/src/bot/database/service.py b/src/bot/database/service.py
index ed47c39..32b2518 100644
--- a/src/bot/database/service.py
+++ b/src/bot/database/service.py
@@ -274,20 +274,13 @@ def delete_user_warnings(self, user_id: int, group_id: int) -> int:
int: Number of warning records deleted.
"""
with Session(self._engine) as session:
- # First count records to be deleted
- count_statement = select(UserWarning).where(
- UserWarning.user_id == user_id,
- UserWarning.group_id == group_id,
- )
- count = len(session.exec(count_statement).all())
-
- # Bulk delete using SQLModel delete statement
delete_statement = delete(UserWarning).where(
UserWarning.user_id == user_id,
UserWarning.group_id == group_id,
)
- session.exec(delete_statement)
+ result = session.exec(delete_statement)
session.commit()
+ count = result.rowcount
logger.info(
f"Deleted warnings: user_id={user_id}, group_id={group_id}, count={count}"
)
@@ -482,29 +475,13 @@ def update_trusted_user_names(
session.add(record)
session.commit()
- def is_user_trusted(self, user_id: int, group_id: int | None = None) -> bool:
+ def is_user_trusted(self, user_id: int) -> bool:
"""
Check whether a user is trusted.
- Currently only global trust (group_id=None or 0) is supported; passing
- a non-zero group_id raises NotImplementedError.
-
- Args:
- user_id: Telegram user ID.
- group_id: Optional group scope. Must be None or 0 for now.
-
Returns:
bool: True if user is trusted globally.
-
- Raises:
- NotImplementedError: If a non-zero group_id is provided.
"""
- if group_id is not None and group_id != 0:
- raise NotImplementedError(
- "Per-group trusted user reads are not yet supported; "
- "only global (group_id=0) trust is implemented."
- )
-
with Session(self._engine) as session:
statement = select(TrustedUser).where(
TrustedUser.user_id == user_id,
@@ -513,54 +490,24 @@ def is_user_trusted(self, user_id: int, group_id: int | None = None) -> bool:
record = session.exec(statement).first()
return record is not None
- def get_trusted_user_ids(self, group_id: int | None = None) -> set[int]:
+ def get_trusted_user_ids(self) -> set[int]:
"""
Get trusted user IDs.
- Currently only global trust (group_id=None or 0) is supported; passing
- a non-zero group_id raises NotImplementedError.
-
- Args:
- group_id: Optional group scope. Must be None or 0 for now.
-
Returns:
set[int]: Trusted user IDs scoped to global (group_id=0).
-
- Raises:
- NotImplementedError: If a non-zero group_id is provided.
"""
- if group_id is not None and group_id != 0:
- raise NotImplementedError(
- "Per-group trusted user reads are not yet supported; "
- "only global (group_id=0) trust is implemented."
- )
-
with Session(self._engine) as session:
statement = select(TrustedUser.user_id).where(TrustedUser.group_id == 0)
return set(session.exec(statement).all())
- def get_trusted_users(self, group_id: int | None = None) -> list[TrustedUser]:
+ def get_trusted_users(self) -> list[TrustedUser]:
"""
Get trusted user records with metadata.
- Currently only global trust (group_id=None or 0) is supported; passing
- a non-zero group_id raises NotImplementedError.
-
- Args:
- group_id: Optional group scope. Must be None or 0 for now.
-
Returns:
list[TrustedUser]: Trusted user records scoped to global (group_id=0).
-
- Raises:
- NotImplementedError: If a non-zero group_id is provided.
"""
- if group_id is not None and group_id != 0:
- raise NotImplementedError(
- "Per-group trusted user reads are not yet supported; "
- "only global (group_id=0) trust is implemented."
- )
-
with Session(self._engine) as session:
statement = (
select(TrustedUser)
@@ -569,33 +516,6 @@ def get_trusted_users(self, group_id: int | None = None) -> list[TrustedUser]:
)
return list(session.exec(statement).all())
- def get_warnings_past_time_threshold(
- self, threshold: timedelta
- ) -> list[UserWarning]:
- """
- Find all active warnings that have exceeded the time threshold.
-
- Looks for non-restricted warning records where the time elapsed
- since first_warned_at exceeds the threshold, regardless of message count.
-
- Args:
- threshold: Time duration since first warning to trigger restriction.
-
- Returns:
- list[UserWarning]: List of warning records that should be auto-restricted.
- """
- with Session(self._engine) as session:
- cutoff_time = datetime.now(UTC) - threshold
- statement = select(UserWarning).where(
- ~UserWarning.is_restricted,
- UserWarning.first_warned_at <= cutoff_time,
- )
- records = session.exec(statement).all()
- logger.info(
- f"Found {len(records)} warnings past {threshold} threshold"
- )
- return [record for record in records]
-
def get_warnings_past_time_threshold_for_group(
self, group_id: int, threshold: timedelta
) -> list[UserWarning]:
diff --git a/src/bot/group_config.py b/src/bot/group_config.py
index 160205a..b321efc 100644
--- a/src/bot/group_config.py
+++ b/src/bot/group_config.py
@@ -125,9 +125,6 @@ def get(self, group_id: int) -> GroupConfig | None:
def all_groups(self) -> list[GroupConfig]:
return list(self._groups.values())
- def is_monitored(self, group_id: int) -> bool:
- return group_id in self._groups
-
def load_groups_from_json(path: str) -> list[GroupConfig]:
"""
Parse a groups.json file into a list of GroupConfig objects.
@@ -188,25 +185,11 @@ def build_group_registry(settings: object) -> GroupRegistry:
else:
logger.info("No groups.json found, using single-group config from .env")
config = GroupConfig(
- group_id=settings.group_id,
- warning_topic_id=settings.warning_topic_id,
- restrict_failed_users=settings.restrict_failed_users,
- warning_threshold=settings.warning_threshold,
- warning_time_threshold_minutes=settings.warning_time_threshold_minutes,
- captcha_enabled=settings.captcha_enabled,
- captcha_timeout_seconds=settings.captcha_timeout_seconds,
- new_user_probation_hours=settings.new_user_probation_hours,
- new_user_violation_threshold=settings.new_user_violation_threshold,
- rules_link=settings.rules_link,
- contact_spam_restrict=settings.contact_spam_restrict,
- duplicate_spam_enabled=settings.duplicate_spam_enabled,
- duplicate_spam_window_seconds=settings.duplicate_spam_window_seconds,
- duplicate_spam_threshold=settings.duplicate_spam_threshold,
- duplicate_spam_min_length=settings.duplicate_spam_min_length,
- duplicate_spam_similarity=settings.duplicate_spam_similarity,
- bio_bait_enabled=getattr(settings, "bio_bait_enabled", True),
- bio_bait_monitor_only=getattr(settings, "bio_bait_monitor_only", False),
- bio_bait_alert_chat_id=getattr(settings, "bio_bait_alert_chat_id", None),
+ **{
+ f: getattr(settings, f, GroupConfig.model_fields[f].default)
+ for f in GroupConfig.model_fields
+ if f != "plugins"
+ },
plugins=getattr(settings, "plugins_default", None),
)
registry.register(config)
diff --git a/src/bot/handlers/anti_spam.py b/src/bot/handlers/anti_spam.py
index 107b3ca..01de11a 100644
--- a/src/bot/handlers/anti_spam.py
+++ b/src/bot/handlers/anti_spam.py
@@ -45,23 +45,6 @@ def is_forwarded(message: Message) -> bool:
return message.forward_origin is not None
-def has_link(message: Message) -> bool:
- """
- Check if a message contains URLs or text links.
-
- Checks both message entities and caption entities for URL types.
-
- Args:
- message: Telegram message to check.
-
- Returns:
- bool: True if message contains links.
- """
- entities = list(message.entities or []) + list(message.caption_entities or [])
- link_types = {MessageEntity.URL, MessageEntity.TEXT_LINK}
- return any(entity.type in link_types for entity in entities)
-
-
def has_external_reply(message: Message) -> bool:
"""
Check if a message has an external reply (quote from another chat).
@@ -220,18 +203,30 @@ def has_contact(message: Message) -> bool:
return message.contact is not None
-async def handle_contact_spam(
- update: Update, context: ContextTypes.DEFAULT_TYPE
+async def _handle_group_spam(
+ update: Update,
+ context: ContextTypes.DEFAULT_TYPE,
+ *,
+ detector,
+ label: str,
+ should_restrict,
+ template_restricted: str,
+ template_no_restrict: str,
) -> None:
"""
- Handle contact card sharing in monitored groups.
+ Shared helper for handling group spam detection and enforcement.
- Blocks ALL non-admin members from sending contact cards.
- Deletes the message and sends a notification to the warning topic.
+ Implements the common skeleton: guard clauses → detector check → delete
+ message → conditionally restrict → send notification → raise ApplicationHandlerStop.
Args:
update: Telegram update containing the message.
context: Bot context with helper methods.
+ detector: Callable that takes a Message and returns bool if spam detected.
+ label: Description for logging (e.g. "contact spam").
+ should_restrict: Callable taking group_config and returning bool.
+ template_restricted: Notification template when user is restricted.
+ template_no_restrict: Notification template when user is not restricted.
"""
if not update.message or not update.message.from_user:
return
@@ -248,26 +243,26 @@ async def handle_contact_spam(
return
msg = update.message
- if not has_contact(msg):
+ if not detector(msg):
return
user_mention = get_user_mention(user)
logger.info(
- f"Contact spam detected: user_id={user.id}, "
+ f"{label.capitalize()} detected: user_id={user.id}, "
f"group_id={group_config.group_id}"
)
try:
await msg.delete()
- logger.info(f"Deleted contact spam from user_id={user.id}")
+ logger.info(f"Deleted {label} from user_id={user.id}")
except Exception:
logger.error(
- f"Failed to delete contact spam: user_id={user.id}",
+ f"Failed to delete {label}: user_id={user.id}",
exc_info=True,
)
restricted = False
- if group_config.contact_spam_restrict:
+ if should_restrict(group_config):
try:
await context.bot.restrict_chat_member(
chat_id=group_config.group_id,
@@ -275,18 +270,15 @@ async def handle_contact_spam(
permissions=RESTRICTED_PERMISSIONS,
)
restricted = True
- logger.info(f"Restricted user_id={user.id} for contact spam")
+ logger.info(f"Restricted user_id={user.id} for {label}")
except Exception:
logger.error(
- f"Failed to restrict user for contact spam: user_id={user.id}",
+ f"Failed to restrict user for {label}: user_id={user.id}",
exc_info=True,
)
try:
- template = (
- CONTACT_SPAM_NOTIFICATION if restricted
- else CONTACT_SPAM_NOTIFICATION_NO_RESTRICT
- )
+ template = template_restricted if restricted else template_no_restrict
notification_text = template.format(
user_mention=user_mention,
rules_link=group_config.rules_link,
@@ -297,16 +289,40 @@ async def handle_contact_spam(
text=notification_text,
parse_mode="Markdown",
)
- logger.info(f"Sent contact spam notification for user_id={user.id}")
+ logger.info(f"Sent {label} notification for user_id={user.id}")
except Exception:
logger.error(
- f"Failed to send contact spam notification: user_id={user.id}",
+ f"Failed to send {label} notification: user_id={user.id}",
exc_info=True,
)
raise ApplicationHandlerStop
+async def handle_contact_spam(
+ update: Update, context: ContextTypes.DEFAULT_TYPE
+) -> None:
+ """
+ Handle contact card sharing in monitored groups.
+
+ Blocks ALL non-admin members from sending contact cards.
+ Deletes the message and sends a notification to the warning topic.
+
+ Args:
+ update: Telegram update containing the message.
+ context: Bot context with helper methods.
+ """
+ await _handle_group_spam(
+ update,
+ context,
+ detector=has_contact,
+ label="contact spam",
+ should_restrict=lambda cfg: cfg.contact_spam_restrict,
+ template_restricted=CONTACT_SPAM_NOTIFICATION,
+ template_no_restrict=CONTACT_SPAM_NOTIFICATION_NO_RESTRICT,
+ )
+
+
async def handle_inline_keyboard_spam(
update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
@@ -321,78 +337,16 @@ async def handle_inline_keyboard_spam(
update: Telegram update containing the message.
context: Bot context with helper methods.
"""
- if not update.message or not update.message.from_user:
- return
-
- group_config = get_group_config_for_update(update)
- if group_config is None:
- return
-
- user = update.message.from_user
- if user.is_bot:
- return
-
- if is_user_admin_or_trusted(context, group_config.group_id, user.id):
- return
-
- msg = update.message
- if not has_non_whitelisted_inline_keyboard_urls(msg):
- return
-
- user_mention = get_user_mention(user)
- logger.info(
- f"Inline keyboard spam detected: user_id={user.id}, "
- f"group_id={group_config.group_id}"
+ await _handle_group_spam(
+ update,
+ context,
+ detector=has_non_whitelisted_inline_keyboard_urls,
+ label="inline keyboard spam",
+ should_restrict=lambda cfg: True,
+ template_restricted=INLINE_KEYBOARD_SPAM_NOTIFICATION,
+ template_no_restrict=INLINE_KEYBOARD_SPAM_NOTIFICATION_NO_RESTRICT,
)
- try:
- await msg.delete()
- logger.info(f"Deleted inline keyboard spam from user_id={user.id}")
- except Exception:
- logger.error(
- f"Failed to delete inline keyboard spam: user_id={user.id}",
- exc_info=True,
- )
-
- restricted = False
- try:
- await context.bot.restrict_chat_member(
- chat_id=group_config.group_id,
- user_id=user.id,
- permissions=RESTRICTED_PERMISSIONS,
- )
- restricted = True
- logger.info(f"Restricted user_id={user.id} for inline keyboard spam")
- except Exception:
- logger.error(
- f"Failed to restrict user for inline keyboard spam: user_id={user.id}",
- exc_info=True,
- )
-
- try:
- template = (
- INLINE_KEYBOARD_SPAM_NOTIFICATION if restricted
- else INLINE_KEYBOARD_SPAM_NOTIFICATION_NO_RESTRICT
- )
- notification_text = template.format(
- user_mention=user_mention,
- rules_link=group_config.rules_link,
- )
- await context.bot.send_message(
- chat_id=group_config.group_id,
- message_thread_id=group_config.warning_topic_id,
- text=notification_text,
- parse_mode="Markdown",
- )
- logger.info(f"Sent inline keyboard spam notification for user_id={user.id}")
- except Exception:
- logger.error(
- f"Failed to send inline keyboard spam notification: user_id={user.id}",
- exc_info=True,
- )
-
- raise ApplicationHandlerStop
-
async def handle_new_user_spam(
update: Update, context: ContextTypes.DEFAULT_TYPE
diff --git a/src/bot/handlers/bio_bait.py b/src/bot/handlers/bio_bait.py
index cee1ce6..8aee4bc 100644
--- a/src/bot/handlers/bio_bait.py
+++ b/src/bot/handlers/bio_bait.py
@@ -61,9 +61,6 @@
# Sentinel value to indicate a cached failure
_BIO_CACHE_FAILURE = "__FAILURE__"
-# Bio bait metrics stored in bot_data.
-BIO_BAIT_METRICS_KEY = "bio_bait_metrics"
-
# Telegram hard limit per message text.
MAX_TELEGRAM_MESSAGE_LENGTH = 4096
@@ -239,33 +236,6 @@ def clear_cached_user_bio(
_get_user_bio_cache(context).pop(user_id, None)
-def _get_bio_bait_metrics(context: ContextTypes.DEFAULT_TYPE) -> dict[str, int]:
- """Get or initialize bio bait metrics stored in bot_data."""
- return context.bot_data.setdefault(BIO_BAIT_METRICS_KEY, {})
-
-
-def _increment_bio_bait_metric(
- context: ContextTypes.DEFAULT_TYPE,
- metric_name: str,
-) -> None:
- """Increment a named bio bait metric counter."""
- metrics = _get_bio_bait_metrics(context)
- metrics[metric_name] = metrics.get(metric_name, 0) + 1
-
-
-def record_bio_bait_detection_metrics(
- context: ContextTypes.DEFAULT_TYPE,
- detection_reason: str,
- monitor_only: bool,
-) -> None:
- """Record counters for a bio bait detection event."""
- _increment_bio_bait_metric(context, "detections_total")
- _increment_bio_bait_metric(context, f"detections_{detection_reason}")
- if monitor_only:
- _increment_bio_bait_metric(context, "monitor_only_matches")
- else:
- _increment_bio_bait_metric(context, "enforced_matches")
-
def _chunk_telegram_text(text: str, max_length: int = MAX_TELEGRAM_MESSAGE_LENGTH) -> list[str]:
"""Split text into Telegram-safe chunks."""
@@ -393,7 +363,6 @@ async def handle_bio_bait_spam(
)
monitor_only = group_config.bio_bait_monitor_only
- record_bio_bait_detection_metrics(context, detection_reason, monitor_only)
if monitor_only:
alert_chat_id = group_config.bio_bait_alert_chat_id
@@ -403,11 +372,10 @@ async def handle_bio_bait_spam(
logger.warning(
f"Skipping bio bait monitor alert: alert_chat_id matches monitored group (warning topic). group_id={group_config.group_id}"
)
- _increment_bio_bait_metric(context, "owner_alert_skipped_warning_topic")
else:
if user_bio is None:
user_bio = await get_cached_user_bio(context, user.id)
- sent = await send_monitor_alert_to_owner(
+ await send_monitor_alert_to_owner(
context=context,
alert_chat_id=alert_chat_id,
group_id=group_config.group_id,
@@ -418,10 +386,6 @@ async def handle_bio_bait_spam(
message_text=text,
profile_bio=user_bio,
)
- if sent:
- _increment_bio_bait_metric(context, "owner_alert_sent")
- else:
- _increment_bio_bait_metric(context, "owner_alert_failed")
logger.info(
f"Bio bait monitor-only mode: no delete/restrict (user_id={user.id}, group_id={group_config.group_id})"
diff --git a/src/bot/handlers/captcha.py b/src/bot/handlers/captcha.py
index 1910b60..973b0f0 100644
--- a/src/bot/handlers/captcha.py
+++ b/src/bot/handlers/captcha.py
@@ -29,7 +29,7 @@
MISSING_ITEMS_SEPARATOR,
RESTRICTED_PERMISSIONS,
)
-from bot.database.service import get_database
+from bot.database.service import DatabaseService, get_database
from bot.group_config import GroupConfig, get_group_config_for_update, get_group_registry
from bot.services.telegram_utils import get_user_mention, unrestrict_user
from bot.services.user_checker import check_user_profile
@@ -135,6 +135,33 @@ async def _initiate_captcha_challenge(
)
+async def _maybe_start_captcha(
+ context: ContextTypes.DEFAULT_TYPE,
+ db: DatabaseService,
+ member: User,
+ group_config: GroupConfig,
+) -> None:
+ """
+ Maybe start captcha verification for a new member.
+
+ Starts probation, checks if captcha is enabled, checks for duplicates,
+ and initiates captcha challenge if appropriate.
+ """
+ user_id = member.id
+
+ db.start_new_user_probation(user_id, group_config.group_id)
+
+ if not group_config.captcha_enabled:
+ logger.info(f"Captcha disabled, probation started for user {user_id}")
+ return
+
+ if db.get_pending_captcha(user_id, group_config.group_id):
+ logger.info(f"Captcha already pending for user {user_id}, skipping duplicate")
+ return
+
+ await _initiate_captcha_challenge(context, member, group_config.group_id, group_config)
+
+
async def new_member_handler(
update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
@@ -166,21 +193,7 @@ async def new_member_handler(
if new_member.is_bot:
continue
- user_id = new_member.id
-
- # Start probation for all new users (regardless of captcha setting)
- db.start_new_user_probation(user_id, group_config.group_id)
-
- # If captcha is disabled, we're done - user just gets probation
- if not group_config.captcha_enabled:
- logger.info(f"Captcha disabled, probation started for user {user_id}")
- continue
-
- if db.get_pending_captcha(user_id, group_config.group_id):
- logger.info(f"Captcha already pending for user {user_id}, skipping duplicate (new_member_handler)")
- continue
-
- await _initiate_captcha_challenge(context, new_member, group_config.group_id, group_config)
+ await _maybe_start_captcha(context, db, new_member, group_config)
async def chat_member_handler(
@@ -231,22 +244,8 @@ async def chat_member_handler(
logger.info(f"Detected new member via ChatMemberUpdated: {new_member.id} ({new_member.full_name})")
- # Start probation for all new users (regardless of captcha setting)
db = get_database()
- db.start_new_user_probation(new_member.id, group_config.group_id)
-
- # If captcha is disabled, we're done - user just gets probation
- if not group_config.captcha_enabled:
- logger.info(f"Captcha disabled, probation started for user {new_member.id}")
- return
-
- user_id = new_member.id
-
- if db.get_pending_captcha(user_id, group_config.group_id):
- logger.info(f"Captcha already pending for user {user_id}, skipping duplicate (chat_member_handler)")
- return
-
- await _initiate_captcha_challenge(context, new_member, group_config.group_id, group_config)
+ await _maybe_start_captcha(context, db, new_member, group_config)
async def captcha_callback_handler(
diff --git a/src/bot/handlers/check.py b/src/bot/handlers/check.py
index 47fc0de..901289e 100644
--- a/src/bot/handlers/check.py
+++ b/src/bot/handlers/check.py
@@ -29,6 +29,7 @@
extract_forwarded_user,
get_user_mention,
get_user_mention_by_id,
+ require_admin_dm_target,
)
from bot.services.user_checker import check_user_profile
@@ -124,35 +125,16 @@ async def handle_check_command(
Only works in bot DMs for admins.
"""
- if not update.message or not update.message.from_user:
- return
-
- if update.effective_chat and update.effective_chat.type != "private":
- await update.message.reply_text(
- "❌ Perintah ini hanya bisa digunakan di chat pribadi dengan bot."
- )
+ target_user_id = await require_admin_dm_target(
+ update,
+ context,
+ "❌ Penggunaan: /check USER_ID",
+ "/check command",
+ )
+ if target_user_id is None:
return
admin_user_id = update.message.from_user.id
- admin_ids = context.bot_data.get("admin_ids", [])
-
- if admin_user_id not in admin_ids:
- await update.message.reply_text("❌ Kamu tidak memiliki izin untuk menggunakan perintah ini.")
- logger.warning(
- f"Non-admin user {admin_user_id} ({update.message.from_user.full_name}) "
- f"attempted to use /check command"
- )
- return
-
- if not context.args or len(context.args) == 0:
- await update.message.reply_text("❌ Penggunaan: /check USER_ID")
- return
-
- try:
- target_user_id = int(context.args[0])
- except ValueError:
- await update.message.reply_text("❌ User ID harus berupa angka.")
- return
try:
# Get user info for display name
diff --git a/src/bot/handlers/duplicate_spam.py b/src/bot/handlers/duplicate_spam.py
index 0199765..bfbd124 100644
--- a/src/bot/handlers/duplicate_spam.py
+++ b/src/bot/handlers/duplicate_spam.py
@@ -70,13 +70,9 @@ def _get_recent_messages(
context: ContextTypes.DEFAULT_TYPE, group_id: int, user_id: int
) -> deque[RecentMessage]:
"""Get or create the recent messages deque for a (group, user) pair."""
- store: dict[tuple[int, int], deque[RecentMessage]] = context.bot_data.setdefault(
- RECENT_MESSAGES_KEY, {}
+ return context.bot_data.setdefault(RECENT_MESSAGES_KEY, {}).setdefault(
+ (group_id, user_id), deque()
)
- key = (group_id, user_id)
- if key not in store:
- store[key] = deque()
- return store[key]
def _prune_old_messages(
diff --git a/src/bot/handlers/trust.py b/src/bot/handlers/trust.py
index 09202ce..e1dce0c 100644
--- a/src/bot/handlers/trust.py
+++ b/src/bot/handlers/trust.py
@@ -36,19 +36,11 @@
def _add_trusted_cache(context: ContextTypes.DEFAULT_TYPE, user_id: int) -> None:
- trusted_ids = context.bot_data.get("trusted_user_ids")
- if trusted_ids is None:
- trusted_ids = set()
- context.bot_data["trusted_user_ids"] = trusted_ids
- trusted_ids.add(user_id)
+ context.bot_data.setdefault("trusted_user_ids", set()).add(user_id)
def _remove_trusted_cache(context: ContextTypes.DEFAULT_TYPE, user_id: int) -> None:
- trusted_ids = context.bot_data.get("trusted_user_ids")
- if trusted_ids is None:
- trusted_ids = set()
- context.bot_data["trusted_user_ids"] = trusted_ids
- trusted_ids.discard(user_id)
+ context.bot_data.setdefault("trusted_user_ids", set()).discard(user_id)
def _format_person(full_name: str, user_id: int) -> str:
@@ -152,14 +144,6 @@ async def trust_user(
return cleared_probation, unrestricted_groups
-async def untrust_user(
- db: DatabaseService,
- target_user_id: int,
-) -> None:
- """Remove trusted user entry."""
- db.remove_trusted_user(user_id=target_user_id)
-
-
async def handle_trust_command(
update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
@@ -245,7 +229,7 @@ async def handle_untrust_command(
db = get_database()
try:
- await untrust_user(db, target_user_id)
+ db.remove_trusted_user(user_id=target_user_id)
_remove_trusted_cache(context, target_user_id)
await update.message.reply_text(
TRUST_REMOVED_MESSAGE.format(user_id=target_user_id),
@@ -384,7 +368,7 @@ async def handle_untrust_callback(
db = get_database()
try:
- await untrust_user(db, target_user_id)
+ db.remove_trusted_user(user_id=target_user_id)
_remove_trusted_cache(context, target_user_id)
await query.edit_message_text(
TRUST_REMOVED_MESSAGE.format(user_id=target_user_id),
diff --git a/src/bot/handlers/verify.py b/src/bot/handlers/verify.py
index e84b038..29e6da7 100644
--- a/src/bot/handlers/verify.py
+++ b/src/bot/handlers/verify.py
@@ -15,7 +15,7 @@
from bot.constants import VERIFICATION_CLEARANCE_MESSAGE
from bot.database.service import DatabaseService, get_database
from bot.group_config import GroupRegistry, get_group_registry
-from bot.services.telegram_utils import get_user_mention, unrestrict_user
+from bot.services.telegram_utils import get_user_mention, require_admin_dm_target, unrestrict_user
logger = logging.getLogger(__name__)
@@ -139,35 +139,16 @@ async def handle_verify_command(
update: Telegram update containing the command.
context: Bot context with helper methods.
"""
- if not update.message or not update.message.from_user:
- return
-
- if update.effective_chat and update.effective_chat.type != "private":
- await update.message.reply_text(
- "❌ Perintah ini hanya bisa digunakan di chat pribadi dengan bot."
- )
+ target_user_id = await require_admin_dm_target(
+ update,
+ context,
+ "❌ Penggunaan: /verify USER_ID",
+ "/verify command",
+ )
+ if target_user_id is None:
return
admin_user_id = update.message.from_user.id
- admin_ids = context.bot_data.get("admin_ids", [])
-
- if admin_user_id not in admin_ids:
- await update.message.reply_text("❌ Kamu tidak memiliki izin untuk menggunakan perintah ini.")
- logger.warning(
- f"Non-admin user {admin_user_id} ({update.message.from_user.full_name}) "
- f"attempted to use /verify command"
- )
- return
-
- if context.args and len(context.args) > 0:
- try:
- target_user_id = int(context.args[0])
- except ValueError:
- await update.message.reply_text("❌ User ID harus berupa angka.")
- return
- else:
- await update.message.reply_text("❌ Penggunaan: /verify USER_ID")
- return
db = get_database()
@@ -201,35 +182,16 @@ async def handle_unverify_command(
update: Telegram update containing the command.
context: Bot context with helper methods.
"""
- if not update.message or not update.message.from_user:
- return
-
- if update.effective_chat and update.effective_chat.type != "private":
- await update.message.reply_text(
- "❌ Perintah ini hanya bisa digunakan di chat pribadi dengan bot."
- )
+ target_user_id = await require_admin_dm_target(
+ update,
+ context,
+ "❌ Penggunaan: /unverify USER_ID",
+ "/unverify command",
+ )
+ if target_user_id is None:
return
admin_user_id = update.message.from_user.id
- admin_ids = context.bot_data.get("admin_ids", [])
-
- if admin_user_id not in admin_ids:
- await update.message.reply_text("❌ Kamu tidak memiliki izin untuk menggunakan perintah ini.")
- logger.warning(
- f"Non-admin user {admin_user_id} ({update.message.from_user.full_name}) "
- f"attempted to use /unverify command"
- )
- return
-
- if context.args and len(context.args) > 0:
- try:
- target_user_id = int(context.args[0])
- except ValueError:
- await update.message.reply_text("❌ User ID harus berupa angka.")
- return
- else:
- await update.message.reply_text("❌ Penggunaan: /unverify USER_ID")
- return
db = get_database()
diff --git a/src/bot/plugins/__init__.py b/src/bot/plugins/__init__.py
index 2f8c09e..955e001 100644
--- a/src/bot/plugins/__init__.py
+++ b/src/bot/plugins/__init__.py
@@ -8,20 +8,17 @@
from typing import TYPE_CHECKING
-from bot.plugins.base import PluginProtocol
-from bot.plugins.config import guard_plugin, is_plugin_enabled, is_plugin_enabled_for_group, resolve_plugin_toggles
+from bot.plugins.config import guard_plugin, is_plugin_enabled_for_group, resolve_plugin_toggles
from bot.plugins.definitions import PluginManifest, get_plugin_definitions
if TYPE_CHECKING:
from bot.plugins.manager import compute_effective_plugin_map
__all__ = [
- "PluginProtocol",
"PluginManifest",
"compute_effective_plugin_map",
"get_plugin_definitions",
"guard_plugin",
- "is_plugin_enabled",
"is_plugin_enabled_for_group",
"resolve_plugin_toggles",
]
diff --git a/src/bot/plugins/base.py b/src/bot/plugins/base.py
deleted file mode 100644
index bf5ce24..0000000
--- a/src/bot/plugins/base.py
+++ /dev/null
@@ -1,33 +0,0 @@
-"""Base plugin contracts for the PythonID bot plugin system."""
-
-from __future__ import annotations
-
-from typing import Protocol, runtime_checkable
-
-from telegram.ext import BaseHandler
-
-
-@runtime_checkable
-class PluginProtocol(Protocol):
- """Protocol that all built-in plugins must satisfy.
-
- Attributes:
- name: Canonical plugin identifier (must match KNOWN_PLUGINS).
- description: Human-readable description of plugin purpose.
- handler_group: PTB handler group integer for registration ordering.
- """
-
- name: str
- description: str
- handler_group: int
-
- def register(self, application: object) -> list[BaseHandler]:
- """Register handlers onto the PTB Application.
-
- Args:
- application: PTB Application instance.
-
- Returns:
- List of registered BaseHandler instances.
- """
- ...
\ No newline at end of file
diff --git a/src/bot/plugins/builtin/captcha.py b/src/bot/plugins/builtin/captcha.py
index c103ba3..1d4cb26 100644
--- a/src/bot/plugins/builtin/captcha.py
+++ b/src/bot/plugins/builtin/captcha.py
@@ -41,19 +41,3 @@ def register_captcha(application: Application) -> list[BaseHandler]: # type: ig
registered.append(cloned)
logger.info("Registered handler: captcha_handlers (group=0)")
return registered
-
-# --- Coarse plugin class (keeps existing API) ---
-
-# Coarse plugin class for API compatibility. Unused by PluginManager.
-class _CaptchaPlugin:
- """Plugin wrapper for captcha handlers."""
-
- name: str = "captcha"
- description: str = "Captcha verification for new members"
- handler_group: int = 0
-
- def register(self, application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
- """Register captcha handlers onto application."""
- return register_captcha(application)
-
-plugin = _CaptchaPlugin()
\ No newline at end of file
diff --git a/src/bot/plugins/builtin/commands.py b/src/bot/plugins/builtin/commands.py
index d661ee9..f355baf 100644
--- a/src/bot/plugins/builtin/commands.py
+++ b/src/bot/plugins/builtin/commands.py
@@ -44,54 +44,51 @@
logger = logging.getLogger(__name__)
+# --- Helper for handler registration ---
+
+def _register(application: Application, handler: BaseHandler, label: str) -> list[BaseHandler]: # type: ignore[type-arg]
+ """Register a handler and log the registration."""
+ application.add_handler(handler)
+ logger.info(f"Registered handler: {label} (group=0)")
+ return [handler]
+
+
# --- Individual registrar functions ---
def register_verify(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register /verify command handler."""
handler: BaseHandler = CommandHandler("verify", handle_verify_command)
- application.add_handler(handler)
- logger.info("Registered handler: verify_command (group=0)")
- return [handler]
+ return _register(application, handler, "verify_command")
def register_unverify(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register /unverify command handler."""
handler: BaseHandler = CommandHandler("unverify", handle_unverify_command)
- application.add_handler(handler)
- logger.info("Registered handler: unverify_command (group=0)")
- return [handler]
+ return _register(application, handler, "unverify_command")
def register_check(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register /check command handler."""
handler: BaseHandler = CommandHandler("check", handle_check_command)
- application.add_handler(handler)
- logger.info("Registered handler: check_command (group=0)")
- return [handler]
+ return _register(application, handler, "check_command")
def register_trust(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register /trust command handler."""
handler: BaseHandler = CommandHandler("trust", handle_trust_command)
- application.add_handler(handler)
- logger.info("Registered handler: trust_command (group=0)")
- return [handler]
+ return _register(application, handler, "trust_command")
def register_untrust(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register /untrust command handler."""
handler: BaseHandler = CommandHandler("untrust", handle_untrust_command)
- application.add_handler(handler)
- logger.info("Registered handler: untrust_command (group=0)")
- return [handler]
+ return _register(application, handler, "untrust_command")
def register_trusted_list(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register /trusted command handler."""
handler: BaseHandler = CommandHandler("trusted", handle_trusted_list_command)
- application.add_handler(handler)
- logger.info("Registered handler: trusted_list_command (group=0)")
- return [handler]
+ return _register(application, handler, "trusted_list_command")
def register_check_forwarded_message(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
@@ -100,9 +97,7 @@ def register_check_forwarded_message(application: Application) -> list[BaseHandl
filters.FORWARDED & filters.ChatType.PRIVATE,
handle_check_forwarded_message,
)
- application.add_handler(handler)
- logger.info("Registered handler: check_forwarded_message (group=0)")
- return [handler]
+ return _register(application, handler, "check_forwarded_message")
def register_verify_callback(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
@@ -113,69 +108,28 @@ def register_verify_callback(application: Application) -> list[BaseHandler]: #
# Group IDs can be negative, but callback data only encodes user IDs.
pattern=r"^verify:\d+$",
)
- application.add_handler(handler)
- logger.info("Registered handler: verify_callback (group=0)")
- return [handler]
+ return _register(application, handler, "verify_callback")
def register_unverify_callback(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register unverify callback handler."""
handler: BaseHandler = CallbackQueryHandler(handle_unverify_callback, pattern=r"^unverify:\d+$")
- application.add_handler(handler)
- logger.info("Registered handler: unverify_callback (group=0)")
- return [handler]
+ return _register(application, handler, "unverify_callback")
def register_warn_callback(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register warn callback handler."""
handler: BaseHandler = CallbackQueryHandler(handle_warn_callback, pattern=r"^warn:\d+:")
- application.add_handler(handler)
- logger.info("Registered handler: warn_callback (group=0)")
- return [handler]
+ return _register(application, handler, "warn_callback")
def register_trust_callback(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register trust callback handler."""
handler: BaseHandler = CallbackQueryHandler(handle_trust_callback, pattern=r"^trust:\d+$")
- application.add_handler(handler)
- logger.info("Registered handler: trust_callback (group=0)")
- return [handler]
+ return _register(application, handler, "trust_callback")
def register_untrust_callback(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register untrust callback handler."""
handler: BaseHandler = CallbackQueryHandler(handle_untrust_callback, pattern=r"^untrust:\d+$")
- application.add_handler(handler)
- logger.info("Registered handler: untrust_callback (group=0)")
- return [handler]
-
-
-# --- Coarse plugin class (keeps existing API) ---
-
-# Coarse plugin class for API compatibility. Unused by PluginManager.
-class _CommandsPlugin:
- """Plugin wrapper for command and callback handlers."""
-
- name: str = "commands"
- description: str = "Admin commands and callback handlers (verify, unverify, check, trust)"
- handler_group: int = 0
-
- def register(self, application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
- """Register all command and callback handlers onto application."""
- handlers: list[BaseHandler] = []
- handlers.extend(register_verify(application))
- handlers.extend(register_unverify(application))
- handlers.extend(register_check(application))
- handlers.extend(register_trust(application))
- handlers.extend(register_untrust(application))
- handlers.extend(register_trusted_list(application))
- handlers.extend(register_check_forwarded_message(application))
- handlers.extend(register_verify_callback(application))
- handlers.extend(register_unverify_callback(application))
- handlers.extend(register_warn_callback(application))
- handlers.extend(register_trust_callback(application))
- handlers.extend(register_untrust_callback(application))
- return handlers
-
-
-plugin = _CommandsPlugin()
\ No newline at end of file
+ return _register(application, handler, "untrust_callback")
diff --git a/src/bot/plugins/builtin/dm.py b/src/bot/plugins/builtin/dm.py
index d4835c7..0d936cb 100644
--- a/src/bot/plugins/builtin/dm.py
+++ b/src/bot/plugins/builtin/dm.py
@@ -33,21 +33,3 @@ def register_dm(application: Application) -> list[BaseHandler]: # type: ignore[
application.add_handler(handler)
logger.info("Registered handler: dm_handler (group=0)")
return [handler]
-
-
-# --- Coarse plugin class (keeps existing API) ---
-
-# Coarse plugin class for API compatibility. Unused by PluginManager.
-class _DmPlugin:
- """Plugin wrapper for DM handler."""
-
- name: str = "dm"
- description: str = "Direct message unrestriction flow"
- handler_group: int = 0
-
- def register(self, application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
- """Register DM handler onto application."""
- return register_dm(application)
-
-
-plugin = _DmPlugin()
\ No newline at end of file
diff --git a/src/bot/plugins/builtin/jobs.py b/src/bot/plugins/builtin/jobs.py
index 662efb3..8f81a72 100644
--- a/src/bot/plugins/builtin/jobs.py
+++ b/src/bot/plugins/builtin/jobs.py
@@ -49,22 +49,3 @@ def register_refresh_admin_ids_job(application: Application) -> list[BaseHandler
logger.info("JobQueue registered: refresh_admin_ids_job (every 10 minutes)")
# Jobs don't return handler objects
return handlers
-
-# --- Coarse plugin class (keeps existing API) ---
-
-# Coarse plugin class for API compatibility. Unused by PluginManager.
-class _JobsPlugin:
- """Plugin wrapper for periodic job handlers."""
-
- name: str = "jobs"
- description: str = "Periodic JobQueue tasks (auto-restrict, admin cache refresh)"
- handler_group: int = 6
-
- def register(self, application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
- """Register repeating jobs onto application.job_queue."""
- handlers: list[BaseHandler] = []
- handlers.extend(register_auto_restrict_job(application))
- handlers.extend(register_refresh_admin_ids_job(application))
- return handlers
-
-plugin = _JobsPlugin()
diff --git a/src/bot/plugins/builtin/profile_monitor.py b/src/bot/plugins/builtin/profile_monitor.py
index 6a13bfc..aac6f83 100644
--- a/src/bot/plugins/builtin/profile_monitor.py
+++ b/src/bot/plugins/builtin/profile_monitor.py
@@ -38,19 +38,3 @@ def register_profile_monitor(application: Application) -> list[BaseHandler]: #
application.add_handler(handler, group=5)
logger.info("Registered handler: message_handler (group=5)")
return [handler]
-
-# --- Coarse plugin class (keeps existing API) ---
-
-# Coarse plugin class for API compatibility. Unused by PluginManager.
-class _ProfileMonitorPlugin:
- """Plugin wrapper for profile compliance monitor."""
-
- name: str = "profile_monitor"
- description: str = "Profile compliance monitoring"
- handler_group: int = 5
-
- def register(self, application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
- """Register profile monitor handler onto application."""
- return register_profile_monitor(application)
-
-plugin = _ProfileMonitorPlugin()
\ No newline at end of file
diff --git a/src/bot/plugins/builtin/spam.py b/src/bot/plugins/builtin/spam.py
index ce94bc1..7cc75c3 100644
--- a/src/bot/plugins/builtin/spam.py
+++ b/src/bot/plugins/builtin/spam.py
@@ -26,6 +26,15 @@
logger = logging.getLogger(__name__)
+# --- Helper for spam handler registration ---
+
+def _register_spam(application: Application, handler: BaseHandler, group: int, label: str) -> list[BaseHandler]: # type: ignore[type-arg]
+ """Register a spam handler at a specific group and log the registration."""
+ application.add_handler(handler, group=group)
+ logger.info(f"Registered handler: {label} (group={group})")
+ return [handler]
+
+
# --- Individual registrar functions ---
def register_inline_keyboard_spam(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
@@ -37,9 +46,7 @@ def register_inline_keyboard_spam(application: Application) -> list[BaseHandler]
filters.ChatType.GROUPS,
guard_plugin("inline_keyboard_spam")(handle_inline_keyboard_spam),
)
- application.add_handler(handler, group=1)
- logger.info("Registered handler: inline_keyboard_spam_handler (group=1)")
- return [handler]
+ return _register_spam(application, handler, 1, "inline_keyboard_spam_handler")
def register_bio_bait_spam(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register bio bait spam handler (group=4).
@@ -50,9 +57,7 @@ def register_bio_bait_spam(application: Application) -> list[BaseHandler]: # ty
BIO_BAIT_FILTER,
guard_plugin("bio_bait_spam")(handle_bio_bait_spam),
)
- application.add_handler(handler, group=4)
- logger.info("Registered handler: bio_bait_spam_handler (group=4)")
- return [handler]
+ return _register_spam(application, handler, 4, "bio_bait_spam_handler")
def register_contact_spam(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register contact spam handler (group=2).
@@ -63,9 +68,7 @@ def register_contact_spam(application: Application) -> list[BaseHandler]: # typ
filters.ChatType.GROUPS & filters.CONTACT,
guard_plugin("contact_spam")(handle_contact_spam),
)
- application.add_handler(handler, group=2)
- logger.info("Registered handler: contact_spam_handler (group=2)")
- return [handler]
+ return _register_spam(application, handler, 2, "contact_spam_handler")
def register_new_user_spam(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register new user spam handler (probation, group=3).
@@ -76,9 +79,7 @@ def register_new_user_spam(application: Application) -> list[BaseHandler]: # ty
filters.ChatType.GROUPS,
guard_plugin("new_user_spam")(handle_new_user_spam),
)
- application.add_handler(handler, group=3)
- logger.info("Registered handler: anti_spam_handler (group=3)")
- return [handler]
+ return _register_spam(application, handler, 3, "anti_spam_handler")
def register_duplicate_spam(application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
"""Register duplicate message spam handler (group=4).
@@ -89,28 +90,4 @@ def register_duplicate_spam(application: Application) -> list[BaseHandler]: # t
filters.ChatType.GROUPS & ~filters.COMMAND,
guard_plugin("duplicate_spam")(handle_duplicate_spam),
)
- application.add_handler(handler, group=4)
- logger.info("Registered handler: duplicate_spam_handler (group=4)")
- return [handler]
-
-# --- Coarse plugin class (keeps existing API) ---
-
-# Coarse plugin class for API compatibility. Unused by PluginManager.
-class _SpamPlugin:
- """Plugin wrapper for all anti-spam handlers."""
-
- name: str = "spam"
- description: str = "Anti-spam handlers (inline keyboards, bio bait, contact, probation, duplicates)"
- handler_group: int = 1
-
- def register(self, application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
- """Register all spam handlers onto application with their respective groups."""
- handlers: list[BaseHandler] = []
- handlers.extend(register_inline_keyboard_spam(application))
- handlers.extend(register_bio_bait_spam(application))
- handlers.extend(register_contact_spam(application))
- handlers.extend(register_new_user_spam(application))
- handlers.extend(register_duplicate_spam(application))
- return handlers
-
-plugin = _SpamPlugin()
\ No newline at end of file
+ return _register_spam(application, handler, 4, "duplicate_spam_handler")
diff --git a/src/bot/plugins/builtin/topic_guard.py b/src/bot/plugins/builtin/topic_guard.py
index ede39dc..b56a96a 100644
--- a/src/bot/plugins/builtin/topic_guard.py
+++ b/src/bot/plugins/builtin/topic_guard.py
@@ -38,19 +38,3 @@ def register_topic_guard(application: Application) -> list[BaseHandler]: # type
application.add_handler(handler, group=-1)
logger.info("Registered handler: topic_guard (group=-1, message + edited_message)")
return [handler]
-
-# --- Coarse plugin class (keeps existing API) ---
-
-# Coarse plugin class for API compatibility. Unused by PluginManager.
-class _TopicGuardPlugin:
- """Plugin wrapper for topic_guard handler."""
-
- name: str = "topic_guard"
- description: str = "Intercept warning-topic messages before other handlers"
- handler_group: int = -1
-
- def register(self, application: Application) -> list[BaseHandler]: # type: ignore[type-arg]
- """Register topic_guard handler onto application."""
- return register_topic_guard(application)
-
-plugin = _TopicGuardPlugin()
\ No newline at end of file
diff --git a/src/bot/plugins/config.py b/src/bot/plugins/config.py
index a84a06f..f1c94b4 100644
--- a/src/bot/plugins/config.py
+++ b/src/bot/plugins/config.py
@@ -53,21 +53,6 @@ def resolve_plugin_toggles(
return result
-def is_plugin_enabled(toggles: dict[str, bool], name: str) -> bool:
- """Check if a single plugin is enabled from a resolved toggle dict.
-
- Args:
- toggles: Resolved toggle dict from ``resolve_plugin_toggles``.
- name: Plugin name to check.
-
- Returns:
- True if plugin is enabled.
-
- Raises:
- KeyError: If ``name`` is not in ``toggles``.
- """
- return toggles[name]
-
def is_plugin_enabled_for_group(
effective_map: dict[int, dict[str, bool]],
group_id: int,
diff --git a/src/bot/plugins/definitions.py b/src/bot/plugins/definitions.py
index a454ece..aa24825 100644
--- a/src/bot/plugins/definitions.py
+++ b/src/bot/plugins/definitions.py
@@ -41,15 +41,6 @@
{"name": "refresh_admin_ids_job", "handler_group": 6, "description": "Periodic admin cache refresh job (every 10 min)"},
]
-# Admin command plugins that intentionally skip guard_plugin wrapping.
-# These handlers must work in every group regardless of plugin toggle state.
-# Documented here to make the architecture decision explicit.
-ADMIN_COMMANDS: frozenset[str] = frozenset({
- "verify", "unverify", "check", "trust", "untrust", "trusted_list",
- "check_forwarded_message", "verify_callback", "unverify_callback",
- "warn_callback", "trust_callback", "untrust_callback",
-})
-
# Single source of truth: canonical set of all known plugin names.
PLUGIN_NAMES: frozenset[str] = frozenset(d["name"] for d in _PLUGIN_DEFINITIONS) # type: ignore[arg-type]
"""Canonical set of all known built-in plugin names.
diff --git a/src/bot/plugins/manager.py b/src/bot/plugins/manager.py
index 545eb82..fc7ebc9 100644
--- a/src/bot/plugins/manager.py
+++ b/src/bot/plugins/manager.py
@@ -44,6 +44,40 @@
# Use string forward ref because BaseHandler is only imported under TYPE_CHECKING.
Registrar = Callable[..., list["BaseHandler"]]
+# Module-level registry constant mapping plugin names to registrar functions.
+_REGISTRY: dict[str, Registrar] = {
+ # topic_guard
+ "topic_guard": tg_mod.register_topic_guard,
+ # commands (group=0)
+ "verify": commands.register_verify,
+ "unverify": commands.register_unverify,
+ "check": commands.register_check,
+ "trust": commands.register_trust,
+ "untrust": commands.register_untrust,
+ "trusted_list": commands.register_trusted_list,
+ "check_forwarded_message": commands.register_check_forwarded_message,
+ "verify_callback": commands.register_verify_callback,
+ "unverify_callback": commands.register_unverify_callback,
+ "warn_callback": commands.register_warn_callback,
+ "trust_callback": commands.register_trust_callback,
+ "untrust_callback": commands.register_untrust_callback,
+ # captcha
+ "captcha": captcha_mod.register_captcha,
+ # dm
+ "dm": dm_mod.register_dm,
+ # spam
+ "inline_keyboard_spam": spam_mod.register_inline_keyboard_spam,
+ "bio_bait_spam": spam_mod.register_bio_bait_spam,
+ "contact_spam": spam_mod.register_contact_spam,
+ "new_user_spam": spam_mod.register_new_user_spam,
+ "duplicate_spam": spam_mod.register_duplicate_spam,
+ # profile_monitor
+ "profile_monitor": pm_mod.register_profile_monitor,
+ # jobs
+ "auto_restrict_job": jobs_mod.register_auto_restrict_job,
+ "refresh_admin_ids_job": jobs_mod.register_refresh_admin_ids_job,
+}
+
def compute_effective_plugin_map(
plugins_default: dict[str, bool],
@@ -88,48 +122,8 @@ class PluginManager:
"""
def __init__(self) -> None:
- # Build registry: manifest name -> registrar callable
- self._registry: dict[str, Registrar] = self._build_registry()
-
- @staticmethod
- def _build_registry() -> dict[str, Registrar]:
- """Return dict mapping each manifest name to its registrar.
-
- Each registrar is a module-level function that accepts an
- ``Application`` and returns ``list[BaseHandler]``.
- """
- return {
- # topic_guard
- "topic_guard": tg_mod.register_topic_guard,
- # commands (group=0)
- "verify": commands.register_verify,
- "unverify": commands.register_unverify,
- "check": commands.register_check,
- "trust": commands.register_trust,
- "untrust": commands.register_untrust,
- "trusted_list": commands.register_trusted_list,
- "check_forwarded_message": commands.register_check_forwarded_message,
- "verify_callback": commands.register_verify_callback,
- "unverify_callback": commands.register_unverify_callback,
- "warn_callback": commands.register_warn_callback,
- "trust_callback": commands.register_trust_callback,
- "untrust_callback": commands.register_untrust_callback,
- # captcha
- "captcha": captcha_mod.register_captcha,
- # dm
- "dm": dm_mod.register_dm,
- # spam
- "inline_keyboard_spam": spam_mod.register_inline_keyboard_spam,
- "bio_bait_spam": spam_mod.register_bio_bait_spam,
- "contact_spam": spam_mod.register_contact_spam,
- "new_user_spam": spam_mod.register_new_user_spam,
- "duplicate_spam": spam_mod.register_duplicate_spam,
- # profile_monitor
- "profile_monitor": pm_mod.register_profile_monitor,
- # jobs
- "auto_restrict_job": jobs_mod.register_auto_restrict_job,
- "refresh_admin_ids_job": jobs_mod.register_refresh_admin_ids_job,
- }
+ """Initialize with a shallow copy of the shared registry."""
+ self._registry: dict[str, Registrar] = dict(_REGISTRY)
def register_all(
self,
diff --git a/src/bot/services/admin_cache.py b/src/bot/services/admin_cache.py
index 92a1435..287cb42 100644
--- a/src/bot/services/admin_cache.py
+++ b/src/bot/services/admin_cache.py
@@ -19,15 +19,19 @@
logger = logging.getLogger(__name__)
-async def refresh_admin_ids(context: ContextTypes.DEFAULT_TYPE) -> None:
+
+async def _sync_admin_ids(context: ContextTypes.DEFAULT_TYPE, *, seed_existing: bool) -> None:
"""
- Periodically refresh cached admin IDs for all monitored groups.
+ Sync admin IDs for all monitored groups with fallback to cached data.
- Called by JobQueue every 10 minutes to keep admin rosters up to date
- when promotions/demotions happen after startup.
+ Args:
+ context: Bot context.
+ seed_existing: If True, seed the dict from existing bot_data cache;
+ if False, start with an empty dict.
"""
registry = get_group_registry()
- group_admin_ids: dict[int, list[int]] = {}
+ old_cache: dict[int, list[int]] = context.bot_data.get("group_admin_ids", {})
+ group_admin_ids: dict[int, list[int]] = dict(old_cache) if seed_existing else {}
all_admin_ids: set[int] = set()
for gc in registry.all_groups():
@@ -36,15 +40,28 @@ async def refresh_admin_ids(context: ContextTypes.DEFAULT_TYPE) -> None:
group_admin_ids[gc.group_id] = ids
all_admin_ids.update(ids)
except TelegramAdminFetchError as e:
- logger.error(f"Failed to refresh admin IDs for group {gc.group_id}: {e}")
- existing = context.bot_data.get("group_admin_ids", {}).get(gc.group_id, [])
+ logger.error(f"Failed to fetch admin IDs for group {gc.group_id}: {e}")
+ existing = old_cache.get(gc.group_id, [])
group_admin_ids[gc.group_id] = existing
all_admin_ids.update(existing)
context.bot_data["group_admin_ids"] = group_admin_ids
context.bot_data["admin_ids"] = list(all_admin_ids)
+
+
+async def refresh_admin_ids(context: ContextTypes.DEFAULT_TYPE) -> None:
+ """
+ Periodically refresh cached admin IDs for all monitored groups.
+
+ Called by JobQueue every 10 minutes to keep admin rosters up to date
+ when promotions/demotions happen after startup.
+ """
+ await _sync_admin_ids(context, seed_existing=False)
+ group_admin_ids = context.bot_data.get("group_admin_ids", {})
+ all_admin_ids = context.bot_data.get("admin_ids", [])
logger.info(f"Refreshed admin IDs: {len(all_admin_ids)} unique admin(s) across {len(group_admin_ids)} group(s)")
+
async def preload_admin_ids(context: ContextTypes.DEFAULT_TYPE) -> None:
"""
Preload admin IDs at startup with fallback to existing cache.
@@ -54,23 +71,9 @@ async def preload_admin_ids(context: ContextTypes.DEFAULT_TYPE) -> None:
to fetch. Used in ``post_init`` to prevent wiping admin cache on
startup failures.
"""
- registry = get_group_registry()
- group_admin_ids: dict[int, list[int]] = dict(context.bot_data.get("group_admin_ids", {}))
- all_admin_ids: set[int] = set()
-
- for gc in registry.all_groups():
- try:
- ids = await fetch_group_admin_ids(context.bot, gc.group_id)
- group_admin_ids[gc.group_id] = ids
- all_admin_ids.update(ids)
- except TelegramAdminFetchError as e:
- logger.error(f"Failed to fetch admin IDs for group {gc.group_id}: {e}")
- existing = group_admin_ids.get(gc.group_id, [])
- group_admin_ids[gc.group_id] = existing
- all_admin_ids.update(existing)
-
- context.bot_data["group_admin_ids"] = group_admin_ids
- context.bot_data["admin_ids"] = list(all_admin_ids)
+ await _sync_admin_ids(context, seed_existing=True)
+ group_admin_ids = context.bot_data.get("group_admin_ids", {})
+ all_admin_ids = context.bot_data.get("admin_ids", [])
logger.info(
f"Preloaded admin IDs: {len(all_admin_ids)} unique admin(s) "
f"across {len(group_admin_ids)} group(s)"
diff --git a/src/bot/services/telegram_utils.py b/src/bot/services/telegram_utils.py
index e98ca33..99e9387 100644
--- a/src/bot/services/telegram_utils.py
+++ b/src/bot/services/telegram_utils.py
@@ -8,9 +8,10 @@
import logging
from urllib.parse import urlparse
-from telegram import Bot, Chat, Message, User
+from telegram import Bot, Chat, Message, Update, User
from telegram.constants import ChatMemberStatus
from telegram.error import BadRequest, Forbidden
+from telegram.ext import ContextTypes
from telegram.helpers import escape_markdown, mention_markdown
from bot.constants import WHITELISTED_TELEGRAM_PATHS, WHITELISTED_URL_DOMAINS
@@ -35,10 +36,7 @@ def get_user_mention(user: User | Chat) -> str:
Returns:
str: Formatted user mention (either @username or markdown mention).
"""
- if user.username:
- escaped = escape_markdown(user.username.lstrip("@"), version=1)
- return f"@{escaped}"
- return mention_markdown(user.id, user.full_name, version=1)
+ return get_user_mention_by_id(user.id, user.full_name, user.username)
def get_user_mention_by_id(
user_id: int,
@@ -279,3 +277,61 @@ async def fetch_group_admin_ids(bot: Bot, group_id: int) -> list[int]:
exc_info=True,
)
raise TelegramAdminFetchError(f"Failed to fetch admins from group {group_id}: {e}") from e
+
+async def require_admin_dm_target(
+ update: Update,
+ context: ContextTypes.DEFAULT_TYPE,
+ usage_message: str,
+ command_label: str,
+) -> int | None:
+ """
+ Validate admin DM command prerequisites and parse target user ID.
+
+ Performs four checks in order:
+ 1. Message and from_user exist
+ 2. Chat is private
+ 3. Caller is an admin
+ 4. Argument is a valid user ID
+
+ Sends appropriate error replies on any failure. Returns the parsed user ID
+ on success, or None after sending an error reply.
+
+ Args:
+ update: Telegram update.
+ context: Bot context.
+ usage_message: Full usage message to send on missing args (e.g., "❌ Penggunaan: /verify USER_ID").
+ command_label: Command name for logging (e.g., "/verify command").
+
+ Returns:
+ int | None: Parsed user ID on success, None if any check failed.
+ """
+ if not update.message or not update.message.from_user:
+ return None
+
+ if update.effective_chat and update.effective_chat.type != "private":
+ await update.message.reply_text(
+ "❌ Perintah ini hanya bisa digunakan di chat pribadi dengan bot."
+ )
+ return None
+
+ admin_user_id = update.message.from_user.id
+ admin_ids = context.bot_data.get("admin_ids", [])
+
+ if admin_user_id not in admin_ids:
+ await update.message.reply_text("❌ Kamu tidak memiliki izin untuk menggunakan perintah ini.")
+ logger.warning(
+ f"Non-admin user {admin_user_id} ({update.message.from_user.full_name}) "
+ f"attempted to use {command_label}"
+ )
+ return None
+
+ if not context.args or len(context.args) == 0:
+ await update.message.reply_text(usage_message)
+ return None
+
+ try:
+ target_user_id = int(context.args[0])
+ return target_user_id
+ except ValueError:
+ await update.message.reply_text("❌ User ID harus berupa angka.")
+ return None
diff --git a/tests/test_anti_spam.py b/tests/test_anti_spam.py
index e84f075..7360932 100644
--- a/tests/test_anti_spam.py
+++ b/tests/test_anti_spam.py
@@ -16,7 +16,6 @@
handle_new_user_spam,
has_contact,
has_external_reply,
- has_link,
has_media,
has_non_whitelisted_inline_keyboard_urls,
has_non_whitelisted_link,
@@ -44,62 +43,6 @@ def test_regular_message_not_forwarded(self):
assert is_forwarded(msg) is False
-class TestHasLink:
- """Tests for the has_link helper function."""
-
- def test_url_entity_detected(self):
- """Test that URL entity is detected."""
- entity = MagicMock(spec=MessageEntity)
- entity.type = MessageEntity.URL
-
- msg = MagicMock(spec=Message)
- msg.entities = [entity]
- msg.caption_entities = None
-
- assert has_link(msg) is True
-
- def test_text_link_entity_detected(self):
- """Test that TEXT_LINK entity is detected."""
- entity = MagicMock(spec=MessageEntity)
- entity.type = MessageEntity.TEXT_LINK
-
- msg = MagicMock(spec=Message)
- msg.entities = [entity]
- msg.caption_entities = None
-
- assert has_link(msg) is True
-
- def test_caption_link_detected(self):
- """Test that link in caption is detected."""
- entity = MagicMock(spec=MessageEntity)
- entity.type = MessageEntity.URL
-
- msg = MagicMock(spec=Message)
- msg.entities = None
- msg.caption_entities = [entity]
-
- assert has_link(msg) is True
-
- def test_no_link_returns_false(self):
- """Test that message without links returns False."""
- msg = MagicMock(spec=Message)
- msg.entities = None
- msg.caption_entities = None
-
- assert has_link(msg) is False
-
- def test_other_entity_not_link(self):
- """Test that non-link entities are not detected as links."""
- entity = MagicMock(spec=MessageEntity)
- entity.type = MessageEntity.BOLD
-
- msg = MagicMock(spec=Message)
- msg.entities = [entity]
- msg.caption_entities = None
-
- assert has_link(msg) is False
-
-
class TestHasExternalReply:
"""Tests for the has_external_reply helper function."""
diff --git a/tests/test_bio_bait.py b/tests/test_bio_bait.py
index e2eb490..065eb87 100644
--- a/tests/test_bio_bait.py
+++ b/tests/test_bio_bait.py
@@ -9,7 +9,6 @@
from bot.group_config import GroupConfig
from bot.handlers.bio_bait import (
BIO_BAIT_MAX_LENGTH,
- BIO_BAIT_METRICS_KEY,
USER_BIO_CACHE_KEY,
USER_BIO_CACHE_MAX_SIZE,
USER_BIO_CACHE_TTL_SECONDS,
@@ -471,7 +470,7 @@ async def test_notification_failure_still_raises_stop(
with pytest.raises(ApplicationHandlerStop):
await handle_bio_bait_spam(mock_update, mock_context)
- async def test_monitor_only_collects_metrics_and_sends_owner_alert(
+ async def test_monitor_only_sends_owner_alert(
self, mock_update, mock_context, group_config
):
group_config.bio_bait_monitor_only = True
@@ -490,30 +489,6 @@ async def test_monitor_only_collects_metrics_and_sends_owner_alert(
assert "message_thread_id" not in kwargs
assert "cek bio aku" in kwargs["text"]
- metrics = mock_context.bot_data[BIO_BAIT_METRICS_KEY]
- assert metrics["detections_total"] == 1
- assert metrics["detections_message_bait"] == 1
- assert metrics["monitor_only_matches"] == 1
- assert metrics["owner_alert_sent"] == 1
-
- async def test_monitor_only_alert_failure_still_collects_metrics(
- self, mock_update, mock_context, group_config
- ):
- group_config.bio_bait_monitor_only = True
- group_config.bio_bait_alert_chat_id = 57747812
- mock_context.bot.send_message = AsyncMock(side_effect=Exception("Send failed"))
-
- with patch("bot.handlers.bio_bait.get_group_config_for_update", return_value=group_config):
- await handle_bio_bait_spam(mock_update, mock_context)
-
- mock_update.message.delete.assert_not_called()
- mock_context.bot.restrict_chat_member.assert_not_called()
-
- metrics = mock_context.bot_data[BIO_BAIT_METRICS_KEY]
- assert metrics["detections_total"] == 1
- assert metrics["monitor_only_matches"] == 1
- assert metrics["owner_alert_failed"] == 1
-
class TestBioBaitReviewFixes:
"""Tests for pending bio-bait review fixes (trusted bypass, monitor semantics,
warning-topic guard, metrics)."""
@@ -589,17 +564,15 @@ async def test_enforcement_mode_alert_chat_id_does_not_send_owner_alert(
kwargs = mock_context.bot.send_message.call_args.kwargs
assert kwargs.get("message_thread_id") == 999 # warning topic
- metrics = mock_context.bot_data.get(BIO_BAIT_METRICS_KEY, {})
- assert "owner_alert_sent" not in metrics
- assert "owner_alert_failed" not in metrics
-
# ── (c) monitor-only + alert target = same group => alert skipped ──
+ # ── (d) enforcement mode (no metrics tracking) ──
+
async def test_monitor_only_alert_target_same_group_skipped(
self, mock_update, mock_context, group_config
):
"""When monitor-only and alert_chat_id equals the monitored group,
- owner alert should be skipped and skip metric incremented."""
+ owner alert should be skipped."""
group_config.bio_bait_monitor_only = True
group_config.bio_bait_alert_chat_id = -1001234567890 # same as group_id
mock_update.message.text = "cek bio aku"
@@ -612,49 +585,6 @@ async def test_monitor_only_alert_target_same_group_skipped(
mock_update.message.delete.assert_not_called()
mock_context.bot.restrict_chat_member.assert_not_called()
- metrics = mock_context.bot_data[BIO_BAIT_METRICS_KEY]
- assert metrics["owner_alert_skipped_warning_topic"] == 1
- assert "owner_alert_sent" not in metrics
- assert "owner_alert_failed" not in metrics
-
- # ── (d) enforcement metrics ──
-
- async def test_enforcement_message_bait_records_enforced_matches(
- self, mock_update, mock_context, group_config
- ):
- """Enforcement mode with message bait should record enforced_matches metric."""
- group_config.bio_bait_monitor_only = False
- mock_update.message.text = "cek bio aku"
-
- with patch("bot.handlers.bio_bait.get_group_config_for_update", return_value=group_config):
- with pytest.raises(ApplicationHandlerStop):
- await handle_bio_bait_spam(mock_update, mock_context)
-
- metrics = mock_context.bot_data[BIO_BAIT_METRICS_KEY]
- assert metrics["detections_total"] == 1
- assert metrics["detections_message_bait"] == 1
- assert metrics["enforced_matches"] == 1
-
- async def test_enforcement_bio_links_records_enforced_matches_and_bio_links(
- self, mock_update, mock_context, group_config
- ):
- """Enforcement mode with bio_links detection should record enforced_matches
- and detections_bio_links."""
- group_config.bio_bait_monitor_only = False
- mock_update.message.text = "halo"
- chat = MagicMock()
- chat.bio = "VIP t.me/+exampleinvitehash777xyz"
- mock_context.bot.get_chat = AsyncMock(return_value=chat)
-
- with patch("bot.handlers.bio_bait.get_group_config_for_update", return_value=group_config):
- with pytest.raises(ApplicationHandlerStop):
- await handle_bio_bait_spam(mock_update, mock_context)
-
- metrics = mock_context.bot_data[BIO_BAIT_METRICS_KEY]
- assert metrics["detections_total"] == 1
- assert metrics["detections_bio_links"] == 1
- assert metrics["enforced_matches"] == 1
-
class TestBioBaitRegistrationFilter:
"""Tests for bio-bait handler filter registration.
diff --git a/tests/test_config.py b/tests/test_config.py
index 0e03697..fcddbf7 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -1,7 +1,5 @@
"""Tests for the config module."""
-from datetime import timedelta
-
import pytest
from pydantic_settings.exceptions import SettingsError
@@ -32,7 +30,7 @@ def test_unknown_environment_falls_back_to_default(self, monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
tmp_path.joinpath(".env").touch()
assert get_env_file() == ".env"
-
+
def test_no_env_file_returns_none(self, monkeypatch, tmp_path):
monkeypatch.delenv("BOT_ENV", raising=False)
monkeypatch.chdir(tmp_path)
@@ -92,39 +90,6 @@ def test_logfire_environment_defaults_to_production(self, monkeypatch):
assert settings.logfire_environment == "production"
- def test_probation_timedelta(self, monkeypatch):
- """Test probation_timedelta property returns correct timedelta."""
- monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "test_token")
- monkeypatch.setenv("GROUP_ID", "-100999")
- monkeypatch.setenv("WARNING_TOPIC_ID", "1")
- monkeypatch.setenv("NEW_USER_PROBATION_HOURS", "48")
-
- settings = Settings(_env_file=None)
-
- assert settings.probation_timedelta == timedelta(hours=48)
-
- def test_warning_time_threshold_timedelta(self, monkeypatch):
- """Test warning_time_threshold_timedelta property returns correct timedelta."""
- monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "test_token")
- monkeypatch.setenv("GROUP_ID", "-100999")
- monkeypatch.setenv("WARNING_TOPIC_ID", "1")
- monkeypatch.setenv("WARNING_TIME_THRESHOLD_MINUTES", "60")
-
- settings = Settings(_env_file=None)
-
- assert settings.warning_time_threshold_timedelta == timedelta(minutes=60)
-
- def test_captcha_timeout_timedelta(self, monkeypatch):
- """Test captcha_timeout_timedelta property returns correct timedelta."""
- monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "test_token")
- monkeypatch.setenv("GROUP_ID", "-100999")
- monkeypatch.setenv("WARNING_TOPIC_ID", "1")
- monkeypatch.setenv("CAPTCHA_TIMEOUT_SECONDS", "90")
-
- settings = Settings(_env_file=None)
-
- assert settings.captcha_timeout_timedelta == timedelta(seconds=90)
-
def test_duplicate_spam_defaults(self, monkeypatch):
"""Test that duplicate_spam fields have correct defaults."""
monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "test_token")
@@ -407,4 +372,4 @@ def test_warning_time_threshold_must_be_positive(self, monkeypatch):
monkeypatch.setenv("WARNING_TIME_THRESHOLD_MINUTES", "0")
with pytest.raises(ValueError, match="warning_time_threshold_minutes must be greater than 0"):
- Settings(_env_file=None)
\ No newline at end of file
+ Settings(_env_file=None)
diff --git a/tests/test_database.py b/tests/test_database.py
index fbf5c11..dd9c92d 100644
--- a/tests/test_database.py
+++ b/tests/test_database.py
@@ -150,62 +150,6 @@ def test_does_nothing_for_non_existent_record(self, db_service):
db_service.mark_user_unrestricted(user_id=999, group_id=-100999)
-class TestGetWarningsPastTimeThreshold:
- def test_returns_expired_warnings(self, db_service):
- # Create record with old timestamp
- old_time = datetime.now(UTC) - timedelta(minutes=1500) # 25 hours
- record = db_service.get_or_create_user_warning(user_id=123, group_id=-100999)
- # Manually update to simulate old warning (in real tests, would mock time)
- from sqlmodel import Session, select
-
- with Session(db_service._engine) as session:
- stmt = select(UserWarning).where(UserWarning.id == record.id)
- db_record = session.exec(stmt).first()
- db_record.first_warned_at = old_time
- session.add(db_record)
- session.commit()
-
- # Should find the expired warning (1440 minutes = 24 hours)
- expired = db_service.get_warnings_past_time_threshold(timedelta(minutes=1440))
- assert len(expired) == 1
- assert expired[0].user_id == 123
-
- def test_ignores_recent_warnings(self, db_service):
- # Create record with recent timestamp (default)
- db_service.get_or_create_user_warning(user_id=123, group_id=-100999)
-
- # Should not find recent warnings
- expired = db_service.get_warnings_past_time_threshold(timedelta(minutes=1440))
- assert len(expired) == 0
-
- def test_ignores_restricted_users(self, db_service):
- # Create and restrict a user
- db_service.get_or_create_user_warning(user_id=123, group_id=-100999)
- db_service.mark_user_restricted(user_id=123, group_id=-100999)
-
- # Should not find restricted users even if old
- expired = db_service.get_warnings_past_time_threshold(timedelta(minutes=0))
- assert len(expired) == 0
-
- def test_returns_multiple_expired_warnings(self, db_service):
- # Create multiple old records
- for user_id in [123, 456, 789]:
- record = db_service.get_or_create_user_warning(
- user_id=user_id, group_id=-100999
- )
- from sqlmodel import Session, select
-
- with Session(db_service._engine) as session:
- stmt = select(UserWarning).where(UserWarning.id == record.id)
- db_record = session.exec(stmt).first()
- db_record.first_warned_at = datetime.now(UTC) - timedelta(minutes=1500)
- session.add(db_record)
- session.commit()
-
- expired = db_service.get_warnings_past_time_threshold(timedelta(minutes=1440))
- assert len(expired) == 3
-
-
class TestDeleteUserWarnings:
def test_delete_user_warnings(self, db_service):
# Create multiple warnings for same user
@@ -503,42 +447,13 @@ def test_update_trusted_user_names_destructive_overwrite(
assert record.admin_full_name == ""
assert record.admin_username is None
- def test_is_user_trusted_non_zero_group_raises(
- self, db_service: DatabaseService
- ):
- with pytest.raises(NotImplementedError):
- db_service.is_user_trusted(user_id=123, group_id=123)
-
- def test_get_trusted_user_ids_non_zero_group_raises(
- self, db_service: DatabaseService
- ):
- with pytest.raises(NotImplementedError):
- db_service.get_trusted_user_ids(group_id=123)
-
- def test_get_trusted_users_non_zero_group_raises(
- self, db_service: DatabaseService
- ):
- with pytest.raises(NotImplementedError):
- db_service.get_trusted_users(group_id=123)
-
- def test_trusted_user_reads_none_and_zero_group_equivalent(
- self, db_service: DatabaseService
- ):
+ def test_trusted_user_reads(self, db_service: DatabaseService):
db_service.add_trusted_user(user_id=3001, trusted_by_admin_id=9001)
db_service.add_trusted_user(user_id=3002, trusted_by_admin_id=9002)
- assert db_service.is_user_trusted(
- user_id=3001, group_id=None
- ) == db_service.is_user_trusted(user_id=3001, group_id=0)
- assert db_service.is_user_trusted(
- user_id=3001, group_id=None
- ) is True
-
- assert db_service.get_trusted_user_ids(
- group_id=None
- ) == db_service.get_trusted_user_ids(group_id=0)
- assert db_service.get_trusted_user_ids(group_id=None) == {3001, 3002}
-
- users_none = db_service.get_trusted_users(group_id=None)
- users_zero = db_service.get_trusted_users(group_id=0)
- assert [u.user_id for u in users_none] == [u.user_id for u in users_zero]
+ assert db_service.is_user_trusted(user_id=3001) is True
+
+ assert db_service.get_trusted_user_ids() == {3001, 3002}
+
+ users = db_service.get_trusted_users()
+ assert [u.user_id for u in users] == [3002, 3001]
diff --git a/tests/test_group_config.py b/tests/test_group_config.py
index 87de5e8..ed251c5 100644
--- a/tests/test_group_config.py
+++ b/tests/test_group_config.py
@@ -149,13 +149,6 @@ def test_get_returns_none_for_unknown(self):
registry = GroupRegistry()
assert registry.get(-999) is None
- def test_is_monitored(self):
- registry = GroupRegistry()
- gc = GroupConfig(group_id=-100, warning_topic_id=1)
- registry.register(gc)
- assert registry.is_monitored(-100) is True
- assert registry.is_monitored(-999) is False
-
def test_all_groups(self):
registry = GroupRegistry()
gc1 = GroupConfig(group_id=-100, warning_topic_id=1)
@@ -178,7 +171,6 @@ def test_empty_registry(self):
registry = GroupRegistry()
assert registry.all_groups() == []
assert registry.get(-100) is None
- assert registry.is_monitored(-100) is False
class TestLoadGroupsFromJson:
def test_load_valid_json(self):
@@ -356,8 +348,8 @@ def test_builds_from_json_file(self):
registry = build_group_registry(settings)
assert len(registry.all_groups()) == 2
- assert registry.is_monitored(-100)
- assert registry.is_monitored(-200)
+ assert registry.get(-100) is not None
+ assert registry.get(-200) is not None
def test_falls_back_to_env(self):
settings = MagicMock()
@@ -502,7 +494,7 @@ def test_init_and_get(self):
registry = init_group_registry(settings)
assert registry is get_group_registry()
- assert registry.is_monitored(-100)
+ assert registry.get(-100) is not None
def test_reset_clears_registry(self):
settings = MagicMock()
diff --git a/tests/test_plugin_definitions.py b/tests/test_plugin_definitions.py
deleted file mode 100644
index 6a8a741..0000000
--- a/tests/test_plugin_definitions.py
+++ /dev/null
@@ -1,24 +0,0 @@
-"""Tests for plugin definitions."""
-from bot.plugins.definitions import ADMIN_COMMANDS, PLUGIN_NAMES
-
-
-def test_admin_commands_constant():
- """Test that ADMIN_COMMANDS contains expected admin-only plugins."""
- expected = {
- "verify", "unverify", "check", "trust", "untrust", "trusted_list",
- "check_forwarded_message", "verify_callback", "unverify_callback",
- "warn_callback", "trust_callback", "untrust_callback",
- }
- assert ADMIN_COMMANDS == expected
-
-
-def test_admin_commands_subset_of_plugin_names():
- """Test that admin commands are a subset of all plugin names."""
- assert ADMIN_COMMANDS.issubset(PLUGIN_NAMES)
-
-
-def test_admin_commands_not_gateable():
- """Test that admin commands are explicitly marked as not gateable."""
- # This documents the architectural decision that admin commands
- # skip guard_plugin wrapping intentionally
- assert len(ADMIN_COMMANDS) == 12
diff --git a/tests/test_plugin_manager.py b/tests/test_plugin_manager.py
index 04cd0b3..728d0d9 100644
--- a/tests/test_plugin_manager.py
+++ b/tests/test_plugin_manager.py
@@ -5,8 +5,7 @@
from bot.group_config import GroupConfig, GroupRegistry
import pytest
-from bot.plugins import base
-from bot.plugins.config import guard_plugin, is_plugin_enabled, is_plugin_enabled_for_group, resolve_plugin_toggles
+from bot.plugins.config import guard_plugin, is_plugin_enabled_for_group, resolve_plugin_toggles
from bot.plugins.definitions import MANIFEST_ORDER, PLUGIN_NAMES as KNOWN_PLUGINS, get_plugin_definitions
from bot.plugins.manager import PluginManager, compute_effective_plugin_map
@@ -54,12 +53,6 @@ def test_result_contains_all_known_plugins(self):
toggles = resolve_plugin_toggles({}, None)
assert set(toggles.keys()) == KNOWN_PLUGINS
- def test_is_plugin_enabled_convenience(self):
- """is_plugin_enabled returns correct bool for a single plugin."""
- toggles = resolve_plugin_toggles({"captcha": False}, None)
- assert is_plugin_enabled(toggles, "captcha") is False
- assert is_plugin_enabled(toggles, "verify") is True
-
def test_group_override_false_overrides_env_true(self):
"""Group override False wins over env default True."""
toggles = resolve_plugin_toggles(
@@ -79,20 +72,6 @@ def test_partial_group_override(self):
assert toggles["verify"] is False # from env
assert toggles["profile_monitor"] is True # default True
-class TestPluginContracts:
- """Verify plugin base contracts are importable and well-typed."""
-
- def test_plugin_protocol_exists(self):
- """Plugin protocol is exported from base module."""
- assert hasattr(base, "PluginProtocol")
-
- def test_plugin_protocol_has_fields(self):
- """Plugin protocol defines expected fields as annotations + register method."""
- assert "name" in base.PluginProtocol.__annotations__
- assert "description" in base.PluginProtocol.__annotations__
- assert "handler_group" in base.PluginProtocol.__annotations__
- assert hasattr(base.PluginProtocol, "register")
-
class TestPluginDefinitions:
"""Verify plugin definitions match KNOWN_PLUGINS and have correct types."""
@@ -242,91 +221,13 @@ def test_manifest_order_sorted_by_handler_group(self):
)
class TestBuiltinModules:
- """Verify built-in wrapper modules exist and export plugin objects."""
+ """Verify built-in wrapper modules exist."""
def test_builtin_init_module_exists(self):
"""builtin/__init__.py is importable."""
import bot.plugins.builtin # noqa: F811
assert hasattr(bot.plugins.builtin, "__file__")
- def test_topic_guard_module_has_plugin(self):
- """builtin/topic_guard.py exports a plugin object."""
- import bot.plugins.builtin.topic_guard # noqa: F811
- plugin = bot.plugins.builtin.topic_guard.plugin
- assert isinstance(plugin, base.PluginProtocol)
- assert plugin.name == "topic_guard"
-
- def test_commands_module_has_plugin(self):
- """builtin/commands.py exports a plugin object."""
- import bot.plugins.builtin.commands # noqa: F811
- plugin = bot.plugins.builtin.commands.plugin
- assert isinstance(plugin, base.PluginProtocol)
- assert plugin.name == "commands"
-
- def test_captcha_module_has_plugin(self):
- """builtin/captcha.py exports a plugin object."""
- import bot.plugins.builtin.captcha # noqa: F811
- plugin = bot.plugins.builtin.captcha.plugin
- assert isinstance(plugin, base.PluginProtocol)
- assert plugin.name == "captcha"
-
- def test_dm_module_has_plugin(self):
- """builtin/dm.py exports a plugin object."""
- import bot.plugins.builtin.dm # noqa: F811
- plugin = bot.plugins.builtin.dm.plugin
- assert isinstance(plugin, base.PluginProtocol)
- assert plugin.name == "dm"
-
- def test_spam_module_has_plugin(self):
- """builtin/spam.py exports a plugin object."""
- import bot.plugins.builtin.spam # noqa: F811
- plugin = bot.plugins.builtin.spam.plugin
- assert isinstance(plugin, base.PluginProtocol)
- assert plugin.name == "spam"
-
- def test_profile_monitor_module_has_plugin(self):
- """builtin/profile_monitor.py exports a plugin object."""
- import bot.plugins.builtin.profile_monitor # noqa: F811
- plugin = bot.plugins.builtin.profile_monitor.plugin
- assert isinstance(plugin, base.PluginProtocol)
- assert plugin.name == "profile_monitor"
-
- def test_jobs_module_has_plugin(self):
- """builtin/jobs.py exports a plugin object."""
- import bot.plugins.builtin.jobs # noqa: F811
- plugin = bot.plugins.builtin.jobs.plugin
- assert isinstance(plugin, base.PluginProtocol)
- assert plugin.name == "jobs"
-
- def test_each_plugin_satisfies_protocol(self):
- """Every builtin plugin object satisfies PluginProtocol with correct fields."""
- import bot.plugins.builtin.captcha as captcha_mod
- import bot.plugins.builtin.commands as commands_mod
- import bot.plugins.builtin.dm as dm_mod
- import bot.plugins.builtin.jobs as jobs_mod
- import bot.plugins.builtin.profile_monitor as pm_mod
- import bot.plugins.builtin.spam as spam_mod
- import bot.plugins.builtin.topic_guard as tg_mod
-
- plugin_map = {
- "topic_guard": tg_mod.plugin,
- "commands": commands_mod.plugin,
- "captcha": captcha_mod.plugin,
- "dm": dm_mod.plugin,
- "spam": spam_mod.plugin,
- "profile_monitor": pm_mod.plugin,
- "jobs": jobs_mod.plugin,
- }
-
- for name, plugin in plugin_map.items():
- assert isinstance(plugin, base.PluginProtocol), f"{name} fails PluginProtocol"
- assert isinstance(plugin.name, str)
- assert len(plugin.name) > 0
- assert isinstance(plugin.handler_group, int)
- assert isinstance(plugin.description, str)
- assert len(plugin.description) > 0
- assert callable(getattr(plugin, "register", None))
-
class TestComputeEffectivePluginMap:
"""compute_effective_plugin_map: per-group toggle dict from registry + env defaults."""
@@ -728,15 +629,6 @@ def test_all_includes_new_exports(self):
assert "is_plugin_enabled_for_group" in bot.plugins.__all__
assert "compute_effective_plugin_map" in bot.plugins.__all__
-class TestIsPluginEnabledEdgeCases:
- """Edge cases for is_plugin_enabled."""
-
- def test_is_plugin_enabled_missing_key_raises_key_error(self):
- """Missing plugin name in toggles raises KeyError."""
- toggles = {"captcha": True, "verify": False}
- with pytest.raises(KeyError):
- is_plugin_enabled(toggles, "non_existent_plugin")
-
class TestComputeEffectivePluginMapEdgeCases:
"""Edge cases for compute_effective_plugin_map."""
diff --git a/tests/test_whitelist.py b/tests/test_whitelist.py
index 1944124..ab24b49 100644
--- a/tests/test_whitelist.py
+++ b/tests/test_whitelist.py
@@ -5,7 +5,6 @@
from bot.handlers.anti_spam import (
extract_urls,
has_external_reply,
- has_link,
has_non_whitelisted_link,
has_story,
is_forwarded,
@@ -249,67 +248,6 @@ def test_is_forwarded_without_forward_origin():
assert not is_forwarded(message)
-# Tests for has_link
-def test_has_link_with_url_entity():
- """Test has_link detects URL entities."""
- message = MagicMock(spec=Message)
- entity = MagicMock(spec=MessageEntity)
- entity.type = MessageEntity.URL
- message.entities = [entity]
- message.caption_entities = None
- assert has_link(message)
-
-
-def test_has_link_with_text_link_entity():
- """Test has_link detects TEXT_LINK entities."""
- message = MagicMock(spec=Message)
- entity = MagicMock(spec=MessageEntity)
- entity.type = MessageEntity.TEXT_LINK
- message.entities = [entity]
- message.caption_entities = None
- assert has_link(message)
-
-
-def test_has_link_in_caption_entities():
- """Test has_link detects links in caption entities."""
- message = MagicMock(spec=Message)
- entity = MagicMock(spec=MessageEntity)
- entity.type = MessageEntity.URL
- message.entities = None
- message.caption_entities = [entity]
- assert has_link(message)
-
-
-def test_has_link_with_multiple_entities():
- """Test has_link works with multiple entities."""
- message = MagicMock(spec=Message)
- url_entity = MagicMock(spec=MessageEntity)
- url_entity.type = MessageEntity.URL
- bold_entity = MagicMock(spec=MessageEntity)
- bold_entity.type = MessageEntity.BOLD
- message.entities = [bold_entity, url_entity]
- message.caption_entities = None
- assert has_link(message)
-
-
-def test_has_link_no_link_entities():
- """Test has_link returns False when no link entities exist."""
- message = MagicMock(spec=Message)
- entity = MagicMock(spec=MessageEntity)
- entity.type = MessageEntity.BOLD
- message.entities = [entity]
- message.caption_entities = None
- assert not has_link(message)
-
-
-def test_has_link_empty_entities():
- """Test has_link returns False for messages without entities."""
- message = MagicMock(spec=Message)
- message.entities = None
- message.caption_entities = None
- assert not has_link(message)
-
-
# Tests for has_external_reply
def test_has_external_reply_with_external_reply():
"""Test has_external_reply detects external replies."""