diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..52b67d9 --- /dev/null +++ b/.env.example @@ -0,0 +1,6 @@ +BOT_TOKEN= +MARZBAN_URL=https://your-marzban.example.com +MARZBAN_USERNAME= +MARZBAN_PASSWORD= +ADMIN_IDS=12345678,87654321 +DB_PATH=stats.db diff --git a/README.md b/README.md new file mode 100644 index 0000000..283daa8 --- /dev/null +++ b/README.md @@ -0,0 +1,42 @@ +# Telegram VPN bot (Marzban + Telegram Stars) + +Async Telegram bot for selling and extending Marzban VPN subscriptions. + +## Features +- Plans: + - 1 month = ⭐️100 + - 3 months = ⭐️270 + - 6 months = ⭐️500 +- Marzban username format: `tg` (example: `tg12345678`). +- New users are created with: + - `data_limit = 50 GB` + - `data_limit_reset_strategy = month` + - `proxies = {"vless": {"flow": "xtls-rprx-vision"}}` + - `inbounds = {"vless": ["VLESS TCP REALITY"]}` +- If user has active subscription, a new purchase extends from current expiration. +- Bot sends **subscription URL** after successful purchase. +- Admin notifications for every creation/extension. +- `/stats` command for admins: + - payments count + - stars earned + - sold months + - marzban users + - traffic usage/total quota + +## Setup +```bash +python -m venv .venv +source .venv/bin/activate +pip install -r requirements.txt +cp .env.example .env +``` + +Fill `.env` and run: +```bash +export $(grep -v '^#' .env | xargs) +python bot.py +``` + +## Notes +- Telegram Stars invoices use `currency="XTR"` and empty `provider_token`. +- Local SQLite database stores only payment statistics. User state is read from Marzban API. diff --git a/bot.py b/bot.py new file mode 100644 index 0000000..3c2c86d --- /dev/null +++ b/bot.py @@ -0,0 +1,373 @@ +import asyncio +import calendar +import logging +import os +from dataclasses import dataclass +from datetime import datetime, timezone +from decimal import Decimal +from typing import Any + +import aiohttp +import aiosqlite +from aiogram import Bot, Dispatcher, F +from aiogram.enums import ParseMode +from aiogram.filters import Command +from aiogram.types import ( + CallbackQuery, + LabeledPrice, + Message, + PreCheckoutQuery, +) +from aiogram.utils.keyboard import InlineKeyboardBuilder + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +GB = 1024**3 +PLANS = { + 1: 100, + 3: 270, + 6: 500, +} + + +@dataclass +class Settings: + bot_token: str + marzban_url: str + marzban_username: str + marzban_password: str + admin_ids: set[int] + db_path: str = "stats.db" + + @classmethod + def from_env(cls) -> "Settings": + admin_raw = os.getenv("ADMIN_IDS", "") + admin_ids = { + int(part.strip()) + for part in admin_raw.split(",") + if part.strip().isdigit() + } + return cls( + bot_token=os.environ["BOT_TOKEN"], + marzban_url=os.environ["MARZBAN_URL"].rstrip("/"), + marzban_username=os.environ["MARZBAN_USERNAME"], + marzban_password=os.environ["MARZBAN_PASSWORD"], + admin_ids=admin_ids, + db_path=os.getenv("DB_PATH", "stats.db"), + ) + + +class MarzbanClient: + def __init__(self, base_url: str, username: str, password: str): + self.base_url = base_url + self.username = username + self.password = password + self._session: aiohttp.ClientSession | None = None + self._token: str | None = None + + async def __aenter__(self) -> "MarzbanClient": + self._session = aiohttp.ClientSession(base_url=self.base_url) + await self._authenticate() + return self + + async def __aexit__(self, exc_type, exc, tb): + if self._session: + await self._session.close() + + async def _authenticate(self) -> None: + assert self._session + resp = await self._session.post( + "/api/admin/token", + data={"username": self.username, "password": self.password}, + ) + resp.raise_for_status() + data = await resp.json() + self._token = data["access_token"] + + async def _request(self, method: str, path: str, **kwargs) -> Any: + assert self._session + if self._token is None: + await self._authenticate() + + headers = kwargs.pop("headers", {}) + headers["Authorization"] = f"Bearer {self._token}" + resp = await self._session.request(method, path, headers=headers, **kwargs) + + if resp.status == 401: + await self._authenticate() + headers["Authorization"] = f"Bearer {self._token}" + resp = await self._session.request(method, path, headers=headers, **kwargs) + + if resp.status == 404: + return None + + resp.raise_for_status() + if resp.content_type.startswith("application/json"): + return await resp.json() + return await resp.text() + + async def get_user(self, username: str) -> dict[str, Any] | None: + return await self._request("GET", f"/api/user/{username}") + + async def create_user(self, payload: dict[str, Any]) -> dict[str, Any]: + return await self._request("POST", "/api/user", json=payload) + + async def update_user(self, username: str, payload: dict[str, Any]) -> dict[str, Any]: + return await self._request("PUT", f"/api/user/{username}", json=payload) + + async def list_all_users(self) -> list[dict[str, Any]]: + users: list[dict[str, Any]] = [] + offset = 0 + limit = 100 + while True: + chunk = await self._request("GET", f"/api/users?offset={offset}&limit={limit}") + if not chunk: + break + batch = chunk.get("users", []) if isinstance(chunk, dict) else chunk + if not batch: + break + users.extend(batch) + if len(batch) < limit: + break + offset += len(batch) + return users + + +def add_months(dt: datetime, months: int) -> datetime: + month = dt.month - 1 + months + year = dt.year + month // 12 + month = month % 12 + 1 + day = min(dt.day, calendar.monthrange(year, month)[1]) + return dt.replace(year=year, month=month, day=day) + + +def bytes_to_gb(value: int | float) -> str: + return f"{Decimal(value) / Decimal(GB):.2f} GB" + + +class VpnBot: + def __init__(self, settings: Settings): + self.settings = settings + self.bot = Bot(settings.bot_token, parse_mode=ParseMode.HTML) + self.dp = Dispatcher() + self.marzban = MarzbanClient( + settings.marzban_url, + settings.marzban_username, + settings.marzban_password, + ) + self.db: aiosqlite.Connection | None = None + self.locks: dict[int, asyncio.Lock] = {} + self._register_handlers() + + async def setup(self) -> None: + self.db = await aiosqlite.connect(self.settings.db_path) + await self.db.execute( + """ + CREATE TABLE IF NOT EXISTS payments ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL, + tg_username TEXT, + marzban_username TEXT NOT NULL, + months INTEGER NOT NULL, + stars INTEGER NOT NULL, + action TEXT NOT NULL, + created_at TEXT NOT NULL + ) + """ + ) + await self.db.commit() + + async def run(self) -> None: + async with self.marzban: + await self.setup() + await self.dp.start_polling(self.bot) + + def _register_handlers(self) -> None: + self.dp.message.register(self.cmd_start, Command("start")) + self.dp.message.register(self.cmd_stats, Command("stats")) + self.dp.callback_query.register(self.choose_plan, F.data.startswith("buy:")) + self.dp.pre_checkout_query.register(self.pre_checkout) + self.dp.message.register(self.successful_payment, F.successful_payment) + + async def cmd_start(self, message: Message) -> None: + kb = InlineKeyboardBuilder() + for months, stars in PLANS.items(): + kb.button(text=f"{months} month — ⭐️{stars}", callback_data=f"buy:{months}") + kb.adjust(1) + await message.answer( + "VPN subscription plans:\n" + "• 1 month — 100 stars\n" + "• 3 months — 270 stars\n" + "• 6 months — 500 stars\n\n" + "Username in VPN panel: tg{your_telegram_id}", + reply_markup=kb.as_markup(), + ) + + async def choose_plan(self, query: CallbackQuery) -> None: + if not query.data or not query.message: + return + months = int(query.data.split(":", 1)[1]) + stars = PLANS[months] + await self.bot.send_invoice( + chat_id=query.message.chat.id, + title=f"VPN {months} month(s)", + description="Buy or extend VPN subscription", + payload=f"vpn:{months}", + provider_token="", + currency="XTR", + prices=[LabeledPrice(label=f"{months} month(s)", amount=stars)], + ) + await query.answer() + + async def pre_checkout(self, pre_checkout_query: PreCheckoutQuery) -> None: + payload = pre_checkout_query.invoice_payload + ok = payload.startswith("vpn:") + await self.bot.answer_pre_checkout_query( + pre_checkout_query.id, + ok=ok, + error_message="Invalid plan" if not ok else None, + ) + + async def successful_payment(self, message: Message) -> None: + payment = message.successful_payment + if not payment: + return + try: + months = int(payment.invoice_payload.split(":", 1)[1]) + except (IndexError, ValueError): + await message.answer("Payment received, but payload is invalid. Contact admin.") + return + + lock = self.locks.setdefault(message.from_user.id, asyncio.Lock()) + async with lock: + info = await self.buy_or_extend(message.from_user.id, months) + + await message.answer( + "✅ Subscription updated!\n" + f"Plan: {months} month(s), paid: ⭐️{payment.total_amount}\n" + f"Expires at (UTC): {info['expire_human']}\n" + f"Subscription URL:\n{info['subscription_url']}" + ) + + await self.save_payment( + user_id=message.from_user.id, + tg_username=message.from_user.username, + marzban_username=info["username"], + months=months, + stars=payment.total_amount, + action=info["action"], + ) + await self.notify_admins( + f"💸 User {message.from_user.id} ({message.from_user.full_name}) " + f"{info['action']} subscription for {months} month(s) and paid ⭐️{payment.total_amount}.\n" + f"Marzban username: {info['username']}\n" + f"Expires: {info['expire_human']}" + ) + + async def buy_or_extend(self, user_id: int, months: int) -> dict[str, str]: + username = f"tg{user_id}" + now = datetime.now(tz=timezone.utc) + expire_from = now + action = "created" + + current = await self.marzban.get_user(username) + if current: + current_expire = int(current.get("expire") or 0) + if current_expire > int(now.timestamp()): + expire_from = datetime.fromtimestamp(current_expire, tz=timezone.utc) + action = "extended" + + expire_at = add_months(expire_from, months) + payload = { + "username": username, + "status": "active", + "data_limit": 50 * GB, + "data_limit_reset_strategy": "month", + "expire": int(expire_at.timestamp()), + "proxies": {"vless": {"flow": "xtls-rprx-vision"}}, + "inbounds": {"vless": ["VLESS TCP REALITY"]}, + } + + if current: + updated = await self.marzban.update_user(username, payload) + else: + updated = await self.marzban.create_user(payload) + + subscription_url = updated.get("subscription_url") or "(not returned by API)" + return { + "username": username, + "action": action, + "subscription_url": subscription_url, + "expire_human": expire_at.strftime("%Y-%m-%d %H:%M:%S"), + } + + async def save_payment( + self, + user_id: int, + tg_username: str | None, + marzban_username: str, + months: int, + stars: int, + action: str, + ) -> None: + assert self.db + await self.db.execute( + """ + INSERT INTO payments (user_id, tg_username, marzban_username, months, stars, action, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + user_id, + tg_username, + marzban_username, + months, + stars, + action, + datetime.now(tz=timezone.utc).isoformat(), + ), + ) + await self.db.commit() + + async def notify_admins(self, text: str) -> None: + for admin_id in self.settings.admin_ids: + try: + await self.bot.send_message(admin_id, text) + except Exception as exc: + logger.warning("Failed to notify admin %s: %s", admin_id, exc) + + async def cmd_stats(self, message: Message) -> None: + if message.from_user.id not in self.settings.admin_ids: + await message.answer("Admins only") + return + + assert self.db + cursor = await self.db.execute( + "SELECT COUNT(*), COALESCE(SUM(stars), 0), COALESCE(SUM(months), 0) FROM payments" + ) + pay_count, stars_total, months_total = await cursor.fetchone() + + users = await self.marzban.list_all_users() + user_count = len(users) + used = sum(int(u.get("used_traffic") or 0) for u in users) + quota = sum(int(u.get("data_limit") or 0) for u in users) + + await message.answer( + "📊 Stats\n" + f"Payments: {pay_count}\n" + f"Stars earned: ⭐️{stars_total}\n" + f"Sold months: {months_total}\n" + f"Marzban users: {user_count}\n" + f"Traffic used: {bytes_to_gb(used)}\n" + f"Traffic quota: {bytes_to_gb(quota)}" + ) + + +async def main() -> None: + settings = Settings.from_env() + bot = VpnBot(settings) + await bot.run() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..70372d5 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +aiogram>=3.7.0 +aiohttp>=3.9.0 +aiosqlite>=0.20.0