|
1 | 1 | """Captcha verification logic (file + secret command flow).""" |
2 | 2 |
|
| 3 | +import re |
3 | 4 | from pathlib import Path |
4 | 5 |
|
5 | | -from telegram import ChatPermissions |
| 6 | +from telegram import Chat, ChatPermissions, InlineKeyboardButton, InlineKeyboardMarkup, User |
6 | 7 |
|
7 | 8 | from ..db.base import AsyncRepository |
8 | 9 |
|
| 10 | +BUTTON_URL_PATTERN = re.compile(r"\[([^\]]+)\]\(buttonurl://([^)]+)\)") |
| 11 | + |
9 | 12 |
|
10 | 13 | class CaptchaService: |
11 | 14 | """Handles welcome captcha: restrict new members until they send secret command in DM.""" |
12 | 15 |
|
13 | 16 | def __init__( |
14 | | - self, repository: AsyncRepository, secret_command: str, file_path: str |
| 17 | + self, |
| 18 | + repository: AsyncRepository, |
| 19 | + secret_command: str, |
| 20 | + file_path: str, |
| 21 | + rules_url: str | None = None, |
15 | 22 | ) -> None: |
16 | 23 | self._repo = repository |
17 | 24 | self._secret_command = secret_command.strip().lower() |
18 | 25 | self._file_path = Path(file_path) |
| 26 | + self._rules_url = rules_url |
19 | 27 |
|
20 | 28 | def _matches_secret(self, text: str) -> bool: |
21 | 29 | return text.strip().lower() == self._secret_command |
22 | 30 |
|
23 | | - def get_welcome_message(self) -> str: |
24 | | - """Return the welcome message with captcha instructions.""" |
| 31 | + def get_default_welcome_template(self, bot_username: str) -> str: |
| 32 | + """Return the default welcome message template with placeholders.""" |
25 | 33 | return ( |
26 | | - "Benvenuto nel gruppo Python Italia! 🐍\n\n" |
27 | | - "Per poter partecipare alle discussioni, leggere il file delle regole " |
28 | | - "e inviare il comando segreto che troverai al bot in chat privata.\n\n" |
29 | | - "Per aprire una chat con il bot, clicca sul suo nome e seleziona 'Avvia'." |
| 34 | + "Benvenuto {username}! Per partecipare alle discussioni, leggi il regolamento.\n" |
| 35 | + f"[Verifica](buttonurl://t.me/{bot_username}?start=verify)" |
30 | 36 | ) |
31 | 37 |
|
| 38 | + def format_welcome_message( |
| 39 | + self, template: str, user: User, chat: Chat, bot_username: str |
| 40 | + ) -> str: |
| 41 | + """Substitute placeholders in the welcome message template.""" |
| 42 | + username = f"@{user.username}" if user.username else user.full_name |
| 43 | + replacements = { |
| 44 | + "{username}": username, |
| 45 | + "{chatname}": chat.title or "this group", |
| 46 | + } |
| 47 | + result = template |
| 48 | + for placeholder, value in replacements.items(): |
| 49 | + result = result.replace(placeholder, value) |
| 50 | + return result |
| 51 | + |
| 52 | + def parse_button_urls(self, text: str) -> tuple[str, InlineKeyboardMarkup | None]: |
| 53 | + """Extract buttonurl:// patterns and build InlineKeyboardMarkup. |
| 54 | + |
| 55 | + Returns (clean_text, keyboard) where clean_text has button syntax removed. |
| 56 | + Multiple buttons on the same line become the same row. |
| 57 | + """ |
| 58 | + lines = text.split("\n") |
| 59 | + keyboard_rows: list[list[InlineKeyboardButton]] = [] |
| 60 | + clean_lines: list[str] = [] |
| 61 | + |
| 62 | + for line in lines: |
| 63 | + matches = list(BUTTON_URL_PATTERN.finditer(line)) |
| 64 | + if matches: |
| 65 | + row = [ |
| 66 | + InlineKeyboardButton(text=m.group(1), url=m.group(2)) |
| 67 | + for m in matches |
| 68 | + ] |
| 69 | + keyboard_rows.append(row) |
| 70 | + clean_line = BUTTON_URL_PATTERN.sub("", line).strip() |
| 71 | + if clean_line: |
| 72 | + clean_lines.append(clean_line) |
| 73 | + else: |
| 74 | + clean_lines.append(line) |
| 75 | + |
| 76 | + clean_text = "\n".join(clean_lines).strip() |
| 77 | + keyboard = InlineKeyboardMarkup(keyboard_rows) if keyboard_rows else None |
| 78 | + return clean_text, keyboard |
| 79 | + |
| 80 | + def get_deep_link_url(self, bot_username: str) -> str: |
| 81 | + """Generate the deep link URL for verification.""" |
| 82 | + return f"https://t.me/{bot_username}?start=verify" |
| 83 | + |
| 84 | + def get_rules_url(self) -> str | None: |
| 85 | + """Return the configured rules URL.""" |
| 86 | + return self._rules_url |
| 87 | + |
32 | 88 | def get_captcha_file_content(self) -> str | None: |
33 | 89 | """Return the captcha file content if it exists. Path is relative to cwd.""" |
34 | 90 | path = Path(self._file_path) |
@@ -80,19 +136,29 @@ def is_secret_command(self, text: str) -> bool: |
80 | 136 | """Check if the message matches the secret command.""" |
81 | 137 | return self._matches_secret(text) |
82 | 138 |
|
| 139 | + async def get_welcome_message(self, chat_id: int) -> str | None: |
| 140 | + """Get custom welcome message for a chat, or None for default.""" |
| 141 | + return await self._repo.get_welcome_message(chat_id) |
| 142 | + |
| 143 | + async def set_welcome_message(self, chat_id: int, message: str | None) -> None: |
| 144 | + """Set or remove custom welcome message for a chat.""" |
| 145 | + await self._repo.set_welcome_message(chat_id, message) |
| 146 | + |
83 | 147 | async def get_pending_chats(self, user_id: int) -> list[int]: |
84 | 148 | """Get chats where user is pending verification.""" |
85 | 149 | return await self._repo.get_pending_chats(user_id) |
86 | 150 |
|
87 | | - async def verify_user(self, user_id: int, chat_id: int) -> None: |
88 | | - """Mark user as verified and remove from pending.""" |
89 | | - await self._repo.mark_user_verified(user_id, chat_id) |
90 | | - await self._repo.remove_pending(user_id, chat_id) |
| 151 | + async def verify_user_globally(self, user_id: int) -> None: |
| 152 | + """Mark user as globally verified and remove from all pending.""" |
| 153 | + await self._repo.mark_globally_verified(user_id) |
| 154 | + pending_chats = await self._repo.get_pending_chats(user_id) |
| 155 | + for chat_id in pending_chats: |
| 156 | + await self._repo.remove_pending(user_id, chat_id) |
91 | 157 |
|
92 | 158 | async def add_pending(self, user_id: int, chat_id: int) -> None: |
93 | 159 | """Record that user joined and needs verification.""" |
94 | 160 | await self._repo.add_pending_verification(user_id, chat_id) |
95 | 161 |
|
96 | | - async def is_verified(self, user_id: int, chat_id: int) -> bool: |
97 | | - """Check if user is verified for the chat.""" |
98 | | - return await self._repo.is_user_verified(user_id, chat_id) |
| 162 | + async def is_globally_verified(self, user_id: int) -> bool: |
| 163 | + """Check if user is globally verified.""" |
| 164 | + return await self._repo.is_globally_verified(user_id) |
0 commit comments