| """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 = {} |
|
|
|
|
| |
|
|
| 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 |
| ) |
|
|
|
|
| |
|
|
| 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 <lead_id>") |
| 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 <lead_id>") |
| 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)) |
|
|
|
|
| |
|
|
| 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) |
|
|
|
|
| |
|
|
| async def handle_text(update: Update, ctx: ContextTypes.DEFAULT_TYPE): |
| import re |
| text = update.message.text or "" |
|
|
| |
| 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 |
|
|
| |
| 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}") |
|
|
|
|
| |
|
|
| 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}") |
|
|
|
|
| |
|
|
| 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 |
|
|
