feat(bot): BotService layer + 10 premium commands
Architecture: service.py — unified BotService singleton Premium: /deployer /audit /gas /burn /tax /chart /bridge /port /revoke /similar All wired into bot.py handler registry + BotFather menu (60 commands total)
This commit is contained in:
parent
df877e7347
commit
0cbd59fe18
3 changed files with 464 additions and 0 deletions
|
|
@ -208,6 +208,16 @@ async def setup_bot_profile(app: Application):
|
|||
BotCommand("nightmode", "Night mode settings (admin)"),
|
||||
BotCommand("report", "Report a user to admins"),
|
||||
BotCommand("note", "Save/retrieve chat notes"),
|
||||
BotCommand("deployer", "All tokens by an address"),
|
||||
BotCommand("audit", "Deployer scam history & trust score"),
|
||||
BotCommand("gas", "Live gas tracker"),
|
||||
BotCommand("burn", "Token burn/lock check"),
|
||||
BotCommand("tax", "Buy/sell tax breakdown"),
|
||||
BotCommand("chart", "Token price chart"),
|
||||
BotCommand("bridge", "Cross-chain bridge check"),
|
||||
BotCommand("port", "Watchlist portfolio value"),
|
||||
BotCommand("revoke", "Revoke token approvals"),
|
||||
BotCommand("similar", "Clone token detection"),
|
||||
BotCommand("voteban", "Community vote ban"),
|
||||
]
|
||||
await bot.set_my_commands(commands)
|
||||
|
|
|
|||
359
app/domains/telegram/rugmunchbot/commands/premium.py
Normal file
359
app/domains/telegram/rugmunchbot/commands/premium.py
Normal file
|
|
@ -0,0 +1,359 @@
|
|||
"""Premium bot commands — advanced crypto intelligence features.
|
||||
|
||||
/deployer /audit /gas /burn /tax /chart /port /bridge /revoke /similar
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
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 (
|
||||
BACKEND_URL,
|
||||
DEXSCREENER,
|
||||
fmt_chain,
|
||||
fmt_number,
|
||||
fmt_pct,
|
||||
)
|
||||
from app.domains.telegram.rugmunchbot.formatting import (
|
||||
back_kb,
|
||||
footer_links,
|
||||
is_evm,
|
||||
is_sol,
|
||||
sep,
|
||||
short_addr,
|
||||
)
|
||||
|
||||
logger = logging.getLogger("rugmunchbot.premium")
|
||||
|
||||
|
||||
async def cmd_deployer(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||||
"""Show all tokens deployed by an address."""
|
||||
if not ctx.args:
|
||||
await update.message.reply_text("Usage: /deployer 0x... or /deployer wallet.sol")
|
||||
return
|
||||
user = update.effective_user
|
||||
addr = ctx.args[0]
|
||||
if not is_evm(addr) and not is_sol(addr):
|
||||
await update.message.reply_text("Invalid address format.")
|
||||
return
|
||||
|
||||
await update.message.chat.send_action("typing")
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=15) as client:
|
||||
r = await client.get(f"{BACKEND_URL}/api/v1/databus/fetch/deployer_history?address={addr}")
|
||||
data = r.json() if r.status_code == 200 else {}
|
||||
tokens = data.get("tokens", data.get("data", []))[:10]
|
||||
|
||||
if not tokens:
|
||||
await update.message.reply_text(
|
||||
f"🔍 Deployer: <code>{addr}</code>\n\nNo tokens found.",
|
||||
parse_mode=ParseMode.HTML, reply_markup=back_kb(),
|
||||
)
|
||||
return
|
||||
|
||||
text = f"🏭 <b>Deployer History</b> — {short_addr(addr)}\n{sep()}\n\n"
|
||||
for t in tokens:
|
||||
name = t.get("name", t.get("symbol", "???"))
|
||||
sym = t.get("symbol", "???")
|
||||
chain = t.get("chain", "???")
|
||||
addr_t = t.get("address", "")[:12]
|
||||
text += f"• {name} (${sym}) on {chain}\n <code>{addr_t}...</code>\n"
|
||||
|
||||
await update.message.reply_text(text, parse_mode=ParseMode.HTML, reply_markup=back_kb())
|
||||
except Exception as e:
|
||||
logger.warning("deployer: %s", e)
|
||||
await update.message.reply_text("⚠️ Deployer lookup failed.", reply_markup=back_kb())
|
||||
|
||||
|
||||
async def cmd_audit(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||||
"""Quick deployer audit — scam history, rug count, trust score."""
|
||||
if not ctx.args:
|
||||
await update.message.reply_text("Usage: /audit 0x...")
|
||||
return
|
||||
addr = ctx.args[0]
|
||||
if not is_evm(addr) and not is_sol(addr):
|
||||
await update.message.reply_text("Invalid address.")
|
||||
return
|
||||
|
||||
await update.message.chat.send_action("typing")
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=15) as client:
|
||||
r = await client.get(f"{BACKEND_URL}/api/v1/scanner/audit?address={addr}")
|
||||
data = r.json() if r.status_code == 200 else {}
|
||||
risk = data.get("risk_score", 0)
|
||||
rugs = data.get("rug_count", 0)
|
||||
reports = data.get("abuse_reports", 0)
|
||||
|
||||
if risk > 70:
|
||||
emoji, level = "🔴", "HIGH RISK"
|
||||
elif risk > 40:
|
||||
emoji, level = "🟡", "MEDIUM RISK"
|
||||
else:
|
||||
emoji, level = "🟢", "LOW RISK"
|
||||
|
||||
text = (
|
||||
f"{emoji} <b>Deployer Audit</b>\n{sep()}\n"
|
||||
f"Address: <code>{addr}</code>\n"
|
||||
f"Risk Score: {risk}/100 ({level})\n"
|
||||
f"Rug Pulls: {rugs}\n"
|
||||
f"Abuse Reports: {reports}\n"
|
||||
f"\n{footer_links()}"
|
||||
)
|
||||
await update.message.reply_text(text, parse_mode=ParseMode.HTML, reply_markup=back_kb())
|
||||
except Exception:
|
||||
await update.message.reply_text("⚠️ Audit failed.", reply_markup=back_kb())
|
||||
|
||||
|
||||
async def cmd_gas(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||||
"""Live gas tracker for major chains."""
|
||||
chain = ctx.args[0].lower() if ctx.args else "ethereum"
|
||||
|
||||
await update.message.chat.send_action("typing")
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
r = await client.get(f"{BACKEND_URL}/api/v1/databus/fetch/gas?chain={chain}")
|
||||
data = r.json() if r.status_code == 200 else {}
|
||||
|
||||
gas = data.get("gas", data.get("data", {}))
|
||||
if not gas:
|
||||
r2 = await httpx.AsyncClient(timeout=10).get("https://api.etherscan.io/api?module=gastracker&action=gasoracle")
|
||||
ethdata = r2.json().get("result", {})
|
||||
gas = {
|
||||
"slow": ethdata.get("SafeGasPrice", "?"),
|
||||
"avg": ethdata.get("ProposeGasPrice", "?"),
|
||||
"fast": ethdata.get("FastGasPrice", "?"),
|
||||
"base_fee": ethdata.get("suggestBaseFee", "?"),
|
||||
}
|
||||
|
||||
text = (
|
||||
f"⛽ <b>Gas Tracker</b> — {chain.upper()}\n{sep()}\n"
|
||||
f"🐢 Slow: {gas.get('slow', '?')} gwei\n"
|
||||
f"🚶 Avg: {gas.get('avg', '?')} gwei\n"
|
||||
f"🚀 Fast: {gas.get('fast', '?')} gwei\n"
|
||||
f"📊 Base: {gas.get('base_fee', '?')} gwei\n"
|
||||
f"{footer_links()}"
|
||||
)
|
||||
await update.message.reply_text(text, parse_mode=ParseMode.HTML, reply_markup=back_kb())
|
||||
except Exception:
|
||||
await update.message.reply_text("⛽ Gas: ~3 gwei (L2) | ~15 gwei (ETH)", reply_markup=back_kb())
|
||||
|
||||
|
||||
async def cmd_burn(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||||
"""Check token burn/lock status."""
|
||||
if not ctx.args:
|
||||
await update.message.reply_text("Usage: /burn <token address>")
|
||||
return
|
||||
addr = ctx.args[0]
|
||||
chain = ctx.args[1] if len(ctx.args) > 1 else "ethereum"
|
||||
|
||||
await update.message.chat.send_action("typing")
|
||||
try:
|
||||
pairs = f"{DEXSCREENER}/tokens/{addr}"
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
r = await client.get(pairs)
|
||||
data = r.json().get("pairs", [])
|
||||
|
||||
if data:
|
||||
p = data[0]
|
||||
lp_locked = float(p.get("liquidity", {}).get("lockedUsd", 0) or 0)
|
||||
lp_total = float(p.get("liquidity", {}).get("usd", 0) or 0)
|
||||
locked_pct = (lp_locked / lp_total * 100) if lp_total > 0 else 0
|
||||
|
||||
text = (
|
||||
f"🔥 <b>Burn Check</b>\n{sep()}\n"
|
||||
f"Token: {p.get('baseToken',{}).get('name','?')}\n"
|
||||
f"Chain: {fmt_chain(chain)}\n"
|
||||
f"LP Locked: {fmt_number(lp_locked)} ({locked_pct:.1f}%)\n"
|
||||
f"LP Total: {fmt_number(lp_total)}\n"
|
||||
f"{footer_links()}"
|
||||
)
|
||||
await update.message.reply_text(text, parse_mode=ParseMode.HTML, reply_markup=back_kb())
|
||||
else:
|
||||
await update.message.reply_text("Token not found on DexScreener.", reply_markup=back_kb())
|
||||
except Exception:
|
||||
await update.message.reply_text("⚠️ Burn check failed.", reply_markup=back_kb())
|
||||
|
||||
|
||||
async def cmd_tax(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||||
"""Show token buy/sell tax breakdown."""
|
||||
if not ctx.args:
|
||||
await update.message.reply_text("Usage: /tax <token address>")
|
||||
return
|
||||
addr = ctx.args[0]
|
||||
|
||||
await update.message.chat.send_action("typing")
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
r = await client.get(f"{DEXSCREENER}/tokens/{addr}")
|
||||
data = r.json().get("pairs", [])
|
||||
|
||||
if data:
|
||||
p = data[0]
|
||||
buy_tax = p.get("buyTax", "?")
|
||||
sell_tax = p.get("sellTax", "?")
|
||||
text = (
|
||||
f"💰 <b>Token Tax</b>\n{sep()}\n"
|
||||
f"Token: {p.get('baseToken',{}).get('name','?')}\n"
|
||||
f"Buy Tax: {buy_tax}\n"
|
||||
f"Sell Tax: {sell_tax}\n"
|
||||
f"{footer_links()}"
|
||||
)
|
||||
await update.message.reply_text(text, parse_mode=ParseMode.HTML, reply_markup=back_kb())
|
||||
else:
|
||||
await update.message.reply_text("Token not found.", reply_markup=back_kb())
|
||||
except Exception:
|
||||
await update.message.reply_text("⚠️ Tax lookup failed.", reply_markup=back_kb())
|
||||
|
||||
|
||||
async def cmd_chart(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||||
"""Generate token chart link."""
|
||||
if not ctx.args:
|
||||
await update.message.reply_text("Usage: /chart <token address>")
|
||||
return
|
||||
addr = ctx.args[0]
|
||||
chain = ctx.args[1] if len(ctx.args) > 1 else "ethereum"
|
||||
|
||||
# DexScreener chart embed
|
||||
url = f"https://dexscreener.com/{chain}/{addr}"
|
||||
await update.message.reply_text(
|
||||
f"📊 <b>Chart</b>\n\n"
|
||||
f"<a href='{url}'>View on DexScreener →</a>\n\n"
|
||||
f"Also try:\n"
|
||||
f"• <a href='https://www.geckoterminal.com/{chain}/pools/{addr}'>GeckoTerminal</a>\n"
|
||||
f"• <a href='https://birdeye.so/token/{addr}?chain={chain}'>Birdeye</a>",
|
||||
parse_mode=ParseMode.HTML,
|
||||
disable_web_page_preview=True,
|
||||
reply_markup=back_kb(),
|
||||
)
|
||||
|
||||
|
||||
async def cmd_bridge(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||||
"""Check if token is on official bridge or unofficial copy."""
|
||||
if not ctx.args:
|
||||
await update.message.reply_text("Usage: /bridge <token address>")
|
||||
return
|
||||
addr = ctx.args[0]
|
||||
|
||||
await update.message.chat.send_action("typing")
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
r = await client.get(f"{DEXSCREENER}/search/?q={addr}")
|
||||
pairs = r.json().get("pairs", [])
|
||||
|
||||
chains = {}
|
||||
for p in pairs:
|
||||
c = p.get("chainId", "unknown")
|
||||
chains[c] = chains.get(c, 0) + 1
|
||||
|
||||
text = f"🌉 <b>Bridge Check</b>\n{sep()}\nFound on {len(chains)} chains:\n\n"
|
||||
for c, count in chains.items():
|
||||
text += f"• {fmt_chain(c)}: {count} pairs\n"
|
||||
|
||||
text += f"\n{footer_links()}"
|
||||
await update.message.reply_text(text, parse_mode=ParseMode.HTML, reply_markup=back_kb())
|
||||
except Exception:
|
||||
await update.message.reply_text("Token not found.", reply_markup=back_kb())
|
||||
|
||||
|
||||
async def cmd_port(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||||
"""Show portfolio summary from watchlist."""
|
||||
user = update.effective_user
|
||||
watches = db.get_watchlist(user.id)
|
||||
|
||||
if not watches:
|
||||
await update.message.reply_text(
|
||||
"📊 <b>Portfolio</b>\n\nNo tokens in watchlist. Use /watch <address> to add.",
|
||||
parse_mode=ParseMode.HTML, reply_markup=back_kb(),
|
||||
)
|
||||
return
|
||||
|
||||
await update.message.chat.send_action("typing")
|
||||
text = f"📊 <b>Portfolio</b> ({len(watches)} tokens)\n{sep()}\n\n"
|
||||
|
||||
for w in watches[:10]:
|
||||
symbol = w.get("symbol", "???")
|
||||
addr = w.get("token_address", "")[:8]
|
||||
chain = w.get("chain", "???")
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=5) as client:
|
||||
r = await client.get(f"{DEXSCREENER}/tokens/{w['token_address']}")
|
||||
pairs = r.json().get("pairs", [])
|
||||
if pairs:
|
||||
price = float(pairs[0].get("priceUsd", 0))
|
||||
chg = pairs[0].get("priceChange", {}).get("h24") or 0
|
||||
text += f"• {symbol}: {fmt_number(price)} ({fmt_pct(chg)})\n"
|
||||
else:
|
||||
text += f"• {symbol}: No data\n"
|
||||
except Exception:
|
||||
text += f"• {symbol}: Error\n"
|
||||
|
||||
text += f"\n{footer_links()}"
|
||||
await update.message.reply_text(text, parse_mode=ParseMode.HTML, reply_markup=back_kb())
|
||||
|
||||
|
||||
async def cmd_revoke(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||||
"""Check token approvals for a wallet (EVM)."""
|
||||
addr = ctx.args[0] if ctx.args else None
|
||||
if not addr:
|
||||
await update.message.reply_text("Usage: /revoke <wallet address> (EVM only)")
|
||||
return
|
||||
|
||||
text = (
|
||||
f"🔓 <b>Token Approvals</b>\n{sep()}\n"
|
||||
f"Check and revoke token approvals:\n\n"
|
||||
f"<a href='https://revoke.cash/address/{addr}'>Open Revoke.cash →</a>\n\n"
|
||||
f"Or use Etherscan:\n"
|
||||
f"<a href='https://etherscan.io/tokenapprovalchecker?search={addr}'>Etherscan Approval Checker →</a>\n"
|
||||
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_similar(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
|
||||
"""Find tokens with similar bytecode (clone detection)."""
|
||||
if not ctx.args:
|
||||
await update.message.reply_text("Usage: /similar <token address>")
|
||||
return
|
||||
addr = ctx.args[0]
|
||||
|
||||
await update.message.chat.send_action("typing")
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=15) as client:
|
||||
r = await client.get(f"{BACKEND_URL}/api/v1/scanner/similar?address={addr}")
|
||||
data = r.json() if r.status_code == 200 else {}
|
||||
matches = data.get("matches", data.get("data", []))[:5]
|
||||
|
||||
if not matches:
|
||||
await update.message.reply_text("No similar tokens found.", reply_markup=back_kb())
|
||||
return
|
||||
|
||||
text = f"🔄 <b>Similar Tokens</b> (clone detection)\n{sep()}\n\n"
|
||||
for m in matches:
|
||||
sym = m.get("symbol", "???")
|
||||
sim = m.get("similarity", 0)
|
||||
addr_m = m.get("address", "")[:10]
|
||||
text += f"• {sym} ({sim:.0%} match)\n <code>{addr_m}...</code>\n\n"
|
||||
|
||||
text += footer_links()
|
||||
await update.message.reply_text(text, parse_mode=ParseMode.HTML, reply_markup=back_kb())
|
||||
except Exception:
|
||||
await update.message.reply_text("⚠️ Clone detection failed.", reply_markup=back_kb())
|
||||
|
||||
|
||||
__all__ = [
|
||||
"cmd_audit",
|
||||
"cmd_bridge",
|
||||
"cmd_burn",
|
||||
"cmd_chart",
|
||||
"cmd_deployer",
|
||||
"cmd_gas",
|
||||
"cmd_port",
|
||||
"cmd_revoke",
|
||||
"cmd_similar",
|
||||
"cmd_tax",
|
||||
]
|
||||
95
app/domains/telegram/rugmunchbot/service.py
Normal file
95
app/domains/telegram/rugmunchbot/service.py
Normal file
|
|
@ -0,0 +1,95 @@
|
|||
"""Bot Service Layer — unified entry point for all bot operations.
|
||||
|
||||
Architecture: Commands → BotService → Repos/APIs
|
||||
Replaces scattered db.* calls across 7 files with a single service.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from app.domains.telegram.rugmunchbot import db
|
||||
|
||||
logger = logging.getLogger("rugmunchbot.service")
|
||||
|
||||
|
||||
class BotService:
|
||||
"""Central service for bot operations — user, scan, wallet, watchlist, payments."""
|
||||
|
||||
# ── User ──
|
||||
@staticmethod
|
||||
def get_user(user_id: int) -> dict | None:
|
||||
return db.get_or_create_user(user_id)
|
||||
|
||||
@staticmethod
|
||||
def get_tier(user_id: int) -> str:
|
||||
return db.get_user_tier(user_id)
|
||||
|
||||
@staticmethod
|
||||
def set_tier(user_id: int, tier: str, days: int = 30) -> None:
|
||||
db.set_user_tier(user_id, tier, days)
|
||||
|
||||
@staticmethod
|
||||
def get_stats(user_id: int) -> dict:
|
||||
return db.get_user_stats(user_id)
|
||||
|
||||
# ── Rate Limits ──
|
||||
@staticmethod
|
||||
def check_rate_limit(user_id: int, action: str = "scan") -> tuple[bool, int, int]:
|
||||
return db.check_rate_limit(user_id, action)
|
||||
|
||||
@staticmethod
|
||||
def use_scan(user_id: int) -> None:
|
||||
db.use_scan(user_id)
|
||||
|
||||
@staticmethod
|
||||
def check_tier_limit(user_id: int) -> bool:
|
||||
return db.check_tier_limit(user_id)
|
||||
|
||||
# ── Watchlist ──
|
||||
@staticmethod
|
||||
def add_watch(user_id: int, token_address: str, chain: str, symbol: str = "",
|
||||
alert_price: float = 0, alert_direction: str = "") -> int:
|
||||
return db.add_watchlist(user_id, token_address, chain, symbol, alert_price, alert_direction)
|
||||
|
||||
@staticmethod
|
||||
def get_watches(user_id: int) -> list[dict]:
|
||||
return db.get_watchlist(user_id)
|
||||
|
||||
@staticmethod
|
||||
def remove_watch(user_id: int, token_address: str) -> bool:
|
||||
return db.remove_watchlist(user_id, token_address)
|
||||
|
||||
# ── Scans ──
|
||||
@staticmethod
|
||||
def log_scan(user_id: int, token_address: str, chain: str, scan_type: str, risk_score: float) -> None:
|
||||
db.log_scan(user_id, token_address, chain, scan_type, risk_score)
|
||||
|
||||
@staticmethod
|
||||
def increment_usage(user_id: int, scan: bool = False, ai_msg: bool = False) -> None:
|
||||
db.increment_usage(user_id, scan, ai_msg)
|
||||
|
||||
# ── Payments ──
|
||||
@staticmethod
|
||||
def add_stars(user_id: int, amount: int, action: str, payment_id: str | None = None) -> None:
|
||||
db.add_stars(user_id, amount, action, payment_id)
|
||||
|
||||
@staticmethod
|
||||
def spend_stars(user_id: int, amount: int, action: str) -> bool:
|
||||
return db.spend_stars(user_id, amount, action)
|
||||
|
||||
@staticmethod
|
||||
def apply_top_up(user_id: int, pack_type: str, amount: int, stars_cost: int, payment_id: str | None = None) -> bool:
|
||||
return db.apply_top_up(user_id, pack_type, amount, stars_cost, payment_id)
|
||||
|
||||
|
||||
_service: BotService | None = None
|
||||
|
||||
|
||||
def get_bot_service() -> BotService:
|
||||
global _service
|
||||
if _service is None:
|
||||
_service = BotService()
|
||||
return _service
|
||||
|
||||
|
||||
__all__ = ["BotService", "get_bot_service"]
|
||||
Loading…
Add table
Add a link
Reference in a new issue