import os import logging from typing import Optional, List from telethon import TelegramClient, events, Button from db.models import Post, TargetChannel from db.repository import Repository from bot.keyboards import get_review_keyboard from core.metrics import ADMIN_ACTIONS_TOTAL from core.proxy import get_telegram_proxy logger = logging.getLogger(__name__) SESSION_DIR = os.getenv("SESSION_DIR", "/app/sessions" if os.path.exists("/app") else "/projects/telegram-bots/copykar/sessions") def get_main_menu_keyboard(): return [ [Button.text("šŸ”‘ Request Login Code", resize=True), Button.text("šŸ“Š Fleet Statistics", resize=True)], [Button.text("šŸ“” Monitored Sources", resize=True), Button.text("šŸŽÆ Target Channels", resize=True)], [Button.text("āž• Add Source Guide", resize=True), Button.text("āž• Add Target Guide", resize=True)], [Button.text("ā“ Help & Documentation", resize=True)] ] class AdminBotService: def __init__( self, repo: Repository, bot_token: Optional[str] = None, api_id: Optional[int] = None, api_hash: Optional[str] = None, review_channel_id: Optional[int] = None, admin_user_ids: Optional[List[int]] = None, session_name: Optional[str] = None, ): self.repo = repo self.bot_token = bot_token or os.getenv("BOT_TOKEN", "") self.api_id = api_id or int(os.getenv("API_ID", "0")) self.api_hash = api_hash or os.getenv("API_HASH", "") self.review_channel_id = review_channel_id or int(os.getenv("REVIEW_CHANNEL_ID", "0")) raw_admins = os.getenv("ADMIN_USER_IDS", "") self.admin_user_ids = admin_user_ids or [int(x.strip()) for x in raw_admins.split(",") if x.strip()] self.session_name = session_name or os.path.join(SESSION_DIR, "admin_bot.session") os.makedirs(os.path.dirname(self.session_name), exist_ok=True) self.client = TelegramClient(self.session_name, self.api_id, self.api_hash, proxy=get_telegram_proxy()) self.collector = None def set_collector(self, collector): self.collector = collector def is_admin(self, user_id: int) -> bool: return not self.admin_user_ids or user_id in self.admin_user_ids async def notify_admins(self, text: str): """Broadcast message to review channel and all admin DMs.""" if self.review_channel_id: try: await self.client.send_message(self.review_channel_id, text, parse_mode="html") except Exception as e: logger.error(f"Failed to notify review channel: {e}") for admin_id in self.admin_user_ids: try: await self.client.send_message(admin_id, text, parse_mode="html", buttons=get_main_menu_keyboard()) except Exception as e: logger.debug(f"Could not send DM to admin {admin_id}: {e}") async def start(self): logger.info("Starting Admin Review Bot...") await self.client.start(bot_token=self.bot_token) logger.info("Admin Review Bot connected successfully.") self._register_handlers() def _register_handlers(self): # --- /start and Main Menu --- @self.client.on(events.NewMessage(pattern=r"(?i)^(/start|/menu|menu)$")) async def cmd_start(event: events.NewMessage.Event): if not self.is_admin(event.sender_id): await event.reply(f"ā›” Unauthorized user ID: {event.sender_id}. Please add this ID to ADMIN_USER_IDS in .env.", parse_mode="html") return userbot_status = "šŸ”“ Not Authorized" if self.collector and self.collector.client.is_connected() and await self.collector.client.is_user_authorized(): me = await self.collector.client.get_me() userbot_status = f"🟢 Online ({me.first_name})" welcome_text = ( "šŸ‘‹ Welcome to Copykar Admin Console!\n\n" f"• šŸ¤– Userbot Status: {userbot_status}\n" f"• šŸ“‹ Review Channel: {self.review_channel_id}\n\n" "Use the interactive menu buttons below to manage the fleet:" ) await event.reply(welcome_text, parse_mode="html", buttons=get_main_menu_keyboard()) # --- Interactive Userbot Authentication Commands --- @self.client.on(events.NewMessage(pattern=r"(?i)^(/request_code|šŸ”‘ Request Login Code)$")) async def cmd_request_code(event: events.NewMessage.Event): if not self.is_admin(event.sender_id): return if not self.collector: await event.reply("Collector service not linked.") return msg = await event.reply("ā³ Contacting Telegram to request login code...") try: sent = await self.collector.client.send_code_request(self.collector.phone) self.collector.phone_code_hash = sent.phone_code_hash await msg.edit( f"šŸ“© Code Sent!\n\n" f"Telegram sent a verification code to {self.collector.phone}.\n\n" f"Please reply with:\n" f"/code <your_code>\n\n" f"Example: /code 12345", parse_mode="html" ) except Exception as e: await msg.edit(f"āŒ Could not request code: {e}") @self.client.on(events.NewMessage(pattern=r"^/code\s+(\S+)")) async def cmd_code(event: events.NewMessage.Event): if not self.is_admin(event.sender_id): return if not self.collector: await event.reply("Collector service not linked.") return code = event.pattern_match.group(1).strip() status_msg = await event.reply("ā³ Submitting code to Telegram...") result = await self.collector.submit_code(code) await status_msg.edit(result, parse_mode="html", buttons=get_main_menu_keyboard()) @self.client.on(events.NewMessage(pattern=r"^/password\s+(.+)")) async def cmd_password(event: events.NewMessage.Event): if not self.is_admin(event.sender_id): return if not self.collector: await event.reply("Collector service not linked.") return pwd = event.pattern_match.group(1).strip() status_msg = await event.reply("ā³ Verifying 2FA password...") result = await self.collector.submit_password(pwd) await status_msg.edit(result, parse_mode="html", buttons=get_main_menu_keyboard()) # --- Statistics --- @self.client.on(events.NewMessage(pattern=r"(?i)^(/stats|šŸ“Š Fleet Statistics)$")) async def cmd_stats(event: events.NewMessage.Event): if not self.is_admin(event.sender_id): return pending_ai = len(await self.repo.get_posts_by_status("pending_ai", limit=1000)) pending_review = len(await self.repo.get_posts_by_status("pending_review", limit=1000)) approved = len(await self.repo.get_posts_by_status("approved", limit=1000)) published = len(await self.repo.get_posts_by_status("published", limit=1000)) rejected = len(await self.repo.get_posts_by_status("rejected", limit=1000)) text = ( "šŸ“Š Copykar Fleet Metrics\n\n" f"• ā³ Pending AI: {pending_ai}\n" f"• šŸ“‹ Pending Review: {pending_review}\n" f"• šŸš€ Approved (In Queue): {approved}\n" f"• āœ… Published: {published}\n" f"• āŒ Rejected: {rejected}\n\n" "šŸ“ˆ Grafana Dashboard: http://localhost:3000" ) await event.reply(text, parse_mode="html", buttons=get_main_menu_keyboard()) # --- Sources Management --- @self.client.on(events.NewMessage(pattern=r"(?i)^(/sources|šŸ“” Monitored Sources)$")) async def cmd_sources(event: events.NewMessage.Event): if not self.is_admin(event.sender_id): return sources = await self.repo.get_active_sources() if not sources: await event.reply("No active source channels configured.\nUse āž• Add Source Guide to add one.", parse_mode="html") return lines = ["šŸ“” Active Monitored Sources:\n"] for s in sources: lines.append(f"• ID: {s.channel_id} | {s.title or 'N/A'} (@{s.username or 'none'})") await event.reply("\n".join(lines), parse_mode="html", buttons=get_main_menu_keyboard()) @self.client.on(events.NewMessage(pattern=r"(?i)^(āž• Add Source Guide)$")) async def cmd_add_source_guide(event: events.NewMessage.Event): if not self.is_admin(event.sender_id): return guide = ( "āž• How to Add a Source Channel:\n\n" "Send the command in this format:\n" "/add_source <channel_id> <title> [username]\n\n" "Example:\n" "/add_source -1001234567890 TechNews technews_chan" ) await event.reply(guide, parse_mode="html") @self.client.on(events.NewMessage(pattern=r"^/add_source\s+(-?\d+)\s+([^\s]+)(?:\s+([^\s]+))?")) async def cmd_add_source(event: events.NewMessage.Event): if not self.is_admin(event.sender_id): return ch_id = int(event.pattern_match.group(1)) title = event.pattern_match.group(2) username = event.pattern_match.group(3) await self.repo.add_source(channel_id=ch_id, title=title, username=username) await event.reply(f"āœ… Added source channel {title} ({ch_id}) to monitoring.", parse_mode="html", buttons=get_main_menu_keyboard()) # --- Targets Management --- @self.client.on(events.NewMessage(pattern=r"(?i)^(/targets|šŸŽÆ Target Channels)$")) async def cmd_targets(event: events.NewMessage.Event): if not self.is_admin(event.sender_id): return targets = await self.repo.get_active_targets() if not targets: await event.reply("No target channels configured.\nUse āž• Add Target Guide to add one.", parse_mode="html") return lines = ["šŸŽÆ Target Publishing Channels:\n"] for t in targets: lines.append(f"• ID: {t.id} (Channel: {t.channel_id})\n Title: {t.title} | Interval: {t.post_interval_min}m") await event.reply("\n".join(lines), parse_mode="html", buttons=get_main_menu_keyboard()) @self.client.on(events.NewMessage(pattern=r"(?i)^(āž• Add Target Guide)$")) async def cmd_add_target_guide(event: events.NewMessage.Event): if not self.is_admin(event.sender_id): return guide = ( "šŸŽÆ How to Add a Target Channel:\n\n" "Send the command in this format:\n" "/add_target <channel_id> <title> <interval_minutes> [username]\n\n" "Example (posts every 30 minutes):\n" "/add_target -1009876543210 MyMainChannel 30 my_main_chan" ) await event.reply(guide, parse_mode="html") @self.client.on(events.NewMessage(pattern=r"^/add_target\s+(-?\d+)\s+([^\s]+)\s+(\d+)(?:\s+([^\s]+))?")) async def cmd_add_target(event: events.NewMessage.Event): if not self.is_admin(event.sender_id): return ch_id = int(event.pattern_match.group(1)) title = event.pattern_match.group(2) interval_min = int(event.pattern_match.group(3)) username = event.pattern_match.group(4) await self.repo.add_target(channel_id=ch_id, title=title, username=username, post_interval_min=interval_min) await event.reply(f"āœ… Added target channel {title} with interval {interval_min}m.", parse_mode="html", buttons=get_main_menu_keyboard()) @self.client.on(events.NewMessage(pattern=r"^/set_interval\s+(\d+)\s+(\d+)")) async def cmd_set_interval(event: events.NewMessage.Event): if not self.is_admin(event.sender_id): return target_id = int(event.pattern_match.group(1)) new_interval = int(event.pattern_match.group(2)) target = await self.repo.get_target_by_id(target_id) if not target: await event.reply("Target channel not found.") return await self.repo.add_target( channel_id=target.channel_id, title=target.title, username=target.username, post_interval_min=new_interval ) await event.reply(f"āœ… Updated interval for {target.title} to {new_interval} minutes.", parse_mode="html", buttons=get_main_menu_keyboard()) @self.client.on(events.NewMessage(pattern=r"(?i)^(/help|ā“ Help & Documentation)$")) async def cmd_help(event: events.NewMessage.Event): if not self.is_admin(event.sender_id): return help_text = ( "šŸ“– Copykar Bot Quick Help\n\n" "1. Authentication: Click šŸ”‘ Request Login Code and submit via /code <12345>.\n" "2. Monitored Sources: Add channels the userbot should listen to via /add_source.\n" "3. Review Flow: AI scans posts, checks duplicates, and sends drafts to the review channel with inline approval buttons.\n" "4. Publishing: Approved posts are published to your target channels strictly according to their interval minutes." ) await event.reply(help_text, parse_mode="html", buttons=get_main_menu_keyboard()) # --- Review Keyboard Callbacks --- @self.client.on(events.CallbackQuery) async def on_callback(event: events.CallbackQuery.Event): if not self.is_admin(event.sender_id): await event.answer("ā›” You are not authorized.", alert=True) return data = event.data.decode("utf-8") if data.startswith("appr:"): _, post_id_str, target_id_str = data.split(":") post_id = int(post_id_str) target_id = int(target_id_str) target = await self.repo.get_target_by_id(target_id) target_title = target.title if target else f"Target #{target_id}" await self.repo.approve_post(post_id, target_id) ADMIN_ACTIONS_TOTAL.labels(action="approved").inc() await event.edit( f"{event.text}\n\nāœ… Approved for {target_title} by admin.", parse_mode="html", buttons=None ) await event.answer(f"Approved for {target_title}!") elif data.startswith("rej:"): _, post_id_str = data.split(":") post_id = int(post_id_str) await self.repo.reject_post(post_id) ADMIN_ACTIONS_TOTAL.labels(action="rejected").inc() await event.edit( f"{event.text}\n\nāŒ Rejected by admin.", parse_mode="html", buttons=None ) await event.answer("Post rejected.") async def send_review_post(self, post_id: int): post = await self.repo.get_post_by_id(post_id) if not post or not self.review_channel_id: return targets = await self.repo.get_active_targets() keyboard = get_review_keyboard(post.id, targets) tags_str = ", ".join(post.tags) if post.tags else "None" dup_warning = "" if post.is_duplicate: dup_warning = ( f"āš ļø [DUPLICATE DETECTED]\n" f"Reason: {post.similarity_reason or 'Similar story already published'}\n" f"Matched Post ID: #{post.duplicate_of_id}\n\n" ) caption = ( f"šŸ“Œ Subject: {post.subject or 'N/A'}\n" f"šŸ· Tags: {tags_str}\n\n" f"{dup_warning}" f"šŸ“ Generated Post Draft:\n" f"{post.ai_text or post.raw_text}\n\n" f"Source: Channel {post.source_channel_id} | Msg #{post.source_message_id}" ) try: if post.media_path and os.path.exists(post.media_path): msg = await self.client.send_file( self.review_channel_id, file=post.media_path, caption=caption, parse_mode="html", buttons=keyboard ) else: msg = await self.client.send_message( self.review_channel_id, caption, parse_mode="html", buttons=keyboard ) await self.repo.update_review_message_id(post.id, msg.id) except Exception as e: logger.error(f"Failed to send review post {post.id} to review channel: {e}", exc_info=True) async def stop(self): if self.client.is_connected(): await self.client.disconnect() logger.info("Admin Review Bot disconnected.")