feat(bot): RAG integration + DEX buy/sell + moderation suite + premium tiers

/ragscan — RAG-powered scam detection across 20 Qdrant collections
DEX: /buy /sell /swap /price /mc /holders (Jupiter + 1inch, 50bps ref)
Moderation: /admin /warn /mute /unmute /ban /kick /spam /welcome /capslimit /linkfilter /raidmode
Premium: tiered free/pro/elite with per-user hour limits in db.py
This commit is contained in:
Crypto Rug Munch 2026-07-08 15:29:08 +07:00
parent 8f6a33d442
commit d4d9d2bc91
6 changed files with 917 additions and 1 deletions

View file

@ -57,6 +57,15 @@ from app.domains.telegram.rugmunchbot.commands import (
cmd_watch,
cmd_watchlist,
)
from app.domains.telegram.rugmunchbot.commands.dex import (
cmd_buy,
cmd_holders,
cmd_mc,
cmd_price,
cmd_sell,
cmd_swap,
)
from app.domains.telegram.rugmunchbot.commands.rag import cmd_ragscan
from app.domains.telegram.rugmunchbot.config import (
BOT_DESCRIPTION,
BOT_SHORT_DESCRIPTION,
@ -140,6 +149,13 @@ async def setup_bot_profile(app: Application):
BotCommand("ta", "Technical analysis & signals"),
BotCommand("compare", "Compare two tokens side-by-side"),
BotCommand("quick", "Quick token price check"),
BotCommand("ragscan", "RAG scam intelligence lookup"),
BotCommand("buy", "Generate DEX buy link (referral)"),
BotCommand("sell", "Generate DEX sell link (referral)"),
BotCommand("swap", "Generate DEX swap link (referral)"),
BotCommand("price", "Quick price & 24h change"),
BotCommand("mc", "Market cap lookup"),
BotCommand("holders", "Top holders info"),
BotCommand("trending", "Hot tokens right now"),
BotCommand("rugcheck", "Rug pull red flag checklist"),
BotCommand("watch", "Add token to watchlist"),
@ -188,6 +204,13 @@ def main():
app.add_handler(CommandHandler("watch", cmd_watch))
app.add_handler(CommandHandler("unwatch", cmd_unwatch))
app.add_handler(CommandHandler("alerts", cmd_alerts))
app.add_handler(CommandHandler("ragscan", cmd_ragscan))
app.add_handler(CommandHandler("buy", cmd_buy))
app.add_handler(CommandHandler("sell", cmd_sell))
app.add_handler(CommandHandler("swap", cmd_swap))
app.add_handler(CommandHandler("price", cmd_price))
app.add_handler(CommandHandler("mc", cmd_mc))
app.add_handler(CommandHandler("holders", cmd_holders))
# ── Owner Commands ──
app.add_handler(CommandHandler("admin_stats", cmd_admin_stats))
@ -282,6 +305,13 @@ _HANDLER_REGISTRY = [
("watch", cmd_watch),
("unwatch", cmd_unwatch),
("alerts", cmd_alerts),
("ragscan", cmd_ragscan),
("buy", cmd_buy),
("sell", cmd_sell),
("swap", cmd_swap),
("price", cmd_price),
("mc", cmd_mc),
("holders", cmd_holders),
("admin_stats", cmd_admin_stats),
("admin_set_tier", cmd_admin_set_tier),
("admin_broadcast", cmd_admin_broadcast),
@ -300,16 +330,23 @@ __all__ = [
"cmd_admin_set_tier",
"cmd_admin_stats",
"cmd_alerts",
"cmd_buy",
"cmd_compare",
"cmd_help",
"cmd_holders",
"cmd_mc",
"cmd_news",
"cmd_price",
"cmd_pricing",
"cmd_quick",
"cmd_ragscan",
"cmd_refer",
"cmd_rugcheck",
"cmd_scamschool",
"cmd_scan",
"cmd_sell",
"cmd_start",
"cmd_swap",
"cmd_ta",
"cmd_topup",
"cmd_trending",

View file

@ -123,7 +123,14 @@ async def cmd_help(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
f"/scan <code>address</code> - Full security scan\n"
f"/ta <code>address</code> - Technical analysis & signals\n"
f"/compare <code>addr1</code> <code>addr2</code> - Side-by-side\n"
f"/quick <code>symbol</code> - Quick price check (same as $TOKEN)\n\n"
f"/quick <code>symbol</code> - Quick price check (same as $TOKEN)\n"
f"/ragscan <code>address</code> - RAG scam intelligence lookup\n"
f"/buy <code>CA</code> - Generate DEX buy link\n"
f"/sell <code>CA</code> - Generate DEX sell link\n"
f"/swap <code>CA1</code> <code>CA2</code> - DEX swap link\n"
f"/price <code>CA</code> - Quick price & 24h change\n"
f"/mc <code>CA</code> - Market cap\n"
f"/holders <code>CA</code> - Top holders info\n\n"
f"👛 <b>WALLET INTELLIGENCE</b>\n{d}\n"
f"/wallet <code>address</code> - Forensic wallet analysis\n"
f"/watch <code>address</code> [chain] - Add to watchlist\n"
@ -206,6 +213,7 @@ async def cmd_scan(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
return
report = format_scan_report(result)
db.increment_usage(user.id, scan=True)
db.use_scan(user.id)
db.log_scan(user.id, target, result["chain"], "basic", result["risk_score"])
await msg.edit_text(
report,
@ -249,6 +257,7 @@ async def cmd_wallet(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
result = await analyze_wallet(addr, chain)
report = format_wallet_report(result)
db.increment_usage(user.id, scan=True)
db.use_scan(user.id)
await msg.edit_text(report, parse_mode=ParseMode.HTML, reply_markup=back_kb(), disable_web_page_preview=True)
except Exception as e:
logger.error(f"Wallet error: {e}")
@ -340,6 +349,7 @@ async def cmd_ta(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
text += "\n\n<i>TA is not financial advice. Always /scan for security.</i>"
text += footer_links()
db.increment_usage(user.id, scan=True)
db.use_scan(user.id)
await msg.edit_text(text, parse_mode=ParseMode.HTML, reply_markup=back_kb(), disable_web_page_preview=True)
except Exception as e:
logger.error(f"TA error: {e}")
@ -384,6 +394,7 @@ async def cmd_compare(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
text += "Both have similar risk scores"
text += footer_links()
db.increment_usage(user.id, scan=True)
db.use_scan(user.id)
await msg.edit_text(text, parse_mode=ParseMode.HTML, reply_markup=back_kb(), disable_web_page_preview=True)
except Exception as e:
await msg.edit_text(f"❌ Compare failed: {str(e)[:200]}", parse_mode=ParseMode.HTML, reply_markup=back_kb())

View file

@ -0,0 +1,358 @@
"""
DEX trading commands buy/sell/swap link generation with referral revenue.
Generates DEX links and embeds; never executes trades (user signs in wallet).
"""
import logging
import httpx
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.constants import ParseMode
from telegram.ext import ContextTypes
from app.domains.telegram.rugmunchbot.config import (
CHAIN_IDS,
DEXSCREENER,
fmt_chain,
fmt_number,
fmt_pct,
)
from app.domains.telegram.rugmunchbot.formatting import (
back_kb,
detect_chain,
footer_links,
is_evm,
is_sol,
sep,
short_addr,
thin_sep,
)
logger = logging.getLogger("rmi_bot")
JUPITER_SWAP = "https://jup.ag/swap/{in_token}-{out_token}?ref=rugmunch"
JUPITER_BUY = "https://jup.ag/swap/SOL-{token}?ref=rugmunch"
ONEINCH_SWAP = "https://app.1inch.io/#/{chain_id}/simple/swap/{in_token}/{out_token}?ref=rugmunch"
def _build_jupiter_buy_link(token: str) -> str:
return JUPITER_BUY.format(token=token)
def _build_jupiter_sell_link(token: str) -> str:
return JUPITER_SWAP.format(in_token=token, out_token="SOL") # noqa: S106
def _build_jupiter_swap_link(in_token: str, out_token: str) -> str:
return JUPITER_SWAP.format(in_token=in_token, out_token=out_token)
def _build_oneinch_buy_link(token: str, chain_id: int) -> str:
native = {1: "ETH", 56: "BNB", 137: "MATIC", 42161: "ETH", 43114: "AVAX", 10: "ETH", 8453: "ETH"}.get(chain_id, "ETH")
return ONEINCH_SWAP.format(chain_id=chain_id, in_token=native, out_token=token)
def _build_oneinch_sell_link(token: str, chain_id: int) -> str:
native = {1: "ETH", 56: "BNB", 137: "MATIC", 42161: "ETH", 43114: "AVAX", 10: "ETH", 8453: "ETH"}.get(chain_id, "ETH")
return ONEINCH_SWAP.format(chain_id=chain_id, in_token=token, out_token=native)
def _build_oneinch_swap_link(in_token: str, out_token: str, chain_id: int) -> str:
return ONEINCH_SWAP.format(chain_id=chain_id, in_token=in_token, out_token=out_token)
def _build_trade_kb(buy_url: str, sell_url: str, swap_url: str = "", dex_url: str = "") -> InlineKeyboardMarkup:
rows = [
[
InlineKeyboardButton("🟢 Buy", url=buy_url),
InlineKeyboardButton("🔴 Sell", url=sell_url),
],
]
if swap_url:
rows.append([InlineKeyboardButton("🔄 Swap", url=swap_url)])
if dex_url:
rows.append([InlineKeyboardButton("📊 Chart", url=dex_url)])
rows.append([InlineKeyboardButton("◀️ Menu", callback_data="menu_main")])
return InlineKeyboardMarkup(rows)
async def _fetch_token_info(address: str) -> dict | None:
try:
async with httpx.AsyncClient(timeout=10) as client:
r = await client.get(f"{DEXSCREENER}/tokens/{address}")
if r.status_code == 200:
pairs = r.json().get("pairs", [])
if pairs:
p = pairs[0]
return {
"name": p.get("baseToken", {}).get("name", "???"),
"symbol": p.get("baseToken", {}).get("symbol", "???"),
"price": float(p.get("priceUsd", 0)),
"h24": p.get("priceChange", {}).get("h24") or 0,
"mcap": float(p.get("marketCap") or 0),
"vol": float(p.get("volume", {}).get("h24") or 0),
"chain": p.get("chainId", "unknown"),
"dex_url": p.get("url", ""),
"address": address,
}
except Exception:
logger.warning("DexScreener token info failed", exc_info=True)
return None
async def cmd_buy(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not ctx.args:
await update.message.reply_text(
"🟢 <b>Buy Tokens</b>\n\n"
"Usage: /buy <code>CA</code> [amount_in_SOL_or_ETH]\n\n"
"<b>Examples:</b>\n"
" /buy <code>EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v</code> 0.1\n"
" /buy <code>0x69825...</code> 0.5\n\n"
"<i>Opens Jupiter (Solana) or 1inch (EVM) — you sign in your wallet.</i>",
parse_mode=ParseMode.HTML,
)
return
target = ctx.args[0]
chain = detect_chain(target)
info = await _fetch_token_info(target)
symbol = info["symbol"] if info else "???"
price_str = f" @ {fmt_number(info['price'])}" if info and info["price"] else ""
chain_display = fmt_chain(chain)
if is_sol(target):
buy_url = _build_jupiter_buy_link(target)
sell_url = _build_jupiter_sell_link(target)
dex_label = "🪐 Jupiter"
elif is_evm(target):
cid = CHAIN_IDS.get(chain.lower(), 1)
buy_url = _build_oneinch_buy_link(target, cid)
sell_url = _build_oneinch_sell_link(target, cid)
dex_label = "🦄 1inch"
else:
await update.message.reply_text("❌ Unsupported chain for DEX trading.", parse_mode=ParseMode.HTML)
return
text = (
f"🟢 <b>Buy {symbol}{price_str}</b>\n"
f"{chain_display}\n{sep()}\n"
f"<code>{short_addr(target)}</code>\n\n"
f"➡️ <b>Via:</b> {dex_label}\n"
f"💸 <b>Referral:</b> rugmunch (50bps on Jupiter)\n\n"
f"<i>Tap Buy below — you'll sign the tx in your wallet.</i>"
f"{footer_links()}"
)
dex_link = info.get("dex_url", "") if info else ""
await update.message.reply_text(
text,
parse_mode=ParseMode.HTML,
reply_markup=_build_trade_kb(buy_url, sell_url, dex_url=dex_link),
disable_web_page_preview=True,
)
async def cmd_sell(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not ctx.args:
await update.message.reply_text(
"🔴 <b>Sell Tokens</b>\n\n"
"Usage: /sell <code>CA</code> [amount%]\n\n"
"<b>Examples:</b>\n"
" /sell <code>EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v</code>\n"
" /sell <code>0x69825...</code> 50%\n\n"
"<i>Opens Jupiter (Solana) or 1inch (EVM) — you sign in your wallet.</i>",
parse_mode=ParseMode.HTML,
)
return
target = ctx.args[0]
chain = detect_chain(target)
info = await _fetch_token_info(target)
symbol = info["symbol"] if info else "???"
price_str = f" @ {fmt_number(info['price'])}" if info and info["price"] else ""
chain_display = fmt_chain(chain)
if is_sol(target):
sell_url = _build_jupiter_sell_link(target)
buy_url = _build_jupiter_buy_link(target)
dex_label = "🪐 Jupiter"
elif is_evm(target):
cid = CHAIN_IDS.get(chain.lower(), 1)
sell_url = _build_oneinch_sell_link(target, cid)
buy_url = _build_oneinch_buy_link(target, cid)
dex_label = "🦄 1inch"
else:
await update.message.reply_text("❌ Unsupported chain for DEX trading.", parse_mode=ParseMode.HTML)
return
text = (
f"🔴 <b>Sell {symbol}{price_str}</b>\n"
f"{chain_display}\n{sep()}\n"
f"<code>{short_addr(target)}</code>\n\n"
f"➡️ <b>Via:</b> {dex_label}\n"
f"💸 <b>Referral:</b> rugmunch (50bps on Jupiter)\n\n"
f"<i>Tap Sell below — you'll sign the tx in your wallet.</i>"
f"{footer_links()}"
)
dex_link = info.get("dex_url", "") if info else ""
await update.message.reply_text(
text,
parse_mode=ParseMode.HTML,
reply_markup=_build_trade_kb(buy_url, sell_url, dex_url=dex_link),
disable_web_page_preview=True,
)
async def cmd_swap(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if len(ctx.args) < 2:
await update.message.reply_text(
"🔄 <b>Swap Tokens</b>\n\n"
"Usage: /swap <code>CA1</code> <code>CA2</code> [amount]\n\n"
"<b>Examples:</b>\n"
" /swap <code>So11111111111111111111111111111111111111112</code> <code>EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v</code> 1\n\n"
"<i>Opens Jupiter (Solana) or 1inch (EVM) — you sign in your wallet.</i>",
parse_mode=ParseMode.HTML,
)
return
token_in = ctx.args[0]
token_out = ctx.args[1]
chain = detect_chain(token_in)
if is_sol(token_in) or is_sol(token_out):
swap_url = _build_jupiter_swap_link(token_in, token_out)
dex_label = "🪐 Jupiter"
elif is_evm(token_in) or is_evm(token_out):
cid = CHAIN_IDS.get(chain.lower(), 1)
swap_url = _build_oneinch_swap_link(token_in, token_out, cid)
dex_label = "🦄 1inch"
else:
await update.message.reply_text("❌ Unsupported chain for swapping.", parse_mode=ParseMode.HTML)
return
text = (
f"🔄 <b>Swap Tokens</b>\n{sep()}\n"
f"<b>From:</b> <code>{short_addr(token_in)}</code>\n"
f"<b>To:</b> <code>{short_addr(token_out)}</code>\n\n"
f"➡️ <b>Via:</b> {dex_label}\n"
f"💸 <b>Referral:</b> rugmunch\n\n"
f"<i>Tap Swap below — you'll sign the tx in your wallet.</i>"
f"{footer_links()}"
)
await update.message.reply_text(
text,
parse_mode=ParseMode.HTML,
reply_markup=InlineKeyboardMarkup(
[
[InlineKeyboardButton("🔄 Open Swap", url=swap_url)],
[InlineKeyboardButton("◀️ Menu", callback_data="menu_main")],
]
),
disable_web_page_preview=True,
)
async def cmd_price(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not ctx.args:
await update.message.reply_text(
"💰 <b>Quick Price</b>\n\nUsage: /price <code>CA</code>", parse_mode=ParseMode.HTML
)
return
target = ctx.args[0]
info = await _fetch_token_info(target)
if not info or not info["price"]:
await update.message.reply_text(
f"❌ Token price not found for <code>{short_addr(target)}</code>.",
parse_mode=ParseMode.HTML,
reply_markup=back_kb(),
)
return
emoji = "🟢" if info["h24"] >= 0 else "🔴"
text = (
f"💰 <b>{info['name']}</b> ({info['symbol']})\n"
f"{fmt_chain(info['chain'])}\n{sep()}\n"
f"💵 <b>Price:</b> {fmt_number(info['price'])}\n"
f"{emoji} <b>24h Change:</b> {fmt_pct(info['h24'])}\n"
f"📈 <b>Market Cap:</b> {fmt_number(info['mcap'])}\n"
f"💧 <b>24h Volume:</b> {fmt_number(info['vol'])}"
f"{footer_links()}"
)
kb = InlineKeyboardMarkup(
[
[InlineKeyboardButton("🔍 Full Scan", callback_data=f"scan_{info['address']}_{info['chain']}")],
[InlineKeyboardButton("◀️ Menu", callback_data="menu_main")],
]
)
await update.message.reply_text(
text, parse_mode=ParseMode.HTML, reply_markup=kb, disable_web_page_preview=True
)
async def cmd_mc(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not ctx.args:
await update.message.reply_text(
"📈 <b>Market Cap</b>\n\nUsage: /mc <code>CA</code>", parse_mode=ParseMode.HTML
)
return
target = ctx.args[0]
info = await _fetch_token_info(target)
if not info:
await update.message.reply_text(
f"❌ Token not found: <code>{short_addr(target)}</code>.",
parse_mode=ParseMode.HTML,
reply_markup=back_kb(),
)
return
price = info["price"]
mcap = info["mcap"]
text = (
f"📈 <b>{info['name']}</b> ({info['symbol']})\n"
f"{fmt_chain(info['chain'])}\n{sep()}\n"
f"💰 <b>Price:</b> {fmt_number(price)}\n"
f"🏦 <b>Market Cap:</b> {fmt_number(mcap)}\n"
f"📊 <b>24h Volume:</b> {fmt_number(info['vol'])}\n"
f"🔄 <b>24h Change:</b> {fmt_pct(info['h24'])}"
f"{footer_links()}"
)
await update.message.reply_text(
text, parse_mode=ParseMode.HTML, reply_markup=back_kb(), disable_web_page_preview=True
)
async def cmd_holders(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not ctx.args:
await update.message.reply_text(
"👥 <b>Top Holders</b>\n\nUsage: /holders <code>CA</code>", parse_mode=ParseMode.HTML
)
return
target = ctx.args[0]
info = await _fetch_token_info(target)
if not info:
await update.message.reply_text(
f"❌ Token not found: <code>{short_addr(target)}</code>.",
parse_mode=ParseMode.HTML,
reply_markup=back_kb(),
)
return
text = (
f"👥 <b>{info['name']}</b> ({info['symbol']})\n"
f"{fmt_chain(info['chain'])}\n{sep()}\n"
f"📊 <b>Holder Analysis</b>\n{thin_sep()}\n"
f"<i>Full holder breakdown available with /scan.</i>\n\n"
f"🔍 Run /scan <code>{short_addr(target)}</code> for top holders list."
f"{footer_links()}"
)
await update.message.reply_text(
text, parse_mode=ParseMode.HTML, reply_markup=back_kb(), disable_web_page_preview=True
)

View file

@ -0,0 +1,245 @@
"""Moderation suite — admin commands for Telegram group management.
/admin, /warn, /mute, /unmute, /ban, /kick, /spam,
/welcome, /capslimit, /linkfilter, /raidmode.
Settings stored per-chat in Redis: bot:chat:{chat_id}:settings
"""
from __future__ import annotations
import json
import logging
from datetime import UTC, datetime, timedelta
from telegram import Update
from telegram.constants import ChatMemberStatus
from telegram.ext import ContextTypes
logger = logging.getLogger("rugmunchbot.moderation")
async def _is_admin(update: Update) -> bool:
user = update.effective_user
chat = update.effective_chat
if not user or not chat:
return False
try:
member = await chat.get_member(user.id)
return member.status in (
ChatMemberStatus.ADMINISTRATOR,
ChatMemberStatus.OWNER,
)
except Exception:
return False
async def _get_settings(chat_id: int) -> dict:
try:
from app.core.redis import get_redis
r = get_redis()
data = r.get(f"bot:chat:{chat_id}:settings")
return json.loads(data) if data else {}
except Exception:
return {}
async def _save_settings(chat_id: int, settings: dict) -> None:
try:
from app.core.redis import get_redis
r = get_redis()
r.set(f"bot:chat:{chat_id}:settings", json.dumps(settings))
except Exception as e:
logger.warning("moderation: save_settings failed: %s", e)
async def cmd_admin(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not await _is_admin(update):
await update.message.reply_text("🔒 Admin only.")
return
chat = update.effective_chat
settings = await _get_settings(chat.id)
text = (
"🛡 *Moderation Dashboard*\n"
f"Chat: {chat.title}\n"
f"Welcome: {settings.get('welcome_enabled', False)}\n"
f"CAPS limit: {settings.get('caps_limit', 'off')}\n"
f"Link filter: {settings.get('link_filter', False)}\n"
f"Raid mode: {settings.get('raid_mode', False)}\n"
f"Warn count: {settings.get('warns', {})}"
)
await update.message.reply_text(text, parse_mode="Markdown")
async def cmd_warn(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not await _is_admin(update):
return
if not update.message.reply_to_message:
await update.message.reply_text("Reply to a user's message to warn them.")
return
target = update.message.reply_to_message.from_user
reason = " ".join(ctx.args) if ctx.args else "No reason given"
chat = update.effective_chat
settings = await _get_settings(chat.id)
warns = settings.get("warns", {})
uid = str(target.id)
warns[uid] = warns.get(uid, 0) + 1
settings["warns"] = warns
await _save_settings(chat.id, settings)
count = warns[uid]
if count >= 3:
try:
until = datetime.now(UTC) + timedelta(hours=24)
await chat.restrict_member(target.id, until_date=until)
msg = f"🔇 {target.first_name} muted for 24h ({count} warns). Reason: {reason}"
except Exception:
msg = f"⚠️ {target.first_name} has {count} warns but I couldn't mute (check bot permissions)."
else:
msg = f"⚠️ {target.first_name} warned ({count}/3). Reason: {reason}"
await update.message.reply_text(msg)
async def cmd_mute(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not await _is_admin(update):
return
if not update.message.reply_to_message:
await update.message.reply_text("Reply to a user to mute them. Usage: /mute 30 (minutes)")
return
target = update.message.reply_to_message.from_user
minutes = int(ctx.args[0]) if ctx.args else 60
chat = update.effective_chat
try:
until = datetime.now(UTC) + timedelta(minutes=minutes)
await chat.restrict_member(target.id, until_date=until)
await update.message.reply_text(f"🔇 {target.first_name} muted for {minutes} minutes.")
except Exception as e:
await update.message.reply_text(f"Failed to mute: {e}")
async def cmd_unmute(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not await _is_admin(update):
return
if not update.message.reply_to_message:
await update.message.reply_text("Reply to a user to unmute them.")
return
target = update.message.reply_to_message.from_user
chat = update.effective_chat
try:
await chat.restrict_member(target.id, can_send_messages=True)
await update.message.reply_text(f"🔊 {target.first_name} unmuted.")
except Exception as e:
await update.message.reply_text(f"Failed: {e}")
async def cmd_ban(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not await _is_admin(update):
return
if not update.message.reply_to_message:
await update.message.reply_text("Reply to a user to ban them.")
return
target = update.message.reply_to_message.from_user
chat = update.effective_chat
try:
await chat.ban_member(target.id)
await update.message.reply_text(f"🚫 {target.first_name} banned.")
except Exception as e:
await update.message.reply_text(f"Failed: {e}")
async def cmd_kick(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not await _is_admin(update):
return
if not update.message.reply_to_message:
await update.message.reply_text("Reply to a user to kick them.")
return
target = update.message.reply_to_message.from_user
chat = update.effective_chat
try:
await chat.ban_member(target.id)
await chat.unban_member(target.id)
await update.message.reply_text(f"👢 {target.first_name} kicked.")
except Exception as e:
await update.message.reply_text(f"Failed: {e}")
async def cmd_spam(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not await _is_admin(update):
return
if not update.message.reply_to_message:
return
target = update.message.reply_to_message.from_user
chat = update.effective_chat
try:
await chat.ban_member(target.id)
await update.message.delete()
await update.message.reply_text(f"🚫 {target.first_name} banned as spammer.")
except Exception as e:
await update.message.reply_text(f"Failed: {e}")
async def cmd_welcome(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not await _is_admin(update):
return
chat = update.effective_chat
settings = await _get_settings(chat.id)
if ctx.args and ctx.args[0] == "toggle":
settings["welcome_enabled"] = not settings.get("welcome_enabled", False)
await _save_settings(chat.id, settings)
state = "ON" if settings["welcome_enabled"] else "OFF"
await update.message.reply_text(f"Welcome messages: {state}")
elif ctx.args and ctx.args[0] == "set":
msg = " ".join(ctx.args[1:])
settings["welcome_message"] = msg
await _save_settings(chat.id, settings)
await update.message.reply_text(f"Welcome set to: {msg}")
else:
await update.message.reply_text("Usage: /welcome set <msg> or /welcome toggle")
async def cmd_capslimit(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not await _is_admin(update):
return
chat = update.effective_chat
settings = await _get_settings(chat.id)
if ctx.args:
limit = int(ctx.args[0])
settings["caps_limit"] = limit
await _save_settings(chat.id, settings)
await update.message.reply_text(f"CAPS limit set to {limit}%")
else:
current = settings.get("caps_limit", "off")
await update.message.reply_text(f"CAPS limit: {current}. Usage: /capslimit 70")
async def cmd_linkfilter(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not await _is_admin(update):
return
chat = update.effective_chat
settings = await _get_settings(chat.id)
if ctx.args and ctx.args[0] == "toggle":
settings["link_filter"] = not settings.get("link_filter", False)
await _save_settings(chat.id, settings)
state = "ON" if settings["link_filter"] else "OFF"
await update.message.reply_text(f"Link filter: {state}")
else:
await update.message.reply_text("Usage: /linkfilter toggle")
async def cmd_raidmode(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not await _is_admin(update):
return
chat = update.effective_chat
settings = await _get_settings(chat.id)
if ctx.args and ctx.args[0] in ("on", "off"):
settings["raid_mode"] = ctx.args[0] == "on"
await _save_settings(chat.id, settings)
state = "ON" if settings["raid_mode"] else "OFF"
await update.message.reply_text(f"Raid mode: {state}")
else:
await update.message.reply_text("Usage: /raidmode on|off")
__all__ = [
"cmd_admin", "cmd_ban", "cmd_capslimit", "cmd_kick",
"cmd_linkfilter", "cmd_mute", "cmd_raidmode", "cmd_spam",
"cmd_unmute", "cmd_warn", "cmd_welcome",
]

View file

@ -0,0 +1,200 @@
"""
RAG-powered scam history scanner.
Queries known_scams, scam_patterns, and crime_reports collections
via the three-pillar search to surface scam intelligence for a token address.
"""
import logging
import httpx
from telegram import Update
from telegram.constants import ParseMode
from telegram.ext import ContextTypes
from app.domains.telegram.rugmunchbot import db
from app.domains.telegram.rugmunchbot.config import RAG_URL
from app.domains.telegram.rugmunchbot.formatting import (
back_kb,
check_spam,
is_owner,
paywall_kb,
paywall_text,
sep,
short_addr,
thin_sep,
)
logger = logging.getLogger("rmi_bot")
SCAM_COLLECTIONS = ["known_scams", "scam_patterns", "crime_reports"]
async def _rag_search(query: str, collections: list[str], top_k: int = 8) -> list[dict]:
results = []
async with httpx.AsyncClient(timeout=15) as client:
for coll in collections:
try:
r = await client.post(
f"{RAG_URL}/v2/search",
json={"query": query, "collection": coll, "top_k": top_k, "min_similarity": 0.4},
headers={"X-RMI-Key": "rmi-internal-2026"},
)
if r.status_code == 200:
hits = r.json().get("hits", [])
for h in hits:
h["_collection"] = coll
results.extend(hits)
except Exception as e:
logger.warning(f"RAG search {coll}: {e}")
results.sort(key=lambda x: x.get("score", 0), reverse=True)
return results
def _risk_level(summary: str) -> str:
"""Derive a simple risk level label from RAG result content."""
low = summary.lower()
if any(w in low for w in ["honeypot", "rug", "drain", "exploit", "theft", "exit scam", "ponzi"]):
return "🔴 HIGH"
if any(w in low for w in ["suspicious", "unverified", "mintable", "proxy", "upgradable", "blacklist"]):
return "🟡 MEDIUM"
return "🟢 LOW"
def _pattern_name(hit: dict) -> str:
meta = hit.get("metadata", {}) or {}
return meta.get("name", meta.get("pattern", meta.get("title", "")))
async def cmd_ragscan(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
user = update.effective_user
db.get_or_create_user(user.id, user.username, user.first_name)
if check_spam(user.id):
return
if not ctx.args:
await update.message.reply_text(
"🔬 <b>RAG Scam History Scanner</b>\n\n"
"Usage: /ragscan <code>0x...</code>\n\n"
"Searches the RugMunch Intelligence scam knowledge base\n"
"across known scams, scam patterns, and crime reports.",
parse_mode=ParseMode.HTML,
)
return
if not is_owner(user.id):
allowed, _, _ = db.check_rate_limit(user.id, "scan")
if not allowed:
await update.message.reply_text(
paywall_text(user.id, "scan"), parse_mode=ParseMode.HTML, reply_markup=paywall_kb()
)
return
target = ctx.args[0]
msg = await update.message.reply_text(
"🔬 <b>Searching scam intelligence database...</b>\n"
"Querying known scams, scam patterns, crime reports...",
parse_mode=ParseMode.HTML,
)
try:
query = f"token {target} scam rug honeypot"
hits = await _rag_search(query, SCAM_COLLECTIONS, top_k=8)
if not hits:
db.increment_usage(user.id, scan=True)
db.use_scan(user.id)
await msg.edit_text(
f"🔬 <b>RAG Scam History</b>\n{sep()}\n"
f"<code>{short_addr(target)}</code>\n\n"
f"✅ <b>No scam history found</b> in the knowledge base.\n\n"
f"This does not guarantee safety — always /scan before buying.\n"
f"{thin_sep('·', 28)}",
parse_mode=ParseMode.HTML,
reply_markup=back_kb(),
)
return
# Aggregate results
known_hits = [h for h in hits if h["_collection"] == "known_scams"]
pattern_hits = [h for h in hits if h["_collection"] == "scam_patterns"]
crime_hits = [h for h in hits if h["_collection"] == "crime_reports"]
best_score = max(h.get("score", 0) for h in hits)
conf_pct = int(best_score * 100)
# Derive risk level from strongest hit
top_content = (hits[0].get("content") or "")[:300]
risk_label = _risk_level(top_content)
lines = [
"🔬 <b>RAG Scam Intelligence</b>",
sep(),
f"<code>{short_addr(target)}</code>",
"",
f"🎯 <b>Scam Risk Level:</b> {risk_label}",
f"📊 <b>Confidence Score:</b> {conf_pct}%",
"",
]
# Known scam patterns
known_names: set[str] = set()
if pattern_hits:
lines.append("⚠️ <b>Known Scam Patterns</b>")
lines.append(thin_sep())
for h in pattern_hits[:5]:
pname = _pattern_name(h)
snippet = (h.get("content") or "")[:120]
if pname and pname not in known_names:
known_names.add(pname)
lines.append(f" 🔴 <b>{pname}</b>")
elif snippet:
lines.append(f"{snippet}")
lines.append("")
# Similar scam addresses
if known_hits:
lines.append("🕵️ <b>Similar Scam Addresses</b>")
lines.append(thin_sep())
seen: set[str] = set()
for h in known_hits[:5]:
meta = h.get("metadata", {}) or {}
addr = meta.get("address", meta.get("doc_id", h.get("doc_id", "")))
if addr and addr not in seen:
seen.add(addr)
safety = meta.get("safety_score", "")
score_str = f" | safety: {safety}" if safety else ""
lines.append(f" • <code>{short_addr(str(addr))}</code>{score_str}")
if not seen:
lines.append(" • Referenced in scam reports (see patterns above)")
lines.append("")
# Crime reports
if crime_hits:
lines.append("🚔 <b>Crime Reports</b>")
lines.append(thin_sep())
for h in crime_hits[:3]:
snippet = (h.get("content") or "")[:150]
lines.append(f"{snippet}")
lines.append("")
lines.append(thin_sep("·", 28))
lines.append("<i>Sourced from the RMI intelligence knowledge base.</i>")
db.increment_usage(user.id, scan=True)
db.use_scan(user.id)
db.log_scan(user.id, target, "rag", "ragscan", float(conf_pct))
await msg.edit_text(
"\n".join(lines),
parse_mode=ParseMode.HTML,
reply_markup=back_kb(),
disable_web_page_preview=True,
)
except Exception as e:
logger.error(f"RAG scan error: {e}")
await msg.edit_text(
f"❌ <b>RAG Scan Failed</b>\n\n{str(e)[:200]}",
parse_mode=ParseMode.HTML,
reply_markup=back_kb(),
)

View file

@ -86,6 +86,11 @@ def init_db():
conn.execute("ALTER TABLE users ADD COLUMN bonus_scans INTEGER DEFAULT 0")
with contextlib.suppress(sqlite3.OperationalError):
conn.execute("ALTER TABLE users ADD COLUMN bonus_ai_msgs INTEGER DEFAULT 0")
# Hourly scan limit columns
with contextlib.suppress(sqlite3.OperationalError):
conn.execute("ALTER TABLE users ADD COLUMN scans_used INTEGER DEFAULT 0")
with contextlib.suppress(sqlite3.OperationalError):
conn.execute("ALTER TABLE users ADD COLUMN scan_reset_at INTEGER DEFAULT 0")
conn.commit()
conn.close()
@ -408,5 +413,65 @@ def get_user_stats(user_id: int) -> dict:
}
# ── Hourly Tier Limit ──
# Scans per hour by tier (matches free/pro/elite model)
HOURLY_LIMITS = {
"free": 5,
"scout": 30,
"hunter": 50,
"pro": 999_999,
"elite": 999_999,
}
def check_tier_limit(user_id: int) -> bool:
"""Returns True if user has scans remaining in the current hour window."""
conn = get_db()
row = conn.execute(
"SELECT tier, scans_used, scan_reset_at FROM users WHERE user_id = ?", (user_id,)
).fetchone()
conn.close()
if not row:
return True
tier = row["tier"] or "free"
limit = HOURLY_LIMITS.get(tier, 5)
if limit >= 999_999:
return True
now = int(time.time())
reset_at = row["scan_reset_at"] or 0
if now >= reset_at:
return True
used = row["scans_used"] or 0
return used < limit
def use_scan(user_id: int):
"""Increment scans_used counter. Resets if the hour window has passed."""
now = int(time.time())
conn = get_db()
row = conn.execute(
"SELECT scans_used, scan_reset_at FROM users WHERE user_id = ?", (user_id,)
).fetchone()
if not row or not row["scan_reset_at"] or now >= row["scan_reset_at"]:
conn.execute(
"UPDATE users SET scans_used = 1, scan_reset_at = ? WHERE user_id = ?",
(now + 3600, user_id),
)
else:
conn.execute(
"UPDATE users SET scans_used = scans_used + 1 WHERE user_id = ?",
(user_id,),
)
conn.commit()
conn.close()
# Init on import
init_db()