fix payment

This commit is contained in:
Yuriy Yuriev
2026-05-22 15:44:46 +07:00
parent 679cb40ec1
commit 61cf029744
8 changed files with 389 additions and 69 deletions
+2 -1
View File
@@ -3,7 +3,8 @@
"allow": [
"Bash(pip install *)",
"Bash(python -c \"import psutil; print\\('psutil ok, version:', psutil.__version__\\)\")",
"Bash(python -c ' *)"
"Bash(python -c ' *)",
"WebFetch(domain:doc.heleket.com)"
]
}
}
+6
View File
@@ -19,6 +19,11 @@ class Settings(BaseSettings):
TOPUP_MAX_RUB: int = 50000 # Максимальное пополнение ₽
CLICKS_MIN: int = 10 # Минимальная покупка переходов
# Heleket webhook
HELEKET_WEBHOOK_URL: str = "" # Публичный URL: https://example.com/heleket/webhook
HELEKET_WEBHOOK_PORT: int = 8080 # Локальный порт для приёма вебхуков
HELEKET_WHITELIST_IP: bool = True # Проверять IP 31.133.220.8
# Telethon/MTProto settings
API_ID: int = 12345
API_HASH: str = "abc123"
@@ -32,6 +37,7 @@ class Settings(BaseSettings):
# Browser Configuration
VIEWPORT_WIDTH: int = 1366
VIEWPORT_HEIGHT: int = 768
BROWSER_POOL_SIZE: int = 3 # Число параллельных браузерных воркеров
# Default values
DEFAULT_MIN_DELAY: int = 30
+9 -2
View File
@@ -25,6 +25,7 @@ from managers.background_tasks import BackgroundTaskManager
from managers.proxy_manager import ProxyManager
from managers.task_manager import TaskManager, TaskParams
from services.browser_service import BrowserService
from services.browser_pool import BrowserPool
from services.visit_service import VisitScheduler
from utils.helpers import parse_range, format_range, extract_domain
from utils.telegram import send_message_safe, send_visit_result
@@ -46,6 +47,7 @@ class BotInterface:
background_tasks: BackgroundTaskManager,
proxy_manager: ProxyManager,
browser_service: BrowserService,
browser_pool: BrowserPool = None,
storage: TaskStorage = None,
chat_storage: ChatStorage = None,
bot_ref=None,
@@ -53,6 +55,7 @@ class BotInterface:
self.background_tasks = background_tasks
self.proxy_manager = proxy_manager
self.browser_service = browser_service
self.browser_pool = browser_pool
self.task_manager = TaskManager()
self.storage = storage
self.chat_storage = chat_storage
@@ -595,6 +598,7 @@ class BotInterface:
return
payment_data = {
"user_id": uid,
"chat_id": callback.message.chat.id,
"payment_id": invoice["payment_id"],
"rub": amount,
"address": invoice.get("address"),
@@ -911,6 +915,7 @@ class BotInterface:
payment_data = {
"user_id": uid,
"chat_id": callback.message.chat.id,
"payment_id": invoice["payment_id"],
"clicks": clicks,
"rub": rub,
@@ -2197,7 +2202,8 @@ class BotInterface:
return
try:
result = await self.browser_service.visit_page(params.url, params.get_reading_time())
visitor = self.browser_pool or self.browser_service
result = await visitor.visit_page(params.url, params.get_reading_time())
params.total_visits += 1
if result.success:
params.successful_visits += 1
@@ -2307,7 +2313,8 @@ class BotInterface:
reading = params.get_reading_time()
try:
result = await self.browser_service.visit_page(url, reading)
visitor = self.browser_pool or self.browser_service
result = await visitor.visit_page(url, reading)
params.total_visits += 1
params.pending_visits = max(0, params.pending_visits - 1)
if result.success:
+42 -50
View File
@@ -1,10 +1,9 @@
"""
Нагрузочный тест: N параллельных браузеров на X секунд.
Нагрузочный тест: N визитов через BrowserPool.
Использование:
python load_test.py --n 5
python load_test.py --n 10 --url https://example.com --time 15
python load_test.py --n 10 --parallel 3 # не более 3 одновременных запусков
python load_test.py --n 10 --url https://example.com --time 15 --workers 3
"""
import asyncio
@@ -22,6 +21,7 @@ import psutil
from core.logger import setup_logger
from managers.proxy_manager import ProxyManager
from services.browser_service import BrowserService
from services.browser_pool import BrowserPool
logger = setup_logger(name="load_test", log_level="INFO", log_file="logs/load_test.log")
@@ -38,46 +38,39 @@ class BrowserResult:
success: bool
duration: float
error: Optional[str] = None
final_url: Optional[str] = None
async def run_one_browser(
async def submit_visit(
index: int,
browser_service: BrowserService,
pool: BrowserPool,
url: str,
visit_time: int,
results: List[BrowserResult],
launch_sem: asyncio.Semaphore,
) -> None:
# Семафор ограничивает сколько браузеров стартует одновременно.
# Когда браузер поднялся (AsyncCamoufox.__aenter__ завершён), семафор освобождается
# и следующий браузер может начать запуск. Так избегаем race на ресурсы ОС.
async with launch_sem:
t0 = time.perf_counter()
try:
result = await browser_service.visit_page(url, reading_time=visit_time)
duration = time.perf_counter() - t0
results.append(BrowserResult(
index=index,
success=result.success,
duration=duration,
error=result.error,
final_url=result.final_url,
))
status = "OK" if result.success else f"FAIL: {result.error}"
print(f" [Browser {index:>2}] {status}{duration:.1f}s | RAM: {_mem_mb():.0f} MB")
except Exception as e:
duration = time.perf_counter() - t0
results.append(BrowserResult(index=index, success=False, duration=duration, error=str(e)))
print(f" [Browser {index:>2}] EXCEPTION: {e}{duration:.1f}s")
t0 = time.perf_counter()
try:
result = await pool.visit_page(url, reading_time=visit_time)
duration = time.perf_counter() - t0
results.append(BrowserResult(
index=index,
success=result.success,
duration=duration,
error=result.error,
))
status = "OK" if result.success else f"FAIL: {result.error}"
print(f" [#{index:>2}] {status}{duration:.1f}s | RAM: {_mem_mb():.0f} MB")
except Exception as e:
duration = time.perf_counter() - t0
results.append(BrowserResult(index=index, success=False, duration=duration, error=str(e)))
print(f" [#{index:>2}] EXCEPTION: {e}{duration:.1f}s")
async def run_load_test(n: int, url: str, visit_time: int, parallel: int) -> None:
async def run_load_test(n: int, url: str, visit_time: int, workers: int) -> None:
print(f"\n{'='*60}")
print(f" Нагрузочный тест")
print(f" Браузеров: {n} | Одновременный запуск: {parallel} | URL: {url} | Время: {visit_time}s")
print(f" Визитов: {n} | Воркеров: {workers} | Время: {visit_time}s")
print(f" URL: {url}")
# Инфо о системе
mem = psutil.virtual_memory()
print(f" RAM всего: {mem.total / (1024**3):.1f} GB | Свободно: {mem.available / (1024**3):.1f} GB")
print(f" CPU ядер: {psutil.cpu_count()}")
@@ -87,24 +80,23 @@ async def run_load_test(n: int, url: str, visit_time: int, parallel: int) -> Non
await proxy_manager.detect_types()
proxy_count = proxy_manager.count
if proxy_count < n:
print(f" ВНИМАНИЕ: прокси {proxy_count} < браузеров {n}.")
print(f" Некоторые браузеры могут не получить прокси.\n")
if proxy_count < workers:
print(f" ВНИМАНИЕ: прокси {proxy_count} < воркеров {workers}.")
print(f" Часть воркеров может ждать свободный прокси.\n")
browser_service = BrowserService(proxy_manager)
results: List[BrowserResult] = []
pool = BrowserPool(size=workers, browser_service=browser_service)
await pool.start()
results: List[BrowserResult] = []
ram_before = _mem_mb()
print(f" RAM до старта: {ram_before:.0f} MB")
print(f"\n Запускаем {n} браузеров (волнами по {parallel})...\n")
print(f"\n Отправляем {n} визитов в пул ({workers} воркеров)...\n")
launch_sem = asyncio.Semaphore(parallel)
wall_start = time.perf_counter()
tasks = [
asyncio.create_task(
run_one_browser(i + 1, browser_service, url, visit_time, results, launch_sem)
)
asyncio.create_task(submit_visit(i + 1, pool, url, visit_time, results))
for i in range(n)
]
await asyncio.gather(*tasks, return_exceptions=True)
@@ -112,9 +104,9 @@ async def run_load_test(n: int, url: str, visit_time: int, parallel: int) -> Non
wall_elapsed = time.perf_counter() - wall_start
ram_after = _mem_mb()
await pool.stop()
await browser_service.cleanup()
# --- Итоги ---
ok = sum(1 for r in results if r.success)
fail = len(results) - ok
avg_dur = sum(r.duration for r in results) / len(results) if results else 0
@@ -124,34 +116,34 @@ async def run_load_test(n: int, url: str, visit_time: int, parallel: int) -> Non
print(f"\n{'='*60}")
print(f" ИТОГИ")
print(f"{'='*60}")
print(f" Всего браузеров : {n}")
print(f" Визитов всего : {n}")
print(f" Успешно : {ok}")
print(f" Ошибки : {fail}")
print(f" Общее время : {wall_elapsed:.1f}s")
print(f" Время посещения : avg={avg_dur:.1f}s min={min_dur:.1f}s max={max_dur:.1f}s")
print(f" Время визита : avg={avg_dur:.1f}s min={min_dur:.1f}s max={max_dur:.1f}s")
print(f" RAM до/после : {ram_before:.0f} MB → {ram_after:.0f} MB (Δ {ram_after - ram_before:+.0f} MB)")
print(f" RAM на браузер : ~{(ram_after - ram_before) / n if n else 0:.0f} MB")
print(f" RAM на воркер : ~{(ram_after - ram_before) / workers if workers else 0:.0f} MB")
if fail:
print(f"\n Ошибки:")
for r in results:
if not r.success:
print(f" Browser {r.index}: {r.error}")
print(f" #{r.index}: {r.error}")
print(f"{'='*60}\n")
def main():
parser = argparse.ArgumentParser(description="Load test: N параллельных браузеров")
parser.add_argument("--n", type=int, default=3, help="Всего браузеров (по умолчанию: 3)")
parser.add_argument("--parallel", type=int, default=2, help="Макс. одновременных запусков (по умолчанию: 2)")
parser = argparse.ArgumentParser(description="Load test через BrowserPool")
parser.add_argument("--n", type=int, default=5, help="Число визитов (по умолчанию: 5)")
parser.add_argument("--workers", type=int, default=2, help="Число параллельных воркеров/браузеров (по умолчанию: 2)")
parser.add_argument("--url", type=str, default="https://httpbin.org/headers", help="URL для посещения")
parser.add_argument("--time", type=int, default=10, dest="visit_time", help="Время посещения в секундах (по умолчанию: 10)")
args = parser.parse_args()
parallel = min(args.parallel, args.n)
workers = min(args.workers, args.n)
try:
asyncio.run(run_load_test(args.n, args.url, args.visit_time, parallel))
asyncio.run(run_load_test(args.n, args.url, args.visit_time, workers))
except KeyboardInterrupt:
print("\nПрервано.")
+23
View File
@@ -13,6 +13,8 @@ from core.logger import setup_logger
from managers.proxy_manager import ProxyManager
from managers.background_tasks import BackgroundTaskManager
from services.browser_service import BrowserService
from services.browser_pool import BrowserPool
from services.webhook_server import HelketWebhookServer
from handlers.commands import BotInterface
from managers.storage import TaskStorage, ChatStorage
@@ -45,13 +47,16 @@ class BotApplication:
self.proxy_manager = ProxyManager()
self.background_tasks = BackgroundTaskManager()
self.browser_service = BrowserService(self.proxy_manager)
self.browser_pool = BrowserPool(settings.BROWSER_POOL_SIZE, self.browser_service)
self.storage = TaskStorage()
self.chat_storage = ChatStorage()
self.webhook_server: HelketWebhookServer = None # создаём после инициализации bot
self.interface = BotInterface(
background_tasks=self.background_tasks,
proxy_manager=self.proxy_manager,
browser_service=self.browser_service,
browser_pool=self.browser_pool,
storage=self.storage,
chat_storage=self.chat_storage,
bot_ref=self,
@@ -82,6 +87,21 @@ class BotApplication:
# Определяем типы прокси без явного протокола
await self.proxy_manager.detect_types()
# Запускаем пул браузеров
await self.browser_pool.start()
logger.info(f"Browser pool: {settings.BROWSER_POOL_SIZE} workers")
# Запускаем webhook-сервер для Heleket (только если настроен)
if settings.HELEKET_API_KEY:
self.webhook_server = HelketWebhookServer(
payment_storage=self.interface.payment_storage,
balance_storage=self.interface.balance_storage,
rub_storage=self.interface.rub_storage,
bot=self.bot,
port=settings.HELEKET_WEBHOOK_PORT,
)
await self.webhook_server.start()
# Восстанавливаем задачи
await self._restore_tasks()
@@ -165,6 +185,9 @@ class BotApplication:
await self.storage.save_tasks(tasks)
await self.background_tasks.cancel_all()
await self.browser_pool.stop()
if self.webhook_server:
await self.webhook_server.stop()
await self.browser_service.cleanup()
if self.bot:
await self.bot.session.close()
+104
View File
@@ -0,0 +1,104 @@
"""
BrowserPool — N воркеров с очередью визитов.
Каждый воркер:
1. Берёт визит из очереди
2. Поднимает свежий браузер (через BrowserService)
3. Делает ровно 1 визит
4. Закрывает браузер
5. Идёт за следующим заданием
Пока воркер читает страницу, другие воркеры уже могут запускать
браузеры для следующих визитов — стартап перекрывается с чтением.
"""
import asyncio
import logging
from dataclasses import dataclass
from typing import Optional
from services.browser_service import BrowserService, VisitResult
logger = logging.getLogger(__name__)
@dataclass
class _VisitRequest:
url: str
reading_time: Optional[int]
future: asyncio.Future
class BrowserPool:
"""
Пул из N воркеров. Drop-in замена для BrowserService.visit_page().
"""
def __init__(self, size: int, browser_service: BrowserService):
self._size = size
self._bs = browser_service
self._queue: asyncio.Queue[Optional[_VisitRequest]] = asyncio.Queue()
self._workers: list[asyncio.Task] = []
self._running = False
async def start(self) -> None:
if self._running:
return
self._running = True
for i in range(self._size):
t = asyncio.create_task(self._worker(i), name=f"browser-pool-{i}")
self._workers.append(t)
logger.info(f"BrowserPool started: {self._size} workers")
async def stop(self) -> None:
if not self._running:
return
self._running = False
for _ in self._workers:
await self._queue.put(None) # sentinel — один на каждого воркера
await asyncio.gather(*self._workers, return_exceptions=True)
self._workers.clear()
logger.info("BrowserPool stopped")
async def visit_page(self, url: str, reading_time: int = None) -> VisitResult:
"""Отправляет визит в очередь и ждёт результата."""
if not self._running:
# Пул не запущен — fallback на прямой вызов
return await self._bs.visit_page(url, reading_time)
loop = asyncio.get_running_loop()
future: asyncio.Future[VisitResult] = loop.create_future()
await self._queue.put(_VisitRequest(url=url, reading_time=reading_time, future=future))
return await future
async def _worker(self, idx: int) -> None:
logger.debug(f"Pool worker-{idx} ready")
while True:
item = await self._queue.get()
if item is None: # sentinel → выход
break
try:
result = await self._bs.visit_page(item.url, item.reading_time)
if not item.future.done():
item.future.set_result(result)
except asyncio.CancelledError:
if not item.future.done():
item.future.cancel()
break
except Exception as e:
logger.error(f"Pool worker-{idx} error: {e}", exc_info=True)
if not item.future.done():
item.future.set_result(
VisitResult(url=item.url, success=False, error=str(e))
)
finally:
self._queue.task_done()
logger.debug(f"Pool worker-{idx} stopped")
@property
def size(self) -> int:
return self._size
@property
def queue_size(self) -> int:
return self._queue.qsize()
+38 -15
View File
@@ -1,7 +1,11 @@
"""Heleket payment integration."""
import base64
import hashlib
import json
import logging
from typing import Optional
import aiohttp
from config.settings import settings
@@ -9,6 +13,12 @@ from config.settings import settings
logger = logging.getLogger(__name__)
def _make_sign(body_bytes: bytes, api_key: str) -> str:
"""Генерирует подпись запроса: md5(base64(body) + api_key)."""
b64 = base64.b64encode(body_bytes).decode("utf-8")
return hashlib.md5((b64 + api_key).encode("utf-8")).hexdigest()
class HelketPayment:
"""Heleket payment gateway. Принимает сумму в рублях напрямую."""
@@ -19,6 +29,21 @@ class HelketPayment:
self.shop_id = settings.HELEKET_SHOP_ID
self._mock = not (self.api_key and self.shop_id)
def _post_headers(self, body_bytes: bytes) -> dict:
"""Заголовки для POST-запроса с подписью."""
return {
"merchant": self.api_key,
"sign": _make_sign(body_bytes, self.api_key),
"Content-Type": "application/json",
}
def _get_headers(self) -> dict:
"""Заголовки для GET-запроса (тело пустое)."""
return {
"merchant": self.api_key,
"sign": _make_sign(b"", self.api_key),
}
async def create_invoice(self, amount_rub: int, order_id: str) -> Optional[dict]:
"""
Создаёт счёт на сумму в рублях.
@@ -32,21 +57,20 @@ class HelketPayment:
"mock": True,
}
payload = {
"amount": str(amount_rub),
"currency": "RUB",
"order_id": order_id,
"url_callback": settings.HELEKET_WEBHOOK_URL,
}
body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
try:
async with aiohttp.ClientSession() as session:
payload = {
"shop_id": self.shop_id,
"amount": str(amount_rub),
"currency": "RUB",
"order_id": order_id,
"url_return": "",
"url_callback": "",
}
headers = {"merchant": self.api_key, "Content-Type": "application/json"}
async with session.post(
f"{self.BASE_URL}/payment",
json=payload,
headers=headers,
data=body,
headers=self._post_headers(body),
timeout=aiohttp.ClientTimeout(total=15),
) as resp:
data = await resp.json()
@@ -70,18 +94,17 @@ class HelketPayment:
return "pending"
try:
async with aiohttp.ClientSession() as session:
headers = {"merchant": self.api_key}
async with session.get(
f"{self.BASE_URL}/payment/info/{payment_id}",
headers=headers,
headers=self._get_headers(),
timeout=aiohttp.ClientTimeout(total=15),
) as resp:
data = await resp.json()
if data.get("state") == 0:
status = data.get("result", {}).get("status", "")
if status == "paid":
if status in ("paid", "paid_over"):
return "paid"
if status in ("cancel", "expired"):
if status in ("cancel", "expired", "fail", "wrong_amount"):
return "expired"
return "pending"
return "error"
+164
View File
@@ -0,0 +1,164 @@
"""
Webhook-сервер для приёма уведомлений от Heleket об оплате.
Heleket шлёт POST на url_callback когда статус счёта меняется.
Сервер верифицирует подпись, начисляет баланс и уведомляет пользователя в Telegram.
"""
import asyncio
import base64
import hashlib
import json
import logging
from typing import Optional
from aiohttp import web
from config.settings import settings
from managers.storage import PaymentStorage, BalanceStorage, RubleBalanceStorage
logger = logging.getLogger(__name__)
# Доверенный IP Heleket из документации
HELEKET_IP = "31.133.220.8"
def _verify_sign(payload: dict, api_key: str) -> bool:
"""Верифицирует подпись webhook-запроса от Heleket.
Алгоритм: md5( base64( json(payload_without_sign) ) + api_key )
ВАЖНО: PHP json_encode() экранирует '/' как '\/', Python — нет.
Heleket вычисляет подпись на PHP, поэтому вручную экранируем слэши.
"""
received_sign = payload.get("sign", "")
data = {k: v for k, v in payload.items() if k != "sign"}
data_json = json.dumps(data, separators=(",", ":"), ensure_ascii=False)
data_json = data_json.replace("/", "\\/") # имитируем PHP json_encode
data_b64 = base64.b64encode(data_json.encode("utf-8")).decode("utf-8")
expected = hashlib.md5((data_b64 + api_key).encode("utf-8")).hexdigest()
return expected == received_sign
class HelketWebhookServer:
"""
aiohttp сервер для вебхуков Heleket.
Запускается параллельно с Telegram-ботом.
"""
def __init__(
self,
payment_storage: PaymentStorage,
balance_storage: BalanceStorage,
rub_storage: RubleBalanceStorage,
bot, # aiogram Bot
port: int = 8080,
):
self._payment_storage = payment_storage
self._balance_storage = balance_storage
self._rub_storage = rub_storage
self._bot = bot
self._port = port
self._app = web.Application()
self._runner: Optional[web.AppRunner] = None
self._api_key = settings.HELEKET_API_KEY
self._app.router.add_post("/heleket/webhook", self._handle_webhook)
self._app.router.add_get("/health", self._handle_health)
async def start(self) -> None:
self._runner = web.AppRunner(self._app)
await self._runner.setup()
site = web.TCPSite(self._runner, "0.0.0.0", self._port)
await site.start()
logger.info(f"Heleket webhook server started on port {self._port}")
async def stop(self) -> None:
if self._runner:
await self._runner.cleanup()
logger.info("Heleket webhook server stopped")
async def _handle_health(self, request: web.Request) -> web.Response:
return web.Response(text="ok")
async def _handle_webhook(self, request: web.Request) -> web.Response:
# Проверяем IP (опционально — может быть за прокси/CDN)
client_ip = request.headers.get("X-Forwarded-For", request.remote)
if settings.HELEKET_WHITELIST_IP and client_ip and HELEKET_IP not in client_ip:
logger.warning(f"Webhook from unknown IP: {client_ip}")
return web.Response(status=403, text="Forbidden")
try:
payload = await request.json()
except Exception:
logger.warning("Webhook: invalid JSON body")
return web.Response(status=400, text="Bad Request")
logger.debug(f"Webhook payload: {payload}")
# Верифицируем подпись если есть API ключ
if self._api_key and not _verify_sign(payload, self._api_key):
logger.warning(f"Webhook: invalid signature. uuid={payload.get('uuid')}")
return web.Response(status=400, text="Invalid signature")
status = payload.get("status", "")
payment_uuid = payload.get("uuid", "")
# paid_over — клиент заплатил больше суммы счёта, тоже засчитываем
if status not in ("paid", "paid_over"):
logger.info(f"Webhook: uuid={payment_uuid} status={status} — ignored")
return web.Response(text="ok")
# Ищем платёж в хранилище
payment = await self._payment_storage.get(payment_uuid)
if not payment:
logger.warning(f"Webhook: payment {payment_uuid} not found in storage")
return web.Response(text="ok") # уже обработан или неизвестный
user_id = payment.get("user_id")
chat_id = payment.get("chat_id") or user_id
clicks = payment.get("clicks")
rub = payment.get("rub", 0)
try:
if clicks:
# Прямая покупка переходов
await self._balance_storage.add(user_id, clicks)
added_clicks = clicks
msg = (
f"✅ Оплата получена!\n\n"
f"💳 Сумма: {rub}\n"
f"🖱 Начислено: {clicks} переходов\n"
f"📊 Проверьте баланс в личном кабинете"
)
else:
# Пополнение рублёвого баланса → конвертируем в переходы
added_clicks = int(rub // settings.CLICK_PRICE_RUB)
await self._balance_storage.add(user_id, added_clicks)
await self._rub_storage.add(user_id, rub)
msg = (
f"✅ Пополнение получено!\n\n"
f"💳 Сумма: {rub}\n"
f"🖱 Начислено: {added_clicks} переходов "
f"({rub} ÷ {settings.CLICK_PRICE_RUB} ₽/переход)\n"
f"📊 Проверьте баланс в личном кабинете"
)
await self._payment_storage.delete(payment_uuid)
logger.info(
f"Webhook: payment {payment_uuid} confirmed — "
f"user={user_id} +{added_clicks} clicks"
)
# Уведомляем пользователя
if chat_id:
try:
await self._bot.send_message(chat_id, msg)
except Exception as e:
logger.warning(f"Webhook: failed to notify user {user_id}: {e}")
except Exception as e:
logger.error(f"Webhook: error processing payment {payment_uuid}: {e}", exc_info=True)
return web.Response(status=500, text="Internal error")
return web.Response(text="ok")