"""admin_bot.py — Admin Telegram bot for full system control.""" import asyncio import logging from telegram import Update, Bot from telegram.ext import (ApplicationBuilder, CommandHandler, MessageHandler, ContextTypes, filters) from telegram.constants import ParseMode import database as db import website_builder import customer_handler import ai_agent from config import ADMIN_BOT_TOKEN log = logging.getLogger(__name__) _admin_chat_id: int | None = None _bot_instance: Bot | None = None _template_mode: dict = {} # chat_id → bool # ── Notification (called by customer_handler) ───────────────────────────── async def notify_admin(text: str): if _bot_instance and _admin_chat_id: try: await _bot_instance.send_message( chat_id=_admin_chat_id, text=text, parse_mode=ParseMode.MARKDOWN ) except Exception as e: log.error("notify_admin: %s", e) def _save_admin(chat_id: int): global _admin_chat_id _admin_chat_id = chat_id db.set_state("admin_chat_id", str(chat_id)) def _load_admin(): global _admin_chat_id val = db.get_state("admin_chat_id") if val: _admin_chat_id = int(val) async def _reply(update: Update, text: str, md=False): await update.message.reply_text( text, parse_mode=ParseMode.MARKDOWN if md else None ) # ── Commands ────────────────────────────────────────────────────────────── async def cmd_start(update: Update, ctx: ContextTypes.DEFAULT_TYPE): _save_admin(update.effective_chat.id) db.set_state("running", "1") await _reply(update, "✅ *WebReach STARTED*\n\n" "What happens now:\n" "1️⃣ Playwright scrapes Google Maps for businesses without websites\n" "2️⃣ DeepSeek R1 writes personalised emails (10 per batch, 1 API call)\n" "3️⃣ Emails rotate across all 10 Gmail accounts (10/day each = 100/day)\n" "4️⃣ Replies are checked every 15 min — AI handles full conversation\n" "5️⃣ When customer claims payment → you get a Telegram alert\n\n" "Commands: /stop /status /leads /template /verify N /reject N", md=True ) asyncio.create_task(_run_scrape()) async def cmd_stop(update: Update, ctx: ContextTypes.DEFAULT_TYPE): db.set_state("running", "0") await _reply(update, "⏸ *System PAUSED* — no new emails will be sent.\nSend /start to resume.", md=True) async def cmd_status(update: Update, ctx: ContextTypes.DEFAULT_TYPE): counts = db.count_leads_by_state() total = sum(counts.values()) running = "🟢 Running" if db.is_running() else "🔴 Paused" from config import SENDER_EMAILS, MAX_EMAILS_PER_DAY quota = "\n".join( f" {a['email']}: {db.get_daily_count(a['email'])}/{MAX_EMAILS_PER_DAY}" for a in SENDER_EMAILS ) emoji = {"FOUND":"🔍","OUTREACH_SENT":"📤","DETAILS_REQUESTED":"📋", "DEMO_BUILDING":"🔨","DEMO_SENT":"🌐","NEGOTIATING":"💬", "PAYMENT_PENDING":"💰","PAYMENT_VERIFIED":"✅","DELIVERED":"🎉","REJECTED":"❌"} state_lines = "\n".join( f" {emoji.get(s,'•')} {s}: {n}" for s,n in sorted(counts.items()) ) groq_used = ai_agent.get_request_count() await _reply(update, f"*📊 Status — {running}*\n\n" f"*Leads ({total} total):*\n{state_lines}\n\n" f"*Groq requests today:* {groq_used}/950\n\n" f"*Email quotas:*\n{quota}", md=True ) async def cmd_leads(update: Update, ctx: ContextTypes.DEFAULT_TYPE): with db.get_conn() as c: rows = c.execute( "SELECT id,business_name,state,email FROM leads ORDER BY id DESC LIMIT 20" ).fetchall() if not rows: await _reply(update, "No leads yet.") return lines = ["*Last 20 leads:*\n"] for r in rows: lines.append(f" #{r['id']} [{r['state']}] {r['business_name']} — {r['email']}") await _reply(update, "\n".join(lines), md=True) async def cmd_template(update: Update, ctx: ContextTypes.DEFAULT_TYPE): chat_id = update.effective_chat.id _template_mode[chat_id] = True existing = website_builder.list_user_template_files() ex_str = ", ".join(existing) if existing else "None (using built-in)" await _reply(update, "📁 *Template Upload Mode*\n\n" f"Current files: {ex_str}\n\n" "Send your template files now.\n" "Main file must be `index.html`.\n" "Send /done when finished.\n" "Send /cleartemplate to reset to built-in.", md=True ) async def cmd_done(update: Update, ctx: ContextTypes.DEFAULT_TYPE): _template_mode.pop(update.effective_chat.id, None) files = website_builder.list_user_template_files() await _reply(update, f"✅ Template saved! Files: {', '.join(files) or 'none'}", md=True) async def cmd_clear_template(update: Update, ctx: ContextTypes.DEFAULT_TYPE): website_builder.clear_user_templates() await _reply(update, "🗑️ Template cleared. Using built-in template.", md=True) async def cmd_verify(update: Update, ctx: ContextTypes.DEFAULT_TYPE): args = ctx.args if not args or not args[0].isdigit(): await _reply(update, "Usage: /verify ") return lead_id = int(args[0]) lead = db.get_lead(lead_id) if not lead: await _reply(update, f"Lead #{lead_id} not found.") return await _reply(update, f"✅ Payment confirmed for *{lead['business_name']}* — delivering now...", md=True) asyncio.create_task(customer_handler.verify_payment(lead_id)) async def cmd_reject(update: Update, ctx: ContextTypes.DEFAULT_TYPE): args = ctx.args if not args or not args[0].isdigit(): await _reply(update, "Usage: /reject ") return lead_id = int(args[0]) lead = db.get_lead(lead_id) if not lead: await _reply(update, f"Lead #{lead_id} not found.") return await _reply(update, f"❌ Rejecting payment for *{lead['business_name']}* — blocking site...", md=True) asyncio.create_task(customer_handler.reject_payment(lead_id)) # ── File uploads (template mode) ────────────────────────────────────────── async def handle_document(update: Update, ctx: ContextTypes.DEFAULT_TYPE): if not _template_mode.get(update.effective_chat.id): await _reply(update, "Use /template first to enter upload mode.") return doc = update.message.document if not doc: return file = await ctx.bot.get_file(doc.file_id) content = await file.download_as_bytearray() website_builder.save_user_template_file(doc.file_name or "file.bin", bytes(content)) await _reply(update, f"✅ Saved `{doc.file_name}` ({len(content):,} bytes)", md=True) # ── Natural language handler ────────────────────────────────────────────── async def handle_text(update: Update, ctx: ContextTypes.DEFAULT_TYPE): import re text = update.message.text or "" # Natural language verify/reject if re.search(r"\b(yes|verify|confirm|paid)\b", text, re.I): m = re.search(r"\b(\d+)\b", text) if m: lead_id = int(m.group(1)) lead = db.get_lead(lead_id) if lead: await _reply(update, f"✅ Verifying payment for #{lead_id}...") asyncio.create_task(customer_handler.verify_payment(lead_id)) return if re.search(r"\b(no|reject|block)\b", text, re.I): m = re.search(r"\b(\d+)\b", text) if m: lead_id = int(m.group(1)) lead = db.get_lead(lead_id) if lead: await _reply(update, f"❌ Rejecting payment for #{lead_id}...") asyncio.create_task(customer_handler.reject_payment(lead_id)) return # AI chat via DeepSeek R1 try: counts = db.count_leads_by_state() system = ( f"You are the AI assistant for WebReach, an automated web dev outreach business. " f"System stats: {counts}. Running: {db.is_running()}. " f"Groq requests used today: {ai_agent.get_request_count()}/950. " "Answer the admin's questions concisely." ) resp = ai_agent._chat(system, text, max_tokens=300) await _reply(update, resp) except Exception as e: await _reply(update, f"Error: {e}") # ── Background scrape task ──────────────────────────────────────────────── async def _run_scrape(): if not db.is_running(): return try: import scraper from config import SCRAPE_LOCATION, SCRAPE_CATEGORIES, LEADS_TARGET await notify_admin("🗺 Starting Google Maps scrape...") n = await scraper.run_scraper(SCRAPE_LOCATION, SCRAPE_CATEGORIES, LEADS_TARGET) await notify_admin( f"🗺 *Scrape Complete!*\n" f"{n} new leads found.\n" "Outreach emails will start sending during business hours (9 AM–5 PM)." ) except Exception as e: log.error("Scrape error: %s", e) await notify_admin(f"⚠️ Scrape error: {e}") # ── App builder ─────────────────────────────────────────────────────────── def build_admin_app(): global _bot_instance _load_admin() customer_handler.set_admin_notifier(notify_admin) app = ApplicationBuilder().token(ADMIN_BOT_TOKEN).build() _bot_instance = app.bot app.add_handler(CommandHandler("start", cmd_start)) app.add_handler(CommandHandler("stop", cmd_stop)) app.add_handler(CommandHandler("status", cmd_status)) app.add_handler(CommandHandler("leads", cmd_leads)) app.add_handler(CommandHandler("template", cmd_template)) app.add_handler(CommandHandler("done", cmd_done)) app.add_handler(CommandHandler("cleartemplate", cmd_clear_template)) app.add_handler(CommandHandler("verify", cmd_verify)) app.add_handler(CommandHandler("reject", cmd_reject)) app.add_handler(MessageHandler(filters.Document.ALL, handle_document)) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text)) return app