"""Admin command panel: stats, products, users, deposits, announcements.""" from telegram import Update from telegram.ext import (ContextTypes, ConversationHandler, CommandHandler, MessageHandler, CallbackQueryHandler, filters) from database.db import DB from utils.decorators import admin_only from utils.helpers import money from utils.keyboards import inline import asyncio, logging log = logging.getLogger("genboost.admin") # ---------- Auto-broadcast helper ---------- async def _broadcast_product(context, pid: int, header: str): """Send a nicely formatted product card to every active user.""" p = await DB.fetchone( "SELECT p.id, p.name, p.description, p.price, p.stock, p.is_active, " "c.name AS cat FROM products p LEFT JOIN categories c ON c.id=p.category_id " "WHERE p.id=?", (pid,)) if not p: return stock = p["stock"] status = ("βœ… In Stock" if (p["is_active"] and stock > 0) else "🚫 Out of Stock") text = ( f"{header}\n\n" f"πŸ› {p['name']}\n" f"πŸ“‚ Category: {p['cat'] or 'β€”'}\n" f"πŸ’΅ Price: {money(p['price'])}\n" f"πŸ“¦ Stock: {stock}\n" f"{status}\n\n" f"{(p['description'] or '').strip()}\n\n" "πŸ‘‰ Open the bot and tap πŸ› Products to buy!" ) rows = await DB.fetchall("SELECT user_id FROM users WHERE is_banned=0") sent = delivered = failed = 0 for r in rows: sent += 1 try: await context.bot.send_message(r["user_id"], text, parse_mode="HTML", disable_web_page_preview=True) delivered += 1 except Exception: failed += 1 if sent % 25 == 0: await asyncio.sleep(1) try: await DB.execute( "INSERT INTO broadcasts(admin_id, payload, sent, delivered, failed) VALUES (?,?,?,?,?)", (0, f"[auto] product#{pid}", sent, delivered, failed)) except Exception: pass return delivered, failed # ---------- /admin overview ---------- @admin_only async def admin_panel(update: Update, context: ContextTypes.DEFAULT_TYPE): text = ( "βš™ Admin Panel\n\n" "Commands:\n" "/stats β€” statistics\n" "/users [query] β€” list/search users\n" "/setbal <user_id> <amount> β€” set wallet balance\n" "/ban <user_id> Β· /unban <user_id>\n\n" "Catalog\n" "/addcat <name> Β· /cats\n" "/rencat <id> <new name> Β· /delcat <id>\n" "/addproduct β€” guided product wizard\n" "/products Β· /delproduct <id> Β· /toggleproduct <id>\n" "/setstock <id> <n> Β· /setprice <id> <amt>\n" "/rename <id> <new name> Β· /editdesc <id> <text>\n\n" "πŸ†” /ids β€” list all category & product IDs (for editing)\n" "πŸ’Ž Premium emoji anywhere in a name/description: " "{pe:EMOJI_ID}πŸ”₯ Name β€” paste the custom emoji id and the next emoji becomes premium.\n" "πŸ“’ New products & stock updates auto-broadcast to all users.\n\n" "Deposits\n" "/pending β€” pending deposits\n" "/approve <id> Β· /reject <id> [reason]\n\n" "Support\n" "/tickets β€” open tickets Β· /reply <id> <text>\n\n" "Announcements & Broadcast\n" "/announce <title> | <body>\n" "/broadcast <text> β€” send to all users" ) await update.message.reply_html(text) # ---------- Stats ---------- @admin_only async def stats(update: Update, context: ContextTypes.DEFAULT_TYPE): async def n(sql, p=()): return (await DB.fetchone(sql, p))[0] total_users = await n("SELECT COUNT(*) FROM users") active_users = await n("SELECT COUNT(*) FROM users WHERE is_active=1") new_today = await n("SELECT COUNT(*) FROM users WHERE created_at >= date('now')") total_dep = await n("SELECT COALESCE(SUM(amount),0) FROM deposits WHERE status='approved'") pending_dep = await n("SELECT COUNT(*) FROM deposits WHERE status='pending'") total_rev = await n("SELECT COALESCE(SUM(amount),0) FROM orders") total_orders = await n("SELECT COUNT(*) FROM orders") ref_paid = await n("SELECT COALESCE(SUM(amount),0) FROM referral_rewards") top = await DB.fetchall( "SELECT product_name, COUNT(*) c FROM orders GROUP BY product_name " "ORDER BY c DESC LIMIT 5") text = ( "πŸ“Š Statistics\n\n" f"πŸ‘₯ Total Users: {total_users} (active: {active_users}, new today: {new_today})\n" f"πŸ’° Total Deposits: {money(total_dep)} (pending: {pending_dep})\n" f"πŸ›’ Orders: {total_orders} Β· Revenue: {money(total_rev)}\n" f"🎁 Referral Rewards Paid: {money(ref_paid)}\n\n" "πŸ† Top Products:\n" + ("\n".join(f"β€’ {t['product_name']} β€” {t['c']}" for t in top) or "β€”") ) await update.message.reply_html(text) # ---------- Users ---------- @admin_only async def users_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE): q = " ".join(context.args).strip() if q: like = f"%{q}%" rows = await DB.fetchall( "SELECT user_id, username, first_name, balance FROM users " "WHERE CAST(user_id AS TEXT) LIKE ? OR username LIKE ? OR first_name LIKE ? " "ORDER BY created_at DESC LIMIT 20", (like, like, like)) else: rows = await DB.fetchall( "SELECT user_id, username, first_name, balance FROM users " "ORDER BY created_at DESC LIMIT 20") if not rows: await update.message.reply_text("No users found."); return text = "Users\n\n" + "\n".join( f"{r['user_id']} @{r['username'] or 'β€”'} {r['first_name'] or ''} β€” {money(r['balance'])}" for r in rows) await update.message.reply_html(text) @admin_only async def setbal(update: Update, context: ContextTypes.DEFAULT_TYPE): if len(context.args) != 2: await update.message.reply_text("Usage: /setbal "); return uid, amt = int(context.args[0]), float(context.args[1]) await DB.execute("UPDATE users SET balance=? WHERE user_id=?", (amt, uid)) await update.message.reply_text(f"βœ… Balance for {uid} set to {money(amt)}") @admin_only async def ban(update: Update, context: ContextTypes.DEFAULT_TYPE): uid = int(context.args[0]) await DB.execute("UPDATE users SET is_banned=1 WHERE user_id=?", (uid,)) await update.message.reply_text(f"β›” User {uid} banned.") @admin_only async def unban(update: Update, context: ContextTypes.DEFAULT_TYPE): uid = int(context.args[0]) await DB.execute("UPDATE users SET is_banned=0 WHERE user_id=?", (uid,)) await update.message.reply_text(f"βœ… User {uid} unbanned.") # ---------- Categories ---------- @admin_only async def addcat(update: Update, context: ContextTypes.DEFAULT_TYPE): name = " ".join(context.args).strip() if not name: await update.message.reply_text("Usage: /addcat "); return try: cur = await DB.execute("INSERT INTO categories(name) VALUES (?)", (name,)) await update.message.reply_text(f"βœ… Category #{cur.lastrowid} added: {name}") except Exception: await update.message.reply_text("Category already exists.") @admin_only async def cats_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE): rows = await DB.fetchall("SELECT id, name FROM categories ORDER BY id") if not rows: await update.message.reply_text("No categories yet."); return await update.message.reply_text( "\n".join(f"#{r['id']} β€” {r['name']}" for r in rows)) # ---------- Product wizard ---------- P_CAT, P_NAME, P_DESC, P_PRICE, P_STOCK, P_TYPE, P_CONTENT, P_FILE = range(8) @admin_only async def addproduct(update, context): rows = await DB.fetchall("SELECT id, name FROM categories ORDER BY name") if not rows: await update.message.reply_text("Create a category first with /addcat .") return ConversationHandler.END kb = inline([[(r["name"], f"padd:cat:{r['id']}")] for r in rows]) await update.message.reply_text("Choose a category:", reply_markup=kb) return P_CAT async def p_cat(update, context): q = update.callback_query; await q.answer() context.user_data["p"] = {"category_id": int(q.data.split(":")[2])} await q.edit_message_text("Send product name:", parse_mode="HTML") return P_NAME async def p_name(update, context): context.user_data["p"]["name"] = update.message.text.strip() await update.message.reply_text("Send description (or - to skip):", parse_mode="HTML") return P_DESC async def p_desc(update, context): t = update.message.text.strip() context.user_data["p"]["description"] = "" if t == "-" else t await update.message.reply_text("Send price in USD (e.g. 4.99):", parse_mode="HTML") return P_PRICE async def p_price(update, context): try: context.user_data["p"]["price"] = float(update.message.text) except ValueError: await update.message.reply_text("Invalid number, try again."); return P_PRICE await update.message.reply_text("Send stock quantity (integer):", parse_mode="HTML") return P_STOCK async def p_stock(update, context): try: context.user_data["p"]["stock"] = int(update.message.text) except ValueError: await update.message.reply_text("Invalid integer, try again."); return P_STOCK kb = inline([[(t, f"padd:type:{t}")] for t in ["text", "credentials", "license", "link", "file"]]) await update.message.reply_text("Choose delivery type:", parse_mode="HTML", reply_markup=kb) return P_TYPE async def p_type(update, context): q = update.callback_query; await q.answer() context.user_data["p"]["product_type"] = q.data.split(":")[2] await q.edit_message_text( "Send the delivery content (text/link/credentials/license key).\n" "If your product is a file, send any short note here β€” you'll upload the file next.", parse_mode="HTML") return P_CONTENT async def p_content(update, context): context.user_data["p"]["delivery_content"] = update.message.text if context.user_data["p"]["product_type"] == "file": await update.message.reply_text("Now upload the file (document):", parse_mode="HTML") return P_FILE return await _save_product(update, context) async def p_file(update, context): if not update.message.document: await update.message.reply_text("Please send a document file."); return P_FILE context.user_data["p"]["delivery_file_id"] = update.message.document.file_id return await _save_product(update, context) async def _save_product(update, context): p = context.user_data.pop("p") cur = await DB.execute( "INSERT INTO products(category_id, name, description, price, stock, " "product_type, delivery_content, delivery_file_id) VALUES (?,?,?,?,?,?,?,?)", (p["category_id"], p["name"], p.get("description", ""), p["price"], p["stock"], p["product_type"], p.get("delivery_content", ""), p.get("delivery_file_id"))) pid = cur.lastrowid await update.message.reply_text(f"βœ… Product #{pid} created. Broadcasting to users…") try: res = await _broadcast_product(context, pid, "πŸ†• New Product Available!") if res: d, f = res await update.message.reply_text(f"πŸ“’ Broadcast sent. Delivered: {d} Β· Failed: {f}") except Exception as e: log.exception("broadcast failed: %s", e) return ConversationHandler.END async def p_cancel(update, context): context.user_data.pop("p", None) if update.message: await update.message.reply_text("Cancelled.") return ConversationHandler.END def product_wizard(): return ConversationHandler( entry_points=[CommandHandler("addproduct", addproduct)], states={ P_CAT: [CallbackQueryHandler(p_cat, pattern=r"^padd:cat:\d+$")], P_NAME: [MessageHandler(filters.TEXT & ~filters.COMMAND, p_name)], P_DESC: [MessageHandler(filters.TEXT & ~filters.COMMAND, p_desc)], P_PRICE: [MessageHandler(filters.TEXT & ~filters.COMMAND, p_price)], P_STOCK: [MessageHandler(filters.TEXT & ~filters.COMMAND, p_stock)], P_TYPE: [CallbackQueryHandler(p_type, pattern=r"^padd:type:")], P_CONTENT: [MessageHandler(filters.TEXT & ~filters.COMMAND, p_content)], P_FILE: [MessageHandler(filters.Document.ALL, p_file)], }, fallbacks=[CommandHandler("cancel", p_cancel)], allow_reentry=True, per_message=False, ) @admin_only async def products_cmd(update, context): rows = await DB.fetchall( "SELECT id, name, price, stock, is_active FROM products ORDER BY id DESC LIMIT 50") if not rows: await update.message.reply_text("No products yet."); return await update.message.reply_text("\n".join( f"#{r['id']} {'βœ…' if r['is_active'] else '🚫'} {r['name']} β€” {money(r['price'])} ({r['stock']} left)" for r in rows)) @admin_only async def delproduct(update, context): pid = int(context.args[0]) await DB.execute("DELETE FROM products WHERE id=?", (pid,)) await update.message.reply_text(f"πŸ—‘ Product #{pid} deleted.") @admin_only async def toggleproduct(update, context): pid = int(context.args[0]) await DB.execute("UPDATE products SET is_active = 1 - is_active WHERE id=?", (pid,)) await update.message.reply_text(f"πŸ” Toggled product #{pid}.") @admin_only async def setstock(update, context): pid, n = int(context.args[0]), int(context.args[1]) await DB.execute("UPDATE products SET stock=? WHERE id=?", (n, pid)) await update.message.reply_text(f"βœ… Product #{pid} stock = {n}. Broadcasting update…") header = ("πŸ“¦ Stock Restocked!" if n > 0 else "🚫 Out of Stock") try: res = await _broadcast_product(context, pid, header) if res: d, f = res await update.message.reply_text(f"πŸ“’ Broadcast sent. Delivered: {d} Β· Failed: {f}") except Exception as e: log.exception("stock broadcast failed: %s", e) # ---------- /ids β€” list all category & product IDs ---------- @admin_only async def ids_cmd(update, context): cats = await DB.fetchall("SELECT id, name FROM categories ORDER BY id") prods = await DB.fetchall( "SELECT p.id, p.name, p.price, p.stock, p.is_active, c.name AS cat " "FROM products p LEFT JOIN categories c ON c.id=p.category_id " "ORDER BY p.category_id, p.id") parts = ["πŸ†” All IDs β€” Categories & Products\n"] parts.append("\nπŸ“‚ Categories") if cats: for c in cats: parts.append(f" #{c['id']} β€” {c['name']}") else: parts.append(" β€”") parts.append("\nπŸ› Products") if prods: cur_cat = None for p in prods: if p["cat"] != cur_cat: cur_cat = p["cat"] parts.append(f"\n{cur_cat or 'Uncategorized'}") mark = "βœ…" if p["is_active"] else "🚫" parts.append( f" #{p['id']} {mark} {p['name']} β€” " f"{money(p['price'])} Β· stock {p['stock']}") else: parts.append(" β€”") parts.append( "\n\nEdit commands:\n" "/rename <id> Β· /setprice <id> Β· /setstock <id>\n" "/editdesc <id> Β· /editcontent <id> Β· /toggleproduct <id>\n" "/delproduct <id> Β· /rencat <id> Β· /delcat <id>") # Telegram 4096 char limit safety β€” chunk by lines so we never split a # tag or a {pe:ID}emoji marker in half. chunk, size = [], 0 LIMIT = 3500 for line in parts: ln = len(line) + 1 if size + ln > LIMIT and chunk: await update.message.reply_html("\n".join(chunk), disable_web_page_preview=True) chunk, size = [], 0 chunk.append(line) size += ln if chunk: await update.message.reply_html("\n".join(chunk), disable_web_page_preview=True) # ---------- Pending deposits ---------- @admin_only async def pending(update, context): rows = await DB.fetchall( "SELECT id, user_id, method, amount, unique_amount, tx_reference, " " detected_txid, status, created_at, expires_at " "FROM deposits " "WHERE status IN ('pending','pending_payment','payment_detected','confirming') " "ORDER BY id DESC LIMIT 30") if not rows: await update.message.reply_text("No pending deposits."); return lines = ["Pending Deposits", ""] for r in rows: method = (r["method"] or "").upper() tx = r["detected_txid"] or r["tx_reference"] or "" ua = r["unique_amount"] amt_line = (f"{money(r['amount'])}" + (f" (exact: {ua})" if ua else "")) lines.append( f"#{r['id']} {r['user_id']} [{method}] {amt_line}\n" f" status: {r['status']}" + (f" Β· tx: {tx[:24]}…" if tx else "") + f"\n created: {r['created_at'][:16]}" + (f" Β· expires: {r['expires_at'][:16]}" if r['expires_at'] else "") ) lines.append("\nCrypto deposits (TRC20/BEP20/BTC) are auto-detected. " "Use /approve only for Binance/Stars.") await update.message.reply_html("\n".join(lines)) # ---------- Tickets ---------- @admin_only async def tickets(update, context): rows = await DB.fetchall( "SELECT id, user_id, message, created_at FROM support_messages " "WHERE status='open' ORDER BY id DESC LIMIT 20") if not rows: await update.message.reply_text("No open tickets."); return text = "Open Tickets\n\n" + "\n".join( f"#{r['id']} {r['user_id']} {r['created_at'][:16]}\n{(r['message'] or '')[:160]}\n" for r in rows) await update.message.reply_html(text) # ---------- Announcements ---------- @admin_only async def announce(update, context): raw = update.message.text.split(" ", 1) if len(raw) < 2 or "|" not in raw[1]: await update.message.reply_text("Usage: /announce Title | Body text..."); return title, body = [s.strip() for s in raw[1].split("|", 1)] await DB.execute("INSERT INTO announcements(title, body) VALUES (?, ?)", (title, body)) await update.message.reply_text("βœ… Announcement published.") # ---------- Category edit/delete ---------- @admin_only async def rencat(update, context): if len(context.args) < 2: await update.message.reply_text("Usage: /rencat "); return cid = int(context.args[0]) name = " ".join(context.args[1:]).strip() try: await DB.execute("UPDATE categories SET name=? WHERE id=?", (name, cid)) await update.message.reply_text(f"βœ… Category #{cid} renamed to: {name}") except Exception: await update.message.reply_text("Name already exists.") @admin_only async def delcat(update, context): if not context.args: await update.message.reply_text("Usage: /delcat "); return cid = int(context.args[0]) await DB.execute("DELETE FROM categories WHERE id=?", (cid,)) await update.message.reply_text( f"πŸ—‘ Category #{cid} deleted. Products in it are now uncategorized.") # ---------- Product edit ---------- @admin_only async def setprice(update, context): if len(context.args) != 2: await update.message.reply_text("Usage: /setprice "); return pid = int(context.args[0]); price = float(context.args[1]) await DB.execute("UPDATE products SET price=? WHERE id=?", (price, pid)) await update.message.reply_text(f"βœ… Product #{pid} price = {money(price)}") @admin_only async def rename_product(update, context): if len(context.args) < 2: await update.message.reply_text("Usage: /rename "); return pid = int(context.args[0]); name = " ".join(context.args[1:]).strip() await DB.execute("UPDATE products SET name=? WHERE id=?", (name, pid)) await update.message.reply_text(f"βœ… Product #{pid} renamed to: {name}") @admin_only async def editdesc(update, context): if len(context.args) < 2: await update.message.reply_text("Usage: /editdesc "); return pid = int(context.args[0]); desc = " ".join(context.args[1:]) await DB.execute("UPDATE products SET description=? WHERE id=?", (desc, pid)) await update.message.reply_text(f"βœ… Product #{pid} description updated.") @admin_only async def editcontent(update, context): """Replace delivery_content (text / link / key) for a product.""" if len(context.args) < 2: await update.message.reply_text("Usage: /editcontent "); return pid = int(context.args[0]); content = " ".join(context.args[1:]) await DB.execute("UPDATE products SET delivery_content=? WHERE id=?", (content, pid)) await update.message.reply_text(f"βœ… Product #{pid} delivery content updated.")