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())
# --- Direct History Scraper Command ---
@self.client.on(events.NewMessage(pattern=r"^/scrape_history\s+(-?\d+)(?:\s+(\d+))?"))
async def cmd_scrape_history(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
ch_id = int(event.pattern_match.group(1))
limit = int(event.pattern_match.group(2)) if event.pattern_match.group(2) else 20
status_msg = await event.reply(f"ā³ Scraping the last {limit} posts from {ch_id} in the background...", parse_mode="html")
async def progress_notify(txt: str):
await status_msg.edit(txt, parse_mode="html", buttons=get_main_menu_keyboard())
await self.collector.scrape_channel_history(channel_id=ch_id, limit=limit, progress_callback=progress_notify)
# --- 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 with Interactive Scrape Buttons ---
@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.\nTap ā Add Source Guide below to add one.", parse_mode="html", buttons=get_main_menu_keyboard())
return
await event.reply(f"š” Monitored Sources ({len(sources)} Active):\nTap any button below to scrape past posts:", parse_mode="html")
for s in sources:
card = (
f"š¢ {s.title or 'Channel'}\n"
f"⢠ID: {s.channel_id}\n"
f"⢠Username: @{s.username or 'none'}"
)
buttons = [
[
Button.inline(f"š„ Scrape 20 Posts", data=f"hist:{s.channel_id}:20"),
Button.inline(f"š„ Scrape 50 Posts", data=f"hist:{s.channel_id}:50"),
]
]
await event.reply(card, parse_mode="html", buttons=buttons)
@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)
buttons = [
[
Button.inline(f"š„ Scrape 20 Posts Now", data=f"hist:{ch_id}:20"),
Button.inline(f"š„ Scrape 50 Posts Now", data=f"hist:{ch_id}:50")
]
]
await event.reply(
f"ā
Added source channel {title} ({ch_id}).\n\nWould you like to scrape past posts now?",
parse_mode="html",
buttons=buttons
)
# --- 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. Monitored Sources: Tap š” Monitored Sources to view channels and click [š„ Scrape Posts] on any channel.\n"
"2. Review Flow: AI scans posts, checks duplicates, and sends drafts to the review channel with inline approval buttons.\n"
"3. 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())
# --- Inline Callback Queries ---
@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")
# 1. Historical Scraping Callbacks
if data.startswith("hist:"):
_, ch_id_str, limit_str = data.split(":")
ch_id = int(ch_id_str)
limit = int(limit_str)
if not self.collector:
await event.answer("Collector service not linked.", alert=True)
return
await event.edit(f"ā³ Scraping the last {limit} posts from {ch_id}...", parse_mode="html", buttons=None)
async def progress_notify(txt: str):
await event.edit(txt, parse_mode="html")
await self.collector.scrape_channel_history(channel_id=ch_id, limit=limit, progress_callback=progress_notify)
await event.answer(f"Started scraping {limit} posts!")
# 2. Approval Callbacks
elif 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}!")
# 3. Reject Callbacks
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.")