from __future__ import annotations
import json
import logging
import os
import re
import threading
import time
import uuid
from datetime import datetime, timedelta
from threading import Thread
from typing import TYPE_CHECKING
import requests
from FunPayAPI.common.enums import OrderStatuses
from telebot.types import CallbackQuery, InlineKeyboardButton as B, InlineKeyboardMarkup as K, Message
if TYPE_CHECKING:
from cardinal import Cardinal
logger = logging.getLogger("FPC.neverboost")
if not logger.handlers:
os.makedirs("logs", exist_ok=True)
file_handler = logging.FileHandler("logs/neverboost.log", encoding="utf-8")
file_handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s"))
logger.addHandler(file_handler)
logger.setLevel(logging.INFO)
NAME = "NeverBoost"
VERSION = "2.1.2"
DESCRIPTION = "Авто-выдача бустов 🙈."
CREDITS = "@neverboost | t.me/neverboost"
UUID = "7db2d6ee-5f37-41b8-ae0d-0533726e8ae4"
SETTINGS_PAGE = False
BIND_TO_DELETE = None
COMMANDS = ["boost", "boosts", "never", "neverboost"]
STATE_ADD_LOT = "neverboost_add_lot"
STATE_SET_API_KEY = "neverboost_set_api_key"
STATE_SET_BPU = "neverboost_set_bpu"
STATE_EDIT_BPU = "neverboost_edit_bpu"
STATE_SET_PRICE_RULE_ONE = "neverboost_set_price_rule_one"
STATE_SET_PRICE_RULE_THREE = "neverboost_set_price_rule_three"
STATE_MANUAL_INVITE = "neverboost_manual_invite"
STATE_MANUAL_QTY = "neverboost_manual_qty"
PENDING_BIND_PATH = "storage/plugins/neverboost_pending_bind.json"
CONFIG_PATH = "storage/plugins/neverboost.json"
PENDING_PATH = "storage/plugins/neverboost_pending.json"
PROCESSING_PATH = "storage/plugins/neverboost_processing.json"
SETTINGS_PATH = "storage/plugins/neverboost_settings.json"
STATS_PATH = "storage/plugins/neverboost_stats.json"
ORDERS_HISTORY_PATH = "storage/plugins/neverboost_orders_history.json"
EDIT_BIND_PATH = "storage/plugins/neverboost_edit_bind.json"
MANUAL_DRAFT_PATH = "storage/plugins/neverboost_manual_draft.json"
API_KEY_PATH = "storage/plugins/neverboost_api_key.txt"
BOOSTS_CONFIG_FALLBACK = "storage/plugins/boosts_notifier.json"
API_BASE_URL = "https://api.neverboost.com"
CHAT_ACTIVITY_PATH = "storage/plugins/neverboost_chat_activity.json"
ORDER_CHAT_MARKERS_PATH = "storage/plugins/neverboost_order_chats.json"
PLUGIN_VERSION_ENDPOINT = "/plugin/version"
PLUGIN_DOWNLOAD_ENDPOINT = "/plugin/download"
LOT_ID_RE = re.compile(r"(?:offer\?id=)?(\d+)")
AUTO_SYNC_LOCK = threading.Lock()
PLUGIN_VERSION_CHECK_LOCK = threading.Lock()
PLUGIN_AUTO_UPDATE_LOCK = threading.Lock()
RECENT_ORDER_IDS_LOCK = threading.Lock()
RECENT_ORDER_IDS: dict[str, float] = {}
PLUGIN_VERSION_LAST_DATA: dict | None = None
PLUGIN_VERSION_LAST_CHECK_TS = 0.0
PLUGIN_VERSION_LAST_OUTDATED = False
PLUGIN_OUTDATED_NOTIFIED_THIS_SESSION = False
PLUGIN_VERSION_CHECK_INTERVAL_SEC = 300
PLUGIN_AUTO_UPDATE_READY_VERSION: str | None = None
DELIVERY_TEXTS = {
# NeverBoost: первое сообщение покупателю после оплаты.
"ru": (
"👋 Спасибо за ваш заказ! Пожалуйста, отправьте ссылку на ваш Discord сервер для буста!\n"
"📦 К выдаче: {qty} буст(ов).\n"
"Пример: discord.gg/neverboost"
),
"en": (
"👋 Thanks for your order! Please send your Discord server invite link for boosting.\n"
"📦 To deliver: {qty} boost(s).\n"
"Example: discord.gg/neverboost"
),
}
BUYER_TEXTS = {
"ru": {
"purchase": DELIVERY_TEXTS["ru"],
"invalid": "❌ Не вижу корректной Discord-ссылки.\nОтправьте ссылку в формате: discord.gg/neverboost",
"bad": "❌ Ссылка недействительна или истекла.\nОтправьте новую ссылку-приглашение.",
"requests_on": "❌ На сервере включён вход по заявкам.\nОтключите заявки и отправьте новую ссылку.",
"api_fail": "❌ Не удалось создать заказ на выдачу бустов. Попробуйте снова.",
"pending_invite_choice": (
"⚠️ У вас уже есть заказ, где я жду Discord ссылку. Что делаем?\n\n"
"1 - Сейчас отправлю ссылку на первый заказ, потом нужно будет другой оформить\n"
"2 - Суммировать эти заказы для выдачи на 1 сервер\n\n"
"📦 Сейчас в первом заказе: {current_qty} буст(ов)\n"
"📦 Если суммировать: будет {total_qty} буст(ов)"
),
"pending_invite_choice_invalid": "⚠️ Ответьте цифрой 1 или 2, чтобы понять, что делать с новым заказом.",
"please_confirm_multiple": (
"🙏 Пожалуйста, подтвердите выполнение заказов кнопкой «Подтвердить заказ».\n\n"
"{orders}"
),
"unconfirmed_block": (
"⚠️ Братан, за тобой косяк 😅\n\n"
"Есть неподтвержденные заказы на FunPay. Пока не подтвердишь их, не могу обработать этот заказ.\n\n"
"🔗 Заказы к подтверждению:\n{orders}\n\n"
"После подтверждения напиши готово, и продолжим с этим заказом."
),
"unconfirmed_block_repeat": (
"⚠️ Братан, заказы всё ещё висят неподтвержденными 😅\n\n"
"{orders}\n\n"
"Подтверди и напиши готово, и продолжим с этим заказом."
),
"order_created": (
"🚀 Заказ на выдачу бустов успешно создан!\n\n"
"⏳ Бусты поступят в течении нескольких секунд.\n\n"
"🦇 Айди заказа: {api_order_id}"
),
"please_confirm": "🙏 Пожалуйста, подтвердите выполнение заказа кнопкой «Подтвердить заказ».",
"please_review": "⭐ Если всё понравилось, пожалуйста, оставьте отзыв о заказе. Это очень помогает.",
"status_completed": "✅ Бусты успешно выданы. Спасибо за заказ!",
"status_failed_refund": "❌ Бусты не начислены. Оформлен возврат.",
"status_failed_no_refund": "❌ Бусты не начислены. Автовозврат отключен, свяжитесь с продавцом.",
"status_wait_cancelled_refund": "😔 В связи с возвратом, заказ отменён.",
},
"en": {
"purchase": DELIVERY_TEXTS["en"],
"invalid": "❌ I can't find a valid Discord invite link.\nSend it like: discord.gg/neverboost",
"bad": "❌ This invite link is invalid or expired.\nPlease send a new invite link.",
"requests_on": "❌ Your server has join requests enabled.\nDisable it and send a new invite.",
"api_fail": "❌ Failed to create a boost order. Please try again.",
"pending_invite_choice": (
"⚠️ You already have an order where I'm waiting for your Discord invite link. What should we do?\n\n"
"1 - I'll send the link for the first order now, and handle the other one after that\n"
"2 - Combine these orders for delivery to one server\n\n"
"📦 Current first order: {current_qty} boost(s)\n"
"📦 If combined: total will be {total_qty} boost(s)"
),
"pending_invite_choice_invalid": "⚠️ Reply with 1 or 2 so I know what to do with the new order.",
"please_confirm_multiple": (
"🙏 Please confirm these orders by pressing “Confirm order”.\n\n"
"{orders}"
),
"unconfirmed_block": (
"⚠️ You still have unconfirmed FunPay orders 😅\n\n"
"There are unconfirmed orders on FunPay. Until you confirm them, I can't process this order.\n\n"
"🔗 Orders to confirm:\n{orders}\n\n"
"After confirming them, send any message and we'll continue with this order."
),
"unconfirmed_block_repeat": (
"⚠️ Your orders are still waiting for confirmation 😅\n\n"
"{orders}\n\n"
"Please confirm them and send any message, then we'll continue with this order."
),
"order_created": (
"🚀 Boost delivery order was created successfully!\n\n"
"⏳ Boosts will arrive within a few seconds.\n\n"
"🦇 Order ID: {api_order_id}"
),
"please_confirm": "🙏 Please confirm the order by pressing “Confirm order”.",
"please_review": "⭐ If everything is good, please leave a review for this order. It helps a lot.",
"status_completed": "✅ Boosts were delivered successfully. Thank you for your order!",
"status_failed_refund": "❌ Boosts were not delivered. A refund has been issued.",
"status_failed_no_refund": "❌ Boosts were not delivered. Auto-refund is disabled, contact the seller.",
"status_wait_cancelled_refund": "😔 The order was cancelled due to a refund.",
},
}
WELCOME_TEXTS = {
"ru": (
"👋 Приветствую! На лотах с бустами стоит авто-выдача!\n\n"
"📦 Наличие бустов:\n"
"🗓 1 месяц - {one_month} шт.\n"
"🗓 3 месяца - {three_month} шт.\n\n"
"💬 Для просмотра наличия бустов отправьте #stock"
),
"en": (
"👋 Welcome! Boost lots use automatic delivery!\n\n"
"📦 Boost stock:\n"
"🗓 1 month - {one_month}\n"
"🗓 3 months - {three_month}\n\n"
"💬 Send #stock to view current stock"
),
}
def load_json(path: str, default):
if not os.path.exists(path):
return default
try:
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
return data
except Exception:
logger.debug("TRACEBACK", exc_info=True)
return default
def save_json(path: str, data) -> None:
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def get_chat_activity() -> dict:
data = load_json(CHAT_ACTIVITY_PATH, {})
return data if isinstance(data, dict) else {}
def set_chat_activity(chat_id: int | str, payload: dict) -> None:
data = get_chat_activity()
data[str(chat_id)] = payload
save_json(CHAT_ACTIVITY_PATH, data)
def get_order_chat_markers() -> dict:
data = load_json(ORDER_CHAT_MARKERS_PATH, {})
return data if isinstance(data, dict) else {}
def mark_order_chat(chat_id: int | str, payload: dict) -> None:
data = get_order_chat_markers()
data[str(chat_id)] = payload
save_json(ORDER_CHAT_MARKERS_PATH, data)
def has_recent_order_chat(chat_id: int | str, ttl_sec: int = 86400) -> bool:
data = get_order_chat_markers()
item = data.get(str(chat_id))
if not isinstance(item, dict):
return False
try:
ts = float(item.get("ts", 0.0) or 0.0)
except Exception:
ts = 0.0
return ts > 0 and (time.time() - ts) <= ttl_sec
def has_recent_order_chat_for_message(message, ttl_sec: int = 86400) -> bool:
data = get_order_chat_markers()
direct = data.get(str(getattr(message, "chat_id", "")))
if isinstance(direct, dict):
try:
ts = float(direct.get("ts", 0.0) or 0.0)
except Exception:
ts = 0.0
if ts > 0 and (time.time() - ts) <= ttl_sec:
return True
author = str(getattr(message, "author", "") or "").strip().lower()
for _, item in data.items():
if not isinstance(item, dict):
continue
buyer = str(item.get("buyer_username", "") or "").strip().lower()
try:
ts = float(item.get("ts", 0.0) or 0.0)
except Exception:
ts = 0.0
if author and buyer and author == buyer and ts > 0 and (time.time() - ts) <= ttl_sec:
return True
return False
def get_chat_activity_row(chat_id: int | str) -> dict | None:
data = get_chat_activity()
item = data.get(str(chat_id))
return item if isinstance(item, dict) else None
def should_send_welcome(chat_id: int | str, ttl_sec: int = 86400) -> bool:
row = get_chat_activity_row(chat_id)
if not row:
return True
try:
ts = float(row.get("ts", 0.0) or 0.0)
except Exception:
ts = 0.0
return ts <= 0 or (time.time() - ts) > ttl_sec
def create_pending_from_order(cardinal: Cardinal, order_obj) -> bool:
if order_obj is None:
return False
current_username = str(getattr(cardinal.account, "username", "") or "").strip().lower()
buyer_username = str(getattr(order_obj, "buyer_username", "") or "").strip().lower()
if current_username and buyer_username and current_username == buyer_username:
logger.info("NeverBoost fallback ignored: current account is buyer for order %s", getattr(order_obj, "id", None))
return False
duration = detect_duration(order_obj)
lot_name = str(getattr(order_obj, "description", "") or "")
binding = find_binding_for_text(lot_name, duration)
if binding is None:
logger.info(
"NeverBoost binding not matched for order fallback: id=%s duration=%s lot=%s",
getattr(order_obj, "id", None),
duration,
lot_name,
)
return False
boosts_per_unit = max(int(binding.get("boosts_per_unit", 1) or 1), 1)
requested_boosts = max(int(getattr(order_obj, "amount", 1) or 1), 1) * boosts_per_unit
locale = detect_order_locale(cardinal, order_obj)
mark_order_chat(order_obj.chat_id, {"ts": time.time(), "order_id": order_obj.id, "buyer_username": order_obj.buyer_username})
set_pending(order_obj.chat_id, {
"order_id": order_obj.id,
"buyer_username": order_obj.buyer_username,
"locale": locale,
"qty": requested_boosts,
"duration": duration,
"lot": binding["lot"],
"boosts_per_unit": boosts_per_unit,
})
logger.info(
"NeverBoost pending created from fallback: id=%s buyer=%s boosts=%s duration=%s lot=%s",
order_obj.id,
order_obj.buyer_username,
requested_boosts,
duration,
binding["lot"],
)
return True
def create_pending_from_payment_text(cardinal: Cardinal, chat_id: int | str, text: str) -> bool:
order_match = re.search(r"заказ\s+#([A-Z0-9]+)", text, flags=re.IGNORECASE)
buyer_match = re.search(r"Покупатель\s+(.+?)\s+оплатил заказ", text, flags=re.IGNORECASE)
qty_match = re.search(r"(\d+)\s*шт", text, flags=re.IGNORECASE)
order_id = order_match.group(1) if order_match else ""
buyer_username = buyer_match.group(1).strip() if buyer_match else ""
current_username = str(getattr(cardinal.account, "username", "") or "").strip().lower()
if current_username and buyer_username and current_username == buyer_username.lower():
logger.info("NeverBoost fallback payment text ignored: current account is buyer for order %s", order_id)
return False
qty_units = max(int(qty_match.group(1)) if qty_match else 1, 1)
duration = "threeMonth" if ("3 месяц" in text.lower() or "3 мес" in text.lower()) else "oneMonth"
binding = find_binding_for_text(text, duration)
if binding is None:
logger.info(
"NeverBoost binding not matched for payment text fallback: order=%s chat=%s duration=%s text=%s",
order_id,
chat_id,
duration,
text[:200],
)
return False
boosts_per_unit = max(int(binding.get("boosts_per_unit", 1) or 1), 1)
requested_boosts = qty_units * boosts_per_unit
locale = extract_chat_language(cardinal, chat_id, buyer_username) or "ru"
mark_order_chat(chat_id, {"ts": time.time(), "order_id": order_id, "buyer_username": buyer_username})
set_pending(chat_id, {
"order_id": order_id,
"buyer_username": buyer_username,
"locale": locale,
"qty": requested_boosts,
"duration": duration,
"lot": binding["lot"],
"boosts_per_unit": boosts_per_unit,
})
logger.info(
"NeverBoost pending created from payment text: id=%s buyer=%s boosts=%s duration=%s lot=%s",
order_id,
buyer_username,
requested_boosts,
duration,
binding["lot"],
)
return True
def load_plugin_data() -> dict:
data = load_json(CONFIG_PATH, {"bindings": []})
bindings = data.get("bindings", []) if isinstance(data, dict) else []
result = []
if isinstance(bindings, list):
for item in bindings:
if not isinstance(item, dict):
continue
lot = str(item.get("lot", "") or "").strip()
lot_id = str(item.get("lot_id", "") or "").strip()
duration = str(item.get("duration", "oneMonth") or "oneMonth")
boosts_per_unit = max(int(item.get("boosts_per_unit", 1) or 1), 1)
if not lot:
continue
if duration not in {"oneMonth", "threeMonth"}:
duration = "oneMonth"
result.append({"lot": lot, "lot_id": lot_id, "duration": duration, "boosts_per_unit": boosts_per_unit})
return {"bindings": result}
def get_bindings() -> list[dict]:
return load_plugin_data()["bindings"]
def save_bindings(bindings: list[dict]) -> None:
normalized = []
seen = set()
for item in bindings:
if not isinstance(item, dict):
continue
lot = str(item.get("lot", "") or "").strip()
duration = str(item.get("duration", "oneMonth") or "oneMonth")
lot_id = str(item.get("lot_id", "") or "").strip()
boosts_per_unit = max(int(item.get("boosts_per_unit", 1) or 1), 1)
if not lot:
continue
if duration not in {"oneMonth", "threeMonth"}:
duration = "oneMonth"
key = (lot.lower(), duration, boosts_per_unit)
if key in seen:
continue
seen.add(key)
normalized.append({"lot": lot, "lot_id": lot_id, "duration": duration, "boosts_per_unit": boosts_per_unit})
save_json(CONFIG_PATH, {"bindings": normalized})
def get_settings() -> dict:
data = load_json(SETTINGS_PATH, {
"auto_sync_enabled": True,
"auto_sync_interval_sec": 60,
"welcome_enabled": True,
"stock_command_enabled": True,
"price_rule_one_month": {"type": "off", "value": 0.0},
"price_rule_three_month": {"type": "off", "value": 0.0},
})
if not isinstance(data, dict):
return {
"auto_sync_enabled": True,
"auto_sync_interval_sec": 60,
"welcome_enabled": True,
"stock_command_enabled": True,
"price_rule_one_month": {"type": "off", "value": 0.0},
"price_rule_three_month": {"type": "off", "value": 0.0},
}
def normalize_price_rule(raw: dict | None) -> dict:
if not isinstance(raw, dict):
return {"type": "off", "value": 0.0}
rule_type = str(raw.get("type", "off") or "off").strip()
if rule_type not in {"off", "percent", "fixed"}:
rule_type = "off"
try:
value = float(raw.get("value", 0.0) or 0.0)
except Exception:
value = 0.0
if rule_type == "off":
value = 0.0
return {"type": rule_type, "value": round(value, 2)}
return {
"auto_sync_enabled": bool(data.get("auto_sync_enabled", True)),
"auto_sync_interval_sec": max(int(data.get("auto_sync_interval_sec", 60) or 60), 10),
"welcome_enabled": bool(data.get("welcome_enabled", True)),
"stock_command_enabled": bool(data.get("stock_command_enabled", True)),
"price_rule_one_month": normalize_price_rule(data.get("price_rule_one_month")),
"price_rule_three_month": normalize_price_rule(data.get("price_rule_three_month")),
}
def save_settings(data: dict) -> None:
settings = get_settings()
settings.update(data)
save_json(SETTINGS_PATH, settings)
def get_stats_rows() -> list[dict]:
data = load_json(STATS_PATH, [])
if not isinstance(data, list):
return []
result = []
for item in data:
if not isinstance(item, dict):
continue
try:
result.append({
"ts": float(item.get("ts", 0.0) or 0.0),
"boosts": max(int(item.get("boosts", 0) or 0), 0),
"profit": round(float(item.get("profit", 0.0) or 0.0), 2),
})
except Exception:
logger.debug("TRACEBACK", exc_info=True)
return result
def save_stats_rows(rows: list[dict]) -> None:
save_json(STATS_PATH, rows)
def get_orders_history_rows() -> list[dict]:
data = load_json(ORDERS_HISTORY_PATH, [])
if not isinstance(data, list):
return []
return [item for item in data if isinstance(item, dict)]
def save_orders_history_rows(rows: list[dict]) -> None:
save_json(ORDERS_HISTORY_PATH, rows)
def upsert_orders_history_row(payload: dict) -> None:
rows = get_orders_history_rows()
api_order_id = str(payload.get("api_order_id", "") or "").strip()
source_order_id = str(payload.get("source_order_id", "") or "").strip()
if not api_order_id and not source_order_id:
return
updated = False
for index, row in enumerate(rows):
if api_order_id and str(row.get("api_order_id", "") or "").strip() == api_order_id:
merged = dict(row)
merged.update(payload)
if row.get("created_ts") and payload.get("created_ts"):
merged["created_ts"] = row.get("created_ts")
rows[index] = merged
updated = True
break
if not api_order_id and source_order_id and str(row.get("source_order_id", "") or "").strip() == source_order_id:
merged = dict(row)
merged.update(payload)
if row.get("created_ts") and payload.get("created_ts"):
merged["created_ts"] = row.get("created_ts")
rows[index] = merged
updated = True
break
if not updated:
rows.append(payload)
rows = rows[-500:]
save_orders_history_rows(rows)
def format_history_order_line(row: dict, include_status: bool = False) -> str:
source_order_id = str(row.get("source_order_id", "") or "").strip()
api_order_id = str(row.get("api_order_id", "") or "").strip()
buyer = str(row.get("buyer_name", "") or "-")
boosted = int(row.get("boosted", 0) or 0)
requested = int(row.get("requested", 0) or 0)
status = str(row.get("status", "") or "-")
parts = [f"#{source_order_id}" if source_order_id else f"API {api_order_id}", buyer, f"{boosted}/{requested}"]
if include_status:
parts.append(status)
return " | ".join(parts)
def format_recent_order_line(row: dict) -> str:
raise RuntimeError("format_recent_order_line requires order snapshot")
def get_order_snapshot(cardinal: Cardinal, source_order_id: str) -> tuple[float, object | None]:
order_obj = None
created_ts = 0.0
if not source_order_id:
return created_ts, order_obj
try:
order_obj = cardinal.account.get_order(source_order_id)
order_date = getattr(order_obj, "date", None)
if order_date is not None:
try:
created_ts = float(order_date.timestamp())
except Exception:
logger.debug("TRACEBACK", exc_info=True)
except Exception:
logger.debug("TRACEBACK", exc_info=True)
return created_ts, order_obj
def format_orders_section(title: str, rows: list[dict], include_status: bool = False) -> str:
if not rows:
body = "- нет заказов"
else:
body = "\n".join(f"{index}. {format_history_order_line(row, include_status=include_status)}" for index, row in enumerate(rows, 1))
return f"{title}\n{body}"
def format_orders_text(cardinal: Cardinal) -> str:
cutoff = time.time() - 86400
history_rows = get_orders_history_rows()
rows = []
for row in history_rows:
source_order_id = str(row.get("source_order_id", "") or "").strip()
created_ts, _order_obj = get_order_snapshot(cardinal, source_order_id)
if created_ts <= 0:
try:
created_ts = float(row.get("created_ts", 0.0) or 0.0)
except Exception:
created_ts = 0.0
if created_ts <= 0 or created_ts > cutoff:
continue
row = dict(row)
row["resolved_created_ts"] = created_ts
rows.append(row)
completed_rows = []
incomplete_rows = []
for row in rows:
status = str(row.get("status", "") or "").lower()
boosted = max(int(row.get("boosted", 0) or 0), 0)
requested = max(int(row.get("requested", 0) or 0), 0)
source_order_id = str(row.get("source_order_id", "") or "").strip()
if not source_order_id:
continue
try:
_created_ts, order_obj = get_order_snapshot(cardinal, source_order_id)
if order_is_confirmed(order_obj):
continue
if getattr(order_obj, "status", None) in {OrderStatuses.REFUNDED, OrderStatuses.PARTIALLY_REFUNDED}:
continue
except Exception:
logger.debug("TRACEBACK", exc_info=True)
continue
if status == "completed" and requested > 0 and boosted >= requested:
completed_rows.append(row)
else:
incomplete_rows.append(row)
incomplete_rows.sort(key=lambda item: float(item.get("resolved_created_ts", item.get("created_ts", 0.0)) or 0.0))
completed_rows.sort(key=lambda item: float(item.get("resolved_created_ts", item.get("created_ts", 0.0)) or 0.0))
completed_ids = [f"#{str(item.get('source_order_id', '') or '').strip()}" for item in completed_rows if str(item.get('source_order_id', '') or '').strip()]
request_confirm_text = "- нет заказов" if not completed_ids else f"Прошу подтвердить заказы: {', '.join(completed_ids)}"
return (
"📦 Заказы NeverBoost\n\n"
"Ниже показаны неподтверждённые заказы старше 24 часов для обращения в поддержку.\n\n"
f"{format_orders_section('✅ Выполнены, которым больше 24 часов и не подтверждены на FunPay.', completed_rows)}\n\n"
f"Текст для поддержки\n{request_confirm_text}\n\n"
f"{format_orders_section('⚠️ Не выполнены или выполнены частично, которым больше 24 часов и не подтверждены на FunPay.', incomplete_rows, include_status=True)}"
)
def append_stats_row(boosts: int, profit: float) -> None:
rows = get_stats_rows()
now_ts = time.time()
cutoff = now_ts - (400 * 86400)
rows = [row for row in rows if float(row.get("ts", 0.0) or 0.0) >= cutoff]
rows.append({
"ts": now_ts,
"boosts": max(int(boosts or 0), 0),
"profit": round(float(profit or 0.0), 2),
})
save_stats_rows(rows)
logger.info("NeverBoost stats appended: boosts=%s profit=%s rows=%s", boosts, round(float(profit or 0.0), 2), len(rows))
def summarize_stats(rows: list[dict], since_ts: float) -> tuple[int, float]:
boosts = 0
profit = 0.0
for row in rows:
ts = float(row.get("ts", 0.0) or 0.0)
if ts < since_ts:
continue
boosts += max(int(row.get("boosts", 0) or 0), 0)
profit += float(row.get("profit", 0.0) or 0.0)
return boosts, round(profit, 2)
def summarize_yesterday(rows: list[dict]) -> tuple[int, float]:
now_dt = datetime.now()
start_today = datetime(now_dt.year, now_dt.month, now_dt.day)
start_yesterday = start_today - timedelta(days=1)
boosts = 0
profit = 0.0
for row in rows:
ts = float(row.get("ts", 0.0) or 0.0)
dt = datetime.fromtimestamp(ts) if ts > 0 else None
if dt is None or dt < start_yesterday or dt >= start_today:
continue
boosts += max(int(row.get("boosts", 0) or 0), 0)
profit += float(row.get("profit", 0.0) or 0.0)
return boosts, round(profit, 2)
def format_stats_text() -> str:
rows = get_stats_rows()
now_ts = time.time()
hour_boosts, hour_profit = summarize_stats(rows, now_ts - 3600)
day_boosts, day_profit = summarize_stats(rows, now_ts - 86400)
week_boosts, week_profit = summarize_stats(rows, now_ts - (7 * 86400))
month_boosts, month_profit = summarize_stats(rows, now_ts - (30 * 86400))
all_boosts, all_profit = summarize_stats(rows, 0.0)
yesterday_boosts, yesterday_profit = summarize_yesterday(rows)
return (
"📊 Статистика\n\n"
f"За час: бустов {hour_boosts}, прибыль {hour_profit:.2f}\n"
f"За 24 часа: бустов {day_boosts}, прибыль {day_profit:.2f}\n"
f"Вчера: бустов {yesterday_boosts}, прибыль {yesterday_profit:.2f}\n"
f"За неделю: бустов {week_boosts}, прибыль {week_profit:.2f}\n"
f"За месяц: бустов {month_boosts}, прибыль {month_profit:.2f}\n"
f"За всё время: бустов {all_boosts}, прибыль {all_profit:.2f}"
)
def get_pending(chat_id: int | str) -> dict | None:
data = load_json(PENDING_PATH, {})
if not isinstance(data, dict):
return None
item = data.get(str(chat_id))
return item if isinstance(item, dict) else None
def find_pending_for_message(message) -> tuple[str | None, dict | None]:
data = load_json(PENDING_PATH, {})
if not isinstance(data, dict):
return None, None
direct = data.get(str(getattr(message, "chat_id", "")))
if isinstance(direct, dict):
return str(getattr(message, "chat_id", "")), direct
author = str(getattr(message, "author", "") or "").strip().lower()
text = str(getattr(message, "text", "") or "")
order_match = re.search(r"#([A-Z0-9]+)", text)
order_id = order_match.group(1) if order_match else None
for key, value in data.items():
if not isinstance(value, dict):
continue
buyer = str(value.get("buyer_username", "") or "").strip().lower()
pending_order = str(value.get("order_id", "") or "").strip()
if author and buyer and author == buyer:
return key, value
if order_id and pending_order == order_id:
return key, value
return None, None
def get_pending_bind(user_id: int | str) -> dict | None:
data = load_json(PENDING_BIND_PATH, {})
if not isinstance(data, dict):
return None
item = data.get(str(user_id))
return item if isinstance(item, dict) else None
def set_pending_bind(user_id: int | str, payload: dict) -> None:
data = load_json(PENDING_BIND_PATH, {})
if not isinstance(data, dict):
data = {}
data[str(user_id)] = payload
save_json(PENDING_BIND_PATH, data)
def clear_pending_bind(user_id: int | str) -> None:
data = load_json(PENDING_BIND_PATH, {})
if isinstance(data, dict) and str(user_id) in data:
del data[str(user_id)]
save_json(PENDING_BIND_PATH, data)
def get_edit_bind(user_id: int | str) -> dict | None:
data = load_json(EDIT_BIND_PATH, {})
if not isinstance(data, dict):
return None
item = data.get(str(user_id))
return item if isinstance(item, dict) else None
def set_edit_bind(user_id: int | str, payload: dict) -> None:
data = load_json(EDIT_BIND_PATH, {})
if not isinstance(data, dict):
data = {}
data[str(user_id)] = payload
save_json(EDIT_BIND_PATH, data)
def clear_edit_bind(user_id: int | str) -> None:
data = load_json(EDIT_BIND_PATH, {})
if isinstance(data, dict) and str(user_id) in data:
del data[str(user_id)]
save_json(EDIT_BIND_PATH, data)
def get_manual_draft(user_id: int | str) -> dict | None:
data = load_json(MANUAL_DRAFT_PATH, {})
if not isinstance(data, dict):
return None
item = data.get(str(user_id))
return item if isinstance(item, dict) else None
def set_manual_draft(user_id: int | str, payload: dict) -> None:
data = load_json(MANUAL_DRAFT_PATH, {})
if not isinstance(data, dict):
data = {}
data[str(user_id)] = payload
save_json(MANUAL_DRAFT_PATH, data)
def clear_manual_draft(user_id: int | str) -> None:
data = load_json(MANUAL_DRAFT_PATH, {})
if isinstance(data, dict) and str(user_id) in data:
del data[str(user_id)]
save_json(MANUAL_DRAFT_PATH, data)
def set_pending(chat_id: int | str, payload: dict) -> None:
data = load_json(PENDING_PATH, {})
if not isinstance(data, dict):
data = {}
data[str(chat_id)] = payload
save_json(PENDING_PATH, data)
def clear_pending(chat_id: int | str) -> None:
data = load_json(PENDING_PATH, {})
if isinstance(data, dict) and str(chat_id) in data:
del data[str(chat_id)]
save_json(PENDING_PATH, data)
def clear_pending_by_key(key: str) -> None:
data = load_json(PENDING_PATH, {})
if isinstance(data, dict) and key in data:
del data[key]
save_json(PENDING_PATH, data)
def pop_pending_by_order_id(order_id: str) -> dict | None:
data = load_json(PENDING_PATH, {})
if not isinstance(data, dict):
return None
for key, value in list(data.items()):
if isinstance(value, dict) and str(value.get("order_id", "") or "") == str(order_id or ""):
del data[key]
save_json(PENDING_PATH, data)
return value
return None
def has_pending_order_id(order_id: str) -> bool:
data = load_json(PENDING_PATH, {})
if not isinstance(data, dict):
return False
target = str(order_id or "").strip()
for value in data.values():
if isinstance(value, dict) and str(value.get("order_id", "") or "").strip() == target:
return True
return False
def remember_order_id(order_id: str) -> None:
target = str(order_id or "").strip()
if not target:
return
with RECENT_ORDER_IDS_LOCK:
RECENT_ORDER_IDS[target] = time.time()
def has_recent_order_id(order_id: str, ttl_sec: int = 120) -> bool:
target = str(order_id or "").strip()
if not target:
return False
now_ts = time.time()
with RECENT_ORDER_IDS_LOCK:
expired = [key for key, ts in RECENT_ORDER_IDS.items() if now_ts - ts > ttl_sec]
for key in expired:
RECENT_ORDER_IDS.pop(key, None)
ts = RECENT_ORDER_IDS.get(target)
return ts is not None and (now_ts - ts) <= ttl_sec
def get_processing(order_id: str) -> dict | None:
data = load_json(PROCESSING_PATH, {})
if not isinstance(data, dict):
return None
item = data.get(str(order_id or ""))
return item if isinstance(item, dict) else None
def set_processing(order_id: str, payload: dict) -> None:
data = load_json(PROCESSING_PATH, {})
if not isinstance(data, dict):
data = {}
data[str(order_id or "")] = payload
save_json(PROCESSING_PATH, data)
def clear_processing(order_id: str) -> None:
data = load_json(PROCESSING_PATH, {})
if isinstance(data, dict) and str(order_id or "") in data:
del data[str(order_id or "")]
save_json(PROCESSING_PATH, data)
def get_api_key() -> str:
if os.path.exists(API_KEY_PATH):
try:
with open(API_KEY_PATH, "r", encoding="utf-8") as f:
return f.read().strip()
except Exception:
logger.debug("TRACEBACK", exc_info=True)
if os.path.exists(BOOSTS_CONFIG_FALLBACK):
try:
data = load_json(BOOSTS_CONFIG_FALLBACK, {})
if isinstance(data, dict):
return str(data.get("api_key", "") or "").strip()
except Exception:
logger.debug("TRACEBACK", exc_info=True)
return ""
def save_api_key(api_key: str) -> None:
os.makedirs(os.path.dirname(API_KEY_PATH), exist_ok=True)
with open(API_KEY_PATH, "w", encoding="utf-8") as f:
f.write(api_key.strip())
def build_menu() -> K:
return K().row(
B("🧷 Мои лоты", callback_data="neverboost:bindings"),
B("⚙️ API и наличие", callback_data="neverboost:settings")
).row(
B("📦 Заказы", callback_data="neverboost:orders"),
B("📊 Статистика", callback_data="neverboost:stats")
).row(
B("🛰 Ручная выдача", callback_data="neverboost:manual")
)
def build_remove_menu() -> K:
keyboard = K()
for index, item in enumerate(get_bindings()):
duration = "1 месяц" if item["duration"] == "oneMonth" else "3 месяца"
bpu = max(int(item.get("boosts_per_unit", 1) or 1), 1)
keyboard.row(B(f"{item['lot'][:38]} [{duration}, x{bpu}]", callback_data=f"neverboost:remove:{index}"))
keyboard.row(B("Назад", callback_data="neverboost:menu"))
return keyboard
def build_bindings_menu() -> K:
keyboard = K().row(B("➕ Добавить лот", callback_data="neverboost:add"))
for index, item in enumerate(get_bindings()):
duration = "1м" if item["duration"] == "oneMonth" else "3м"
bpu = max(int(item.get("boosts_per_unit", 1) or 1), 1)
keyboard.row(B(f"#{index + 1} [{duration} | x{bpu}] {item['lot'][:24]}", callback_data=f"neverboost:binding:{index}"))
keyboard.row(B("⬅️ Назад", callback_data="neverboost:menu"))
return keyboard
def build_binding_actions_menu(index: int, item: dict) -> K:
current_duration = str(item.get("duration", "oneMonth") or "oneMonth")
toggle_duration = "threeMonth" if current_duration == "oneMonth" else "oneMonth"
toggle_label = "🗓️ Сделать 3 месяца" if current_duration == "oneMonth" else "🗓️ Сделать 1 месяц"
return K().row(
B(toggle_label, callback_data=f"neverboost:binding_duration:{index}:{toggle_duration}"),
B("✏️ Изменить x/шт", callback_data=f"neverboost:binding_bpu:{index}")
).row(
B("🗑️ Удалить", callback_data=f"neverboost:remove:{index}"),
B("⬅️ Назад", callback_data="neverboost:bindings")
)
def build_settings_menu() -> K:
auto_sync_enabled = bool(get_settings().get("auto_sync_enabled", True))
welcome_enabled = bool(get_settings().get("welcome_enabled", True))
stock_enabled = bool(get_settings().get("stock_command_enabled", True))
toggle_label = "🟢 Автообновление наличия: ВКЛ" if auto_sync_enabled else "🔴 Автообновление наличия: ВЫКЛ"
welcome_label = "🟢 Приветствие: ВКЛ" if welcome_enabled else "🔴 Приветствие: ВЫКЛ"
stock_label = "🟢 Команда #stock: ВКЛ" if stock_enabled else "🔴 Команда #stock: ВЫКЛ"
return K().row(
B("🔑 Ввести API ключ", callback_data="neverboost:api_key"),
B("📡 Проверить API ключ", callback_data="neverboost:api_check")
).row(
B(toggle_label, callback_data="neverboost:toggle_auto_sync")
).row(
B(welcome_label, callback_data="neverboost:toggle_welcome")
).row(
B(stock_label, callback_data="neverboost:toggle_stock")
).row(
B("💸 Авто-цена 1 месяц", callback_data="neverboost:price_rule:oneMonth"),
B("💸 Авто-цена 3 месяца", callback_data="neverboost:price_rule:threeMonth")
).row(
B("📦 Обновить наличие сейчас", callback_data="neverboost:sync_stock")
).row(
B("⬅️ Назад", callback_data="neverboost:menu")
)
def build_duration_menu() -> K:
return K().row(
B("1 месяц", callback_data="neverboost:duration:oneMonth"),
B("3 месяца", callback_data="neverboost:duration:threeMonth")
).row(
B("❌ Отмена", callback_data="neverboost:cancel_bind")
)
def build_manual_duration_menu() -> K:
return K().row(
B("1 месяц", callback_data="neverboost:manual_duration:oneMonth"),
B("3 месяца", callback_data="neverboost:manual_duration:threeMonth")
).row(
B("❌ Отмена", callback_data="neverboost:cancel_bind")
)
def build_manual_confirm_menu() -> K:
return K().row(
B("✅ Выдать", callback_data="neverboost:manual_confirm"),
B("❌ Отмена", callback_data="neverboost:cancel_bind")
)
def build_cancel_menu() -> K:
return K().row(B("❌ Отмена", callback_data="neverboost:cancel_bind"))
def format_menu_text() -> str:
bindings = get_bindings()
api_key = get_api_key()
api_key_status = "установлен" if api_key else "не установлен"
auto_sync = "включен" if get_settings().get("auto_sync_enabled", True) else "выключен"
if bindings:
linked = "\n".join(
f"{idx + 1}. [{'1м' if item['duration'] == 'oneMonth' else '3м'} | x{max(int(item.get('boosts_per_unit', 1) or 1), 1)}] {item['lot'][:42]}"
for idx, item in enumerate(bindings[:8])
)
if len(bindings) > 8:
linked += f"\n... и ещё {len(bindings) - 8}"
else:
linked = "- нет привязанных лотов"
return (
"NeverBoost\n\n"
"Разделы:\n"
"🧷 Мои лоты - здесь ты добавляешь и меняешь лоты\n"
"⚙️ API и наличие - здесь API ключ, автообновление наличия и авто-цена\n\n"
"📦 Заказы - старые выполненные и проблемные заказы по бустам\n\n"
"📊 Статистика - сколько бустов выдано и какая прибыль по периодам\n\n"
"🛰 Ручная выдача - отправить бусты вручную без заказа FunPay\n\n"
f"API ключ: {api_key_status}\n"
f"Автообновление наличия: {auto_sync}\n"
f"Привязок: {len(bindings)}\n\n"
"После покупки бот попросит ссылку на Discord сервер и отправит заказ в API.\n\n"
f"Список лотов:\n{linked}"
)
def format_bindings_text() -> str:
bindings = get_bindings()
if not bindings:
body = "- нет привязанных лотов"
else:
body = "\n".join(
f"{idx + 1}. [{'1 месяц' if item['duration'] == 'oneMonth' else '3 месяца'} | x{max(int(item.get('boosts_per_unit', 1) or 1), 1)}] {item['lot']}"
for idx, item in enumerate(bindings)
)
return (
"🧷 Мои лоты\n\n"
"Здесь находятся все лоты, которые работают через NeverBoost.\n"
"Нажми на нужный лот, чтобы изменить его.\n\n"
f"{body}"
)
def format_settings_text() -> str:
api_key_status = "установлен" if get_api_key() else "не установлен"
settings = get_settings()
auto_sync = "включен" if settings.get("auto_sync_enabled", True) else "выключен"
welcome = "включено" if settings.get("welcome_enabled", True) else "выключено"
stock = "включена" if settings.get("stock_command_enabled", True) else "выключена"
interval = max(int(settings.get("auto_sync_interval_sec", 60) or 60), 10)
one_rule = format_price_rule(settings.get("price_rule_one_month"))
three_rule = format_price_rule(settings.get("price_rule_three_month"))
return (
"⚙️ API и наличие\n\n"
"Здесь настраивается API ключ, автоматическое обновление наличия и авто-цена в лотах.\n\n"
f"API ключ: {api_key_status}\n"
f"Автообновление наличия: {auto_sync}\n"
f"Приветствие: {welcome}\n"
f"Команда #stock: {stock}\n"
f"Авто-цена 1 месяц: {one_rule}\n"
f"Авто-цена 3 месяца: {three_rule}\n"
f"Интервал: {interval} сек"
)
def format_price_rule(rule: dict | None) -> str:
if not isinstance(rule, dict):
return "выключена"
rule_type = str(rule.get("type", "off") or "off")
try:
value = float(rule.get("value", 0.0) or 0.0)
except Exception:
value = 0.0
if rule_type == "percent":
return f"{value:.2f}%"
if rule_type == "fixed":
return f"{value:.2f} RUB"
return "выключена"
def parse_price_rule_input(raw: str) -> dict:
text = str(raw or "").strip().replace(",", ".")
if not text:
raise ValueError("empty")
if text.lower() in {"off", "disable", "выкл", "выключить", "нет", "0%", "0"}:
return {"type": "off", "value": 0.0}
if text.endswith("%"):
value = float(text[:-1].strip())
return {"type": "percent", "value": round(value, 2)}
value = float(text)
return {"type": "fixed", "value": round(value, 2)}
def calc_target_lot_price(api_price_per_boost: float, boosts_per_unit: int, rule: dict | None) -> float | None:
base_price = float(api_price_per_boost or 0.0) * max(int(boosts_per_unit or 1), 1)
if base_price <= 0:
return None
rule_type = str((rule or {}).get("type", "off") or "off")
try:
rule_value = float((rule or {}).get("value", 0.0) or 0.0)
except Exception:
rule_value = 0.0
if rule_type == "percent":
target = base_price * (1 + (rule_value / 100.0))
elif rule_type == "fixed":
target = base_price + rule_value
else:
target = base_price
return round(max(target, 0.01), 2)
def notify_authorized_users(cardinal: Cardinal, text: str) -> None:
if cardinal.telegram is None:
return
for user_id in cardinal.telegram.authorized_users.keys():
try:
cardinal.telegram.bot.send_message(user_id, text)
except Exception:
logger.debug("TRACEBACK", exc_info=True)
def normalize_locale(raw: str) -> str | None:
value = (raw or "").strip().lower()
if not value:
return None
if "англ" in value or "english" in value or value.startswith("en"):
return "en"
if "рус" in value or "russian" in value or value.startswith("ru"):
return "ru"
return None
def buyer_text(locale: str, key: str, **kwargs) -> str:
lang = "en" if locale == "en" else "ru"
if key == "status_partial":
if lang == "ru":
return f"⚠️ Заказ выполнен частично: выдано {kwargs.get('boosted', '{boosted}')}/{kwargs.get('requested', '{requested}')} бустов."
return f"⚠️ Order completed partially: delivered {kwargs.get('boosted', '{boosted}')}/{kwargs.get('requested', '{requested}')} boosts."
text = BUYER_TEXTS.get(lang, BUYER_TEXTS["ru"]).get(key, "")
return text.format(**kwargs) if kwargs else text
def delivery_text(locale: str, qty: int) -> str:
lang = "en" if locale == "en" else "ru"
text = DELIVERY_TEXTS.get(lang, DELIVERY_TEXTS["ru"])
return text.format(qty=max(int(qty or 0), 1))
def make_source_order_payload(order_id: str, qty: int, chat_id: int | str | None = None) -> dict:
payload = {"order_id": str(order_id or "").strip(), "qty": max(int(qty or 0), 1)}
if chat_id is not None:
payload["chat_id"] = chat_id
return payload
def get_pending_source_orders(pending: dict | None) -> list[dict]:
if not isinstance(pending, dict):
return []
raw = pending.get("source_orders")
result = []
if isinstance(raw, list):
for item in raw:
if not isinstance(item, dict):
continue
order_id = str(item.get("order_id", "") or "").strip()
if not order_id:
continue
result.append(make_source_order_payload(order_id, int(item.get("qty", 1) or 1), item.get("chat_id")))
if result:
return result
order_id = str(pending.get("order_id", "") or "").strip()
if not order_id:
return []
return [make_source_order_payload(order_id, int(pending.get("qty", 1) or 1), pending.get("chat_id"))]
def sum_order_qty(items: list[dict]) -> int:
total = 0
for item in items:
if isinstance(item, dict):
total += max(int(item.get("qty", 1) or 1), 1)
return max(total, 0)
def get_queue_orders(pending: dict | None) -> list[dict]:
if not isinstance(pending, dict):
return []
raw = pending.get("queue_orders")
result = []
if isinstance(raw, list):
for item in raw:
if not isinstance(item, dict):
continue
order_id = str(item.get("order_id", "") or "").strip()
if not order_id:
continue
result.append(make_source_order_payload(order_id, int(item.get("qty", 1) or 1), item.get("chat_id")))
if result:
return result
queued = pending.get("queued_order")
if isinstance(queued, dict):
order_id = str(queued.get("order_id", "") or "").strip()
if order_id:
return [make_source_order_payload(order_id, int(queued.get("qty", 1) or 1), queued.get("chat_id"))]
return []
def queue_invite_choice_text(locale: str, current_qty: int, total_qty: int) -> str:
return buyer_text(locale, "pending_invite_choice", current_qty=current_qty, total_qty=total_qty)
def unique_order_ids(order_ids: list[str]) -> list[str]:
result = []
seen = set()
for order_id in order_ids:
value = str(order_id or "").strip().upper()
if not value or value in seen:
continue
seen.add(value)
result.append(value)
return result
def format_order_links(order_ids: list[str]) -> str:
rows = []
for index, order_id in enumerate(unique_order_ids(order_ids), 1):
rows.append(f"{index}. #{order_id} - https://funpay.com/orders/{order_id}/")
return "\n".join(rows)
def extract_strong_order_language_candidates(order) -> list[str]:
result = []
try:
lot_params = getattr(order, "lot_params", None)
if isinstance(lot_params, list):
for item in lot_params:
if isinstance(item, tuple) and len(item) >= 2:
name = str(item[0] or "").lower()
value = str(item[1] or "")
if "язык" in name or "language" in name or "lang" in name:
result.append(value)
except Exception:
logger.debug("TRACEBACK", exc_info=True)
try:
lot_params_dict = getattr(order, "lot_params_dict", None)
if isinstance(lot_params_dict, dict):
for name, value in lot_params_dict.items():
name = str(name or "").lower()
if "язык" in name or "language" in name or "lang" in name:
result.append(str(value or ""))
except Exception:
logger.debug("TRACEBACK", exc_info=True)
return result
def extract_chat_language_from_html(chat_html: str) -> str | None:
if not chat_html:
return None
pattern = re.compile(
r']*class="[^"]*param-item[^"]*"[^>]*>\s*'
r']*>(.*?)\s*'
r']*>(.*?)',
re.IGNORECASE | re.DOTALL,
)
for title_raw, value_raw in pattern.findall(chat_html):
title = re.sub(r"<[^>]+>", "", title_raw or "").strip().lower()
value = re.sub(r"<[^>]+>", "", value_raw or "").strip()
if value and ("язык собеседника" in title or "interlocutor language" in title or title == "language"):
return value
return None
def extract_chat_language(cardinal: Cardinal, chat_id: int | str | None, buyer_username: str | None = None) -> str | None:
if chat_id is not None:
try:
chat_obj = cardinal.account.get_chat(chat_id, with_history=False)
raw = extract_chat_language_from_html(getattr(chat_obj, "html", "") or "")
normalized = normalize_locale(raw or "")
if normalized:
return normalized
except Exception:
logger.debug("TRACEBACK", exc_info=True)
if buyer_username:
try:
shortcut = cardinal.account.get_chat_by_name(buyer_username)
if shortcut:
chat_obj = cardinal.account.get_chat(shortcut.id, with_history=False)
raw = extract_chat_language_from_html(getattr(chat_obj, "html", "") or "")
normalized = normalize_locale(raw or "")
if normalized:
return normalized
except Exception:
logger.debug("TRACEBACK", exc_info=True)
return None
def detect_order_locale(cardinal: Cardinal, event_order) -> str:
order = event_order if hasattr(event_order, "buyer_username") and hasattr(event_order, "chat_id") else cardinal.get_order_from_object(event_order)
chat_id = getattr(order, "chat_id", None) if order else None
if chat_id is None:
chat_id = getattr(event_order, "chat_id", None)
buyer_username = getattr(order, "buyer_username", None) if order else None
if not buyer_username:
buyer_username = getattr(event_order, "buyer_username", None)
chat_locale = extract_chat_language(cardinal, chat_id, buyer_username)
if chat_locale:
return chat_locale
for candidate in extract_strong_order_language_candidates(order):
normalized = normalize_locale(candidate)
if normalized:
return normalized
raw_joined = f"{order}\n{event_order}".lower()
if "язык собеседника" in raw_joined:
if "англий" in raw_joined or "english" in raw_joined:
return "en"
if "русск" in raw_joined or "russian" in raw_joined:
return "ru"
return "ru"
def detect_duration(order_obj) -> str:
parts = [str(getattr(order_obj, "description", "") or "")]
try:
lot_params = getattr(order_obj, "lot_params", None)
if isinstance(lot_params, list):
for item in lot_params:
if isinstance(item, tuple) and len(item) >= 2:
parts.append(str(item[1] or ""))
except Exception:
logger.debug("TRACEBACK", exc_info=True)
try:
lot_params_dict = getattr(order_obj, "lot_params_dict", None)
if isinstance(lot_params_dict, dict):
for _, value in lot_params_dict.items():
parts.append(str(value or ""))
except Exception:
logger.debug("TRACEBACK", exc_info=True)
joined = " ".join(parts).lower()
if "3 месяц" in joined or "3 мес" in joined or "three" in joined or "3m" in joined:
return "threeMonth"
return "oneMonth"
def detect_duration_from_lot_name(lot_name: str) -> str:
joined = str(lot_name or "").lower()
if "3 месяц" in joined or "3 мес" in joined or "three" in joined or "3m" in joined:
return "threeMonth"
return "oneMonth"
def extract_viewing_lot_from_chat_html(chat_html: str) -> tuple[str, str] | None:
if not chat_html:
return None
match = re.search(
r']*class="[^"]*param-item\s+chat-panel[^"]*"[^>]*>.*?]*href="([^"]+)"[^>]*>(.*?)',
chat_html,
flags=re.IGNORECASE | re.DOTALL,
)
if not match:
return None
link = re.sub(r"&", "&", match.group(1) or "").strip()
text = re.sub(r"<[^>]+>", "", match.group(2) or "")
text = re.sub(r"\s+", " ", text).strip()
if not text and not link:
return None
return text, link
def parse_lot_id_from_link(link: str) -> str:
href = str(link or "").strip()
if not href:
return ""
offer_match = re.search(r"[?&]offer=(\d+)", href, flags=re.IGNORECASE)
if offer_match:
return offer_match.group(1)
fallback = LOT_ID_RE.search(href)
return fallback.group(1) if fallback else ""
def resolve_viewing_lot(cardinal: Cardinal, chat_id: int | str, buyer_viewing=None) -> tuple[str, str] | None:
lot_text = ""
lot_link = ""
if buyer_viewing is not None:
parsed = extract_viewing_lot_from_chat_html(getattr(buyer_viewing, "html", "") or "")
if parsed is not None:
lot_text, lot_link = parsed
if not lot_text:
lot_text = str(getattr(buyer_viewing, "text", "") or "").strip()
if not lot_link:
lot_link = str(getattr(buyer_viewing, "link", "") or "").strip()
if not lot_text or not lot_link:
try:
chat_obj = cardinal.account.get_chat(int(chat_id), with_history=False)
except Exception:
logger.debug("TRACEBACK", exc_info=True)
return None
parsed = extract_viewing_lot_from_chat_html(getattr(chat_obj, "html", "") or "")
if parsed is not None:
if not lot_text:
lot_text = parsed[0]
if not lot_link:
lot_link = parsed[1]
if not lot_text:
lot_text = str(getattr(chat_obj, "looking_text", "") or "").strip()
if not lot_link:
lot_link = str(getattr(chat_obj, "looking_link", "") or "").strip()
if not lot_text and not lot_link:
return None
return lot_text, lot_link
def find_binding_for_viewing_lot(cardinal: Cardinal, chat_id: int | str, buyer_viewing=None) -> dict | None:
resolved = resolve_viewing_lot(cardinal, chat_id, buyer_viewing=buyer_viewing)
if resolved is None:
return None
lot_text, lot_link = resolved
duration = detect_duration_from_lot_name(lot_text)
link_lot_id = parse_lot_id_from_link(lot_link)
bindings = [i for i in get_bindings() if i.get("duration") == duration]
for item in bindings:
lot_name = str(item.get("lot", "") or "").strip()
lot_id = str(item.get("lot_id", "") or "").strip()
if lot_id and link_lot_id and lot_id == link_lot_id:
return item
if lot_name and lot_name == lot_text:
return item
logger.info(
"NeverBoost viewing lot not matched: chat=%s duration=%s lot=%s link=%s",
chat_id,
duration,
lot_text,
lot_link,
)
return None
def maybe_send_welcome_for_viewing_lot(cardinal: Cardinal, message, settings: dict, buyer_viewing=None) -> bool:
if not settings.get("welcome_enabled", True):
return False
if has_recent_order_chat_for_message(message) or not should_send_welcome(message.chat_id):
return False
viewing_binding = find_binding_for_viewing_lot(cardinal, message.chat_id, buyer_viewing=buyer_viewing)
if viewing_binding is None:
return False
ok, one_month, three_month = available_boosts_summary()
locale = extract_chat_language(cardinal, message.chat_id, getattr(message, "author", None) or getattr(message, "chat_name", None)) or "ru"
welcome_template = WELCOME_TEXTS.get(locale, WELCOME_TEXTS["ru"])
welcome_text = welcome_template.format(one_month=one_month if ok else "?", three_month=three_month if ok else "?")
cardinal.send_message(message.chat_id, welcome_text, getattr(message, "author", None) or getattr(message, "chat_name", None), watermark=False)
set_chat_activity(message.chat_id, {"ts": time.time(), "reason": "welcome", "lot": viewing_binding.get("lot")})
logger.info(
"NeverBoost welcome sent for viewing lot: chat=%s author=%s lot=%s",
getattr(message, "chat_id", None),
getattr(message, "author", None),
viewing_binding.get("lot"),
)
return True
def parse_balance_value(data: dict) -> float | None:
try:
return float(data.get("balance"))
except Exception:
return None
def safe_get_prices(client: NeverBoostApiClient) -> tuple[bool, dict[str, float]]:
ok, data = client._request("GET", "/prices")
if not ok or not isinstance(data, dict):
return False, {"oneMonth": 0.0, "threeMonth": 0.0}
prices = data.get("prices", {})
if not isinstance(prices, dict):
return False, {"oneMonth": 0.0, "threeMonth": 0.0}
return True, {
"oneMonth": float(prices.get("oneMonth", 0.0) or 0.0),
"threeMonth": float(prices.get("threeMonth", 0.0) or 0.0),
}
def safe_get_stock(client: NeverBoostApiClient) -> tuple[bool, dict[str, int]]:
ok, data = client.stock()
if not ok or not isinstance(data, dict):
return False, {"oneMonth": 0, "threeMonth": 0}
stock = data.get("stock", {})
if not isinstance(stock, dict):
return False, {"oneMonth": 0, "threeMonth": 0}
return True, {
"oneMonth": max(int(stock.get("oneMonth", 0) or 0), 0),
"threeMonth": max(int(stock.get("threeMonth", 0) or 0), 0),
}
def max_boosts_by_balance(balance: float, price_per_boost: float) -> int:
if balance <= 0 or price_per_boost <= 0:
return 0
return max(int(balance // price_per_boost), 0)
def fetch_api_state(client: "NeverBoostApiClient") -> tuple[float | None, dict[str, float], dict[str, int]]:
ok_balance, balance_data = client.balance()
balance = parse_balance_value(balance_data) if ok_balance else None
ok_prices, prices = safe_get_prices(client)
ok_stock, stock = safe_get_stock(client)
if balance is None or not ok_prices or not ok_stock:
return None, {}, {}
return balance, prices, stock
def available_boosts_summary() -> tuple[bool, int, int]:
api_key = get_api_key()
if not api_key:
return False, 0, 0
client = NeverBoostApiClient(api_key)
balance, prices, stock = fetch_api_state(client)
if balance is None:
return False, 0, 0
one_month = min(max_boosts_by_balance(balance, float(prices.get("oneMonth", 0.0) or 0.0)), max(int(stock.get("oneMonth", 0) or 0), 0))
three_month = min(max_boosts_by_balance(balance, float(prices.get("threeMonth", 0.0) or 0.0)), max(int(stock.get("threeMonth", 0) or 0), 0))
return True, one_month, three_month
def stock_text(locale: str = "ru") -> str:
lang = "en" if locale == "en" else "ru"
ok, one_month, three_month = available_boosts_summary()
if not ok:
if lang == "en":
return "⚠️ Stock is currently unavailable. Check the API key first."
return "⚠️ Наличие сейчас недоступно. Сначала проверь API ключ."
if lang == "en":
return (
"📦 Boost stock:\n"
f"🗓 1 month - {one_month}\n"
f"🗓 3 months - {three_month}"
)
return (
"📦 Наличие бустов:\n"
f"🗓 1 месяц - {one_month} шт.\n"
f"🗓 3 месяца - {three_month} шт."
)
def manual_invite_preview_text(invite_data: dict, qty: int) -> str:
guild = invite_data.get("guild", {}) if isinstance(invite_data, dict) else {}
guild_name = str(guild.get("name") or "Unknown")
guild_id = str(guild.get("id") or "?")
members = invite_data.get("approximate_member_count", "?") if isinstance(invite_data, dict) else "?"
online = invite_data.get("approximate_presence_count", "?") if isinstance(invite_data, dict) else "?"
return (
"🛰 Проверка сервера перед выдачей\n\n"
f"Название: {guild_name}\n"
f"ID сервера: {guild_id}\n"
f"Участников: {members}\n"
f"Онлайн: {online}\n"
f"Бустов к выдаче: {qty}"
)
def set_lot_enabled_amount_and_price(cardinal: Cardinal, lot_id: str, units_available: int, target_price: float | None = None) -> bool:
try:
lf = cardinal.account.get_lot_fields(int(lot_id))
except Exception:
logger.info("NeverBoost stock sync failed to load lot fields: lot_id=%s", lot_id)
logger.debug("TRACEBACK", exc_info=True)
return False
target_units = max(int(units_available), 0)
target_active = target_units > 0
changed = False
logger.info(
"NeverBoost stock sync current lot state: lot_id=%s current_amount=%s current_active=%s current_price=%s target_amount=%s target_active=%s target_price=%s",
lot_id,
getattr(lf, "amount", None),
getattr(lf, "active", None),
getattr(lf, "price", None),
target_units,
target_active,
target_price,
)
try:
if lf.amount != target_units:
lf.amount = target_units
changed = True
except Exception:
logger.debug("TRACEBACK", exc_info=True)
try:
if bool(lf.active) != target_active:
lf.active = target_active
changed = True
except Exception:
logger.debug("TRACEBACK", exc_info=True)
try:
if target_price is not None and round(float(getattr(lf, "price", 0.0) or 0.0), 2) != round(float(target_price), 2):
lf.price = round(float(target_price), 2)
changed = True
except Exception:
logger.debug("TRACEBACK", exc_info=True)
if not changed:
return False
try:
cardinal.account.save_lot(lf)
logger.info("NeverBoost stock sync saved lot: lot_id=%s", lot_id)
return True
except Exception:
logger.info("NeverBoost stock sync failed to save lot: lot_id=%s", lot_id)
logger.debug("TRACEBACK", exc_info=True)
return False
def sync_lots_availability(cardinal: Cardinal) -> tuple[int, str]:
# NeverBoost: считаем реальное наличие по API stock и доступному балансу.
api_key = get_api_key()
if not api_key:
return 0, "API ключ не установлен."
client = NeverBoostApiClient(api_key)
ok_balance, balance_data = client.balance()
balance = parse_balance_value(balance_data) if ok_balance else None
ok_prices, prices = safe_get_prices(client)
ok_stock, stock = safe_get_stock(client)
if balance is None or not ok_prices or not ok_stock:
return 0, "Не удалось получить баланс / цены / сток API."
max_by_balance = {
"oneMonth": max_boosts_by_balance(balance, float(prices.get("oneMonth", 0.0) or 0.0)),
"threeMonth": max_boosts_by_balance(balance, float(prices.get("threeMonth", 0.0) or 0.0)),
}
max_boosts = {
"oneMonth": min(max_by_balance["oneMonth"], max(int(stock.get("oneMonth", 0) or 0), 0)),
"threeMonth": min(max_by_balance["threeMonth"], max(int(stock.get("threeMonth", 0) or 0), 0)),
}
settings = get_settings()
price_rules = {
"oneMonth": settings.get("price_rule_one_month"),
"threeMonth": settings.get("price_rule_three_month"),
}
changed = 0
for item in get_bindings():
lot_id = str(item.get("lot_id", "") or "").strip()
if not lot_id:
continue
duration = str(item.get("duration") or "oneMonth")
boosts_per_unit = max(int(item.get("boosts_per_unit", 1) or 1), 1)
units = max_boosts.get(duration, 0) // boosts_per_unit
target_price = calc_target_lot_price(float(prices.get(duration, 0.0) or 0.0), boosts_per_unit, price_rules.get(duration))
logger.info(
"NeverBoost stock sync lot: lot_id=%s lot=%s duration=%s boosts_per_unit=%s target_units=%s target_price=%s rule=%s",
lot_id,
item.get("lot"),
duration,
boosts_per_unit,
units,
target_price,
price_rules.get(duration),
)
if set_lot_enabled_amount_and_price(cardinal, lot_id, units, target_price=target_price):
changed += 1
return changed, (
f"Синхронизация завершена.\n"
f"Баланс: {balance:.2f}\n"
f"Сток 1м: {max(int(stock.get('oneMonth', 0) or 0), 0)}\n"
f"Сток 3м: {max(int(stock.get('threeMonth', 0) or 0), 0)}\n"
f"Обновлено лотов: {changed}"
)
def sync_lots_availability_safe(cardinal: Cardinal, silent: bool = False) -> None:
if not AUTO_SYNC_LOCK.acquire(False):
return
try:
changed, text = sync_lots_availability(cardinal)
logger.info("NeverBoost stock sync completed: changed=%s", changed)
if not silent and changed:
notify_authorized_users(cardinal, f"🔄 NeverBoost авто-наличие\n\n{text}")
except Exception as exc:
logger.info("NeverBoost stock sync failed: %s", exc)
logger.debug("TRACEBACK", exc_info=True)
finally:
AUTO_SYNC_LOCK.release()
def parse_version_tuple(value: str) -> tuple[int, ...]:
parts = [int(part) for part in re.findall(r"\d+", str(value or ""))]
return tuple(parts) if parts else (0,)
def extract_version_from_source(source_text: str) -> str:
match = re.search(r'^VERSION\s*=\s*["\']([^"\']+)["\']', source_text, flags=re.MULTILINE)
return str(match.group(1)).strip() if match else ""
def plugin_outdated_info(ttl_sec: int = 300) -> tuple[bool, bool, str]:
global PLUGIN_VERSION_LAST_DATA, PLUGIN_VERSION_LAST_CHECK_TS, PLUGIN_VERSION_LAST_OUTDATED
with PLUGIN_VERSION_CHECK_LOCK:
now_ts = time.time()
if PLUGIN_VERSION_LAST_DATA is not None and now_ts - PLUGIN_VERSION_LAST_CHECK_TS < ttl_sec:
latest_cached = str(PLUGIN_VERSION_LAST_DATA.get("latest_version") or VERSION).strip() or VERSION
required_update = bool(PLUGIN_VERSION_LAST_DATA.get("required_update", False))
return PLUGIN_VERSION_LAST_OUTDATED, required_update, latest_cached
latest_version = VERSION
outdated = False
required_update = False
try:
response = requests.get(
f"{API_BASE_URL}{PLUGIN_VERSION_ENDPOINT}",
params={"current": VERSION},
headers={"Accept": "application/json"},
timeout=15,
)
response.raise_for_status()
data = response.json() if response.content else {}
if isinstance(data, dict):
latest_version = str(data.get("latest_version") or VERSION).strip() or VERSION
raw_outdated = data.get("outdated")
outdated = raw_outdated if isinstance(raw_outdated, bool) else parse_version_tuple(VERSION) < parse_version_tuple(latest_version)
required_update = bool(data.get("required_update", False))
PLUGIN_VERSION_LAST_DATA = data
PLUGIN_VERSION_LAST_CHECK_TS = now_ts
PLUGIN_VERSION_LAST_OUTDATED = outdated
return outdated, required_update, latest_version
except Exception:
logger.debug("TRACEBACK", exc_info=True)
return False, False, latest_version
def plugin_download_url() -> str:
cached = PLUGIN_VERSION_LAST_DATA if isinstance(PLUGIN_VERSION_LAST_DATA, dict) else {}
raw = str(cached.get("download_url") or "").strip()
if raw.startswith(("http://", "https://")):
return raw
if raw.startswith("/"):
return f"{API_BASE_URL}{raw}"
return f"{API_BASE_URL}{PLUGIN_DOWNLOAD_ENDPOINT}"
def apply_plugin_auto_update(latest_version: str) -> tuple[bool, str]:
global PLUGIN_AUTO_UPDATE_READY_VERSION
with PLUGIN_AUTO_UPDATE_LOCK:
if PLUGIN_AUTO_UPDATE_READY_VERSION == latest_version:
return True, ""
current_path = os.path.abspath(__file__)
temp_path = f"{current_path}.new"
try:
response = requests.get(plugin_download_url(), headers={"Accept": "text/x-python, text/plain, */*"}, timeout=20)
response.raise_for_status()
source_text = response.content.decode("utf-8-sig")
if not source_text.strip():
raise ValueError("empty plugin source")
compile(source_text, current_path, "exec")
downloaded_version = extract_version_from_source(source_text)
if downloaded_version and parse_version_tuple(downloaded_version) < parse_version_tuple(latest_version):
raise ValueError(f"downloaded plugin version {downloaded_version} is older than expected {latest_version}")
with open(temp_path, "w", encoding="utf-8", newline="\n") as f:
f.write(source_text)
os.replace(temp_path, current_path)
PLUGIN_AUTO_UPDATE_READY_VERSION = latest_version
logger.info("NeverBoost plugin auto-update downloaded version %s.", latest_version)
return True, ""
except Exception as exc:
try:
if os.path.exists(temp_path):
os.remove(temp_path)
except Exception:
logger.debug("TRACEBACK", exc_info=True)
return False, str(exc)
def plugin_outdated_warning_text(latest_version: str, required_update: bool) -> str:
if required_update:
return (
"⚠️ Плагин NeverBoost устарел и требует обязательного обновления\n\n"
f"Текущая версия: {VERSION}\n"
f"Минимальная версия: 2.0.14\n"
f"Актуальная версия: {latest_version}\n\n"
"Автообновление не удалось выполнить. Обновите плагин и перезапустите Cardinal."
)
return (
"⚠️ Доступна новая версия плагина NeverBoost\n\n"
f"Текущая версия: {VERSION}\n"
f"Актуальная версия: {latest_version}\n\n"
"Автообновление не удалось выполнить. Обновите плагин вручную."
)
def plugin_auto_updated_text(latest_version: str) -> str:
return (
"⚠️ Вышло новое обновление плагина NeverBoost.\n"
"Файл уже обновлён автоматически.\n\n"
"Во избежание багов напишите /restart"
)
def version_check_loop(cardinal: Cardinal) -> None:
global PLUGIN_OUTDATED_NOTIFIED_THIS_SESSION
while True:
try:
outdated, required_update, latest_version = plugin_outdated_info(ttl_sec=PLUGIN_VERSION_CHECK_INTERVAL_SEC)
if outdated or required_update:
if PLUGIN_AUTO_UPDATE_READY_VERSION == latest_version:
PLUGIN_OUTDATED_NOTIFIED_THIS_SESSION = True
else:
updated, details = apply_plugin_auto_update(latest_version)
if updated:
notify_authorized_users(cardinal, plugin_auto_updated_text(latest_version))
PLUGIN_OUTDATED_NOTIFIED_THIS_SESSION = True
elif not PLUGIN_OUTDATED_NOTIFIED_THIS_SESSION:
logger.warning("NeverBoost plugin auto-update failed: %s", details)
notify_authorized_users(cardinal, plugin_outdated_warning_text(latest_version, required_update))
PLUGIN_OUTDATED_NOTIFIED_THIS_SESSION = True
except Exception:
logger.debug("TRACEBACK", exc_info=True)
time.sleep(PLUGIN_VERSION_CHECK_INTERVAL_SEC)
def auto_sync_loop(cardinal: Cardinal) -> None:
while True:
settings = get_settings()
interval = max(int(settings.get("auto_sync_interval_sec", 60) or 60), 10)
time.sleep(interval)
if not get_settings().get("auto_sync_enabled", True):
logger.info("NeverBoost auto sync skipped: disabled")
continue
logger.info("NeverBoost auto sync tick")
sync_lots_availability_safe(cardinal, silent=True)
def extract_invite_code(text: str) -> str | None:
if not text:
return None
patterns = (
r"(?:https?://)?discord\.gg/([A-Za-z0-9-]+)",
r"(?:https?://)?discord(?:app)?\.com/invite/([A-Za-z0-9-]+)",
)
for pattern in patterns:
match = re.search(pattern, text, flags=re.IGNORECASE)
if match:
return match.group(1).strip()
stripped = text.strip()
if re.fullmatch(r"[A-Za-z0-9-]{2,}", stripped):
return stripped
return None
def fetch_invite_data(invite_code: str) -> dict | None:
try:
response = requests.get(
f"https://discord.com/api/v9/invites/{invite_code}?with_counts=true",
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"},
timeout=10,
)
if response.status_code != 200:
return None
data = response.json()
return data if isinstance(data, dict) else None
except Exception:
logger.debug("TRACEBACK", exc_info=True)
return None
def is_join_requests(invite_data: dict) -> bool:
features = invite_data.get("guild", {}).get("features", [])
if not isinstance(features, list):
return False
manual_approval_enabled = "MEMBER_VERIFICATION_MANUAL_APPROVAL" in features
try:
invite_flags = int(invite_data.get("flags", 0) or 0)
except Exception:
invite_flags = 0
application_bypass = bool(invite_flags & (1 << 3))
return manual_approval_enabled and not application_bypass
def get_lot_name(cardinal: Cardinal, value: str) -> str | None:
value = value.strip()
if not value:
return None
match = LOT_ID_RE.search(value)
if match:
lot_id = int(match.group(1))
try:
lot_fields = cardinal.account.get_lot_fields(lot_id)
fields = getattr(lot_fields, "fields", {}) or {}
name = str(fields.get("fields[summary][ru]", "") or "").strip()
if name:
return name
except Exception:
logger.debug("TRACEBACK", exc_info=True)
if cardinal.profile is not None:
lot_obj = cardinal.profile.get_lot(lot_id)
if lot_obj is not None and str(getattr(lot_obj, "title", "") or "").strip():
return str(lot_obj.title).strip()
return value
def ensure_delivery_binding(cardinal: Cardinal, lot_name: str) -> None:
if lot_name not in cardinal.AD_CFG.sections():
cardinal.AD_CFG.add_section(lot_name)
cardinal.AD_CFG.set(lot_name, "response", DELIVERY_TEXTS["ru"])
cardinal.save_config(cardinal.AD_CFG, "configs/auto_delivery.cfg")
def resolve_binding_lot_name(cardinal: Cardinal, binding: dict) -> str:
lot_name = str(binding.get("lot", "") or "").strip()
lot_id = str(binding.get("lot_id", "") or "").strip()
if lot_id:
try:
lot_fields = cardinal.account.get_lot_fields(int(lot_id))
fields = getattr(lot_fields, "fields", {}) or {}
name = str(fields.get("fields[summary][ru]", "") or "").strip()
if name:
return name
except Exception:
logger.debug("TRACEBACK", exc_info=True)
if cardinal.profile is not None:
try:
lot_obj = cardinal.profile.get_lot(int(lot_id))
if lot_obj is not None and str(getattr(lot_obj, "title", "") or "").strip():
return str(lot_obj.title).strip()
except Exception:
logger.debug("TRACEBACK", exc_info=True)
return lot_name
def find_binding_for_text(description_text: str, duration: str) -> dict | None:
bindings = [i for i in get_bindings() if i.get("duration") == duration]
for item in bindings:
lot_name = str(item.get("lot", "") or "").strip()
if lot_name and lot_name in description_text:
return item
return None
def normalize_bindings_with_profile(cardinal: Cardinal) -> int:
bindings = get_bindings()
updated = False
for item in bindings:
resolved_name = resolve_binding_lot_name(cardinal, item)
if resolved_name and resolved_name != str(item.get("lot", "") or ""):
item["lot"] = resolved_name
updated = True
if updated:
save_bindings(bindings)
return len(bindings) if updated else 0
def sync_delivery_bindings(cardinal: Cardinal) -> None:
bindings = get_bindings()
for item in bindings:
lot_name = str(item.get("lot", "") or "").strip()
if lot_name:
ensure_delivery_binding(cardinal, lot_name)
def remove_delivery_binding_if_unused(cardinal: Cardinal, lot_name: str) -> None:
if any(item["lot"] == lot_name for item in get_bindings()):
return
if lot_name in cardinal.AD_CFG.sections():
cardinal.AD_CFG.remove_section(lot_name)
cardinal.save_config(cardinal.AD_CFG, "configs/auto_delivery.cfg")
def purge_legacy_delivery_bindings(cardinal: Cardinal) -> int:
removed = 0
for section in list(cardinal.AD_CFG.sections()):
try:
response = cardinal.AD_CFG[section].get("response")
except Exception:
response = None
if response in DELIVERY_TEXTS.values():
cardinal.AD_CFG.remove_section(section)
removed += 1
if removed:
cardinal.save_config(cardinal.AD_CFG, "configs/auto_delivery.cfg")
return removed
def order_is_confirmed(order_obj) -> bool:
try:
return getattr(order_obj, "status", None) == OrderStatuses.CLOSED
except Exception:
return False
def get_unconfirmed_orders(cardinal: Cardinal, buyer_username: str, exclude_order_ids: list[str] | None = None) -> list[str]:
if not buyer_username:
return []
result: list[str] = []
seen = set()
start_from = None
excluded = {str(item or "").strip().upper() for item in (exclude_order_ids or []) if str(item or "").strip()}
try:
while True:
start_from, sales, _loc, _sub = cardinal.account.get_sales(
start_from=start_from,
include_paid=True,
include_closed=False,
include_refunded=False,
buyer=buyer_username,
state="paid",
)
if not sales:
break
for sale in sales:
order_id = str(getattr(sale, "id", "") or "").strip().upper()
if not order_id or order_id in excluded or order_id in seen:
continue
seen.add(order_id)
result.append(order_id)
if not start_from or len(result) >= 20:
break
except Exception:
logger.debug("TRACEBACK", exc_info=True)
return result
def format_unconfirmed_orders(order_ids: list[str]) -> str:
return "\n".join(f"{index}. #{order_id} - https://funpay.com/orders/{order_id}/" for index, order_id in enumerate(order_ids, 1))
def check_unconfirmed_orders(cardinal: Cardinal, chat_id: int | str, buyer_name: str, locale: str,
current_order_ids: list[str] | None = None, repeat: bool = False) -> bool:
order_ids = get_unconfirmed_orders(cardinal, buyer_name, exclude_order_ids=current_order_ids)
if not order_ids:
return False
key = "unconfirmed_block_repeat" if repeat else "unconfirmed_block"
cardinal.send_message(
chat_id,
buyer_text(locale, key, orders=format_unconfirmed_orders(order_ids)),
buyer_name,
watermark=False,
)
logger.info(
"NeverBoost blocked by unconfirmed orders: buyer=%s current_order=%s blocked=%s repeat=%s",
buyer_name,
current_order_ids,
order_ids,
repeat,
)
return True
def buyer_has_review_last_month(cardinal: Cardinal, buyer_username: str, days: int = 30) -> bool:
if not buyer_username:
return False
threshold = datetime.now() - timedelta(days=days)
start_from = None
checked = 0
try:
while True:
start_from, sales, _loc, _sub = cardinal.account.get_sales(
start_from=start_from,
include_paid=True,
include_closed=True,
include_refunded=True,
buyer=buyer_username,
)
if not sales:
return False
for sale in sales:
checked += 1
sale_date = getattr(sale, "date", None)
if isinstance(sale_date, datetime) and sale_date < threshold:
return False
try:
order_full = cardinal.account.get_order(str(getattr(sale, "id", "") or ""))
except Exception:
logger.debug("TRACEBACK", exc_info=True)
continue
review_obj = getattr(order_full, "review", None)
if review_obj and str(getattr(review_obj, "text", "") or "").strip():
return True
if not start_from or checked >= 80:
return False
except Exception:
logger.debug("TRACEBACK", exc_info=True)
return False
def post_delivery_reminders(cardinal: Cardinal, source_order_id: str, chat_id: int | str, buyer_name: str, locale: str,
order_ids: list[str] | None = None) -> None:
try:
target_order_ids = unique_order_ids(order_ids or [source_order_id])
try:
order_obj = cardinal.account.get_order(source_order_id)
except Exception:
logger.debug("TRACEBACK", exc_info=True)
order_obj = None
confirmed = order_is_confirmed(order_obj)
has_recent_review = buyer_has_review_last_month(cardinal, buyer_name, days=30)
if confirmed:
if not has_recent_review:
time.sleep(30)
cardinal.send_message(chat_id, buyer_text(locale, "please_review"), buyer_name, watermark=False)
return
time.sleep(10)
try:
confirmed = order_is_confirmed(cardinal.account.get_order(source_order_id))
except Exception:
logger.debug("TRACEBACK", exc_info=True)
confirmed = False
if confirmed:
if not has_recent_review:
time.sleep(30)
cardinal.send_message(chat_id, buyer_text(locale, "please_review"), buyer_name, watermark=False)
return
if len(target_order_ids) > 1:
cardinal.send_message(
chat_id,
buyer_text(locale, "please_confirm_multiple", orders=format_order_links(target_order_ids)),
buyer_name,
watermark=False,
)
else:
cardinal.send_message(chat_id, buyer_text(locale, "please_confirm"), buyer_name, watermark=False)
if not has_recent_review:
time.sleep(10)
cardinal.send_message(chat_id, buyer_text(locale, "please_review"), buyer_name, watermark=False)
except Exception:
logger.debug("TRACEBACK", exc_info=True)
class NeverBoostApiClient:
def __init__(self, api_key: str | None = None):
self.api_key = (api_key or "").strip()
def _headers(self) -> dict:
headers = {"Accept": "application/json"}
if self.api_key:
headers["X-API-Key"] = self.api_key
return headers
def _request(self, method: str, endpoint: str, payload: dict | None = None) -> tuple[bool, dict]:
try:
response = requests.request(
method=method,
url=f"{API_BASE_URL}{endpoint}",
headers=self._headers(),
json=payload if payload is not None else None,
timeout=15,
)
try:
data = response.json() if response.content else {}
except Exception:
data = {"raw": response.text}
logger.info(f"API {method} {endpoint} -> {response.status_code}: {data}")
return response.ok, data if isinstance(data, dict) else {}
except Exception as e:
logger.debug("TRACEBACK", exc_info=True)
logger.info(f"API {method} {endpoint} -> EXCEPTION: {e}")
return False, {}
def create_boost_order(self, order_id: str, duration: str, invite_url: str, count: int) -> tuple[bool, dict]:
return self._request("POST", "/boost", {
"order_id": order_id,
"duration": duration if duration in {"oneMonth", "threeMonth"} else "oneMonth",
"url": invite_url,
"count": int(count),
})
def get_order(self, order_id: str) -> tuple[bool, dict]:
return self._request("GET", f"/order/{order_id}")
def balance(self) -> tuple[bool, dict]:
return self._request("GET", "/balance")
def stock(self) -> tuple[bool, dict]:
return self._request("GET", "/stock")
def prices(self) -> tuple[bool, dict]:
return self._request("GET", "/prices")
def watch_api_order_status(cardinal: Cardinal, source_order_id: str, api_order_id: str, chat_id: int | str,
buyer_name: str, locale: str, requested_boosts: int, api_spent_total: float) -> None:
# NeverBoost: дожидаемся финального статуса API-заказа и только потом шлем итог покупателю.
api_key = get_api_key()
if not api_key:
return
client = NeverBoostApiClient(api_key)
final_statuses = {"completed", "partially_completed", "failed"}
for _ in range(120):
ok, data = client.get_order(api_order_id)
if ok and isinstance(data, dict):
order_data = data.get("order", data)
if isinstance(order_data, dict):
status = str(order_data.get("status", "") or "").lower()
boosted = int(order_data.get("boosted", 0) or 0)
requested = int(order_data.get("requested", requested_boosts) or requested_boosts)
if status in final_statuses:
try:
final_amount = float(order_data.get("final_amount", order_data.get("amount", api_spent_total)) or api_spent_total)
except Exception:
final_amount = float(api_spent_total or 0.0)
try:
refund_amount = float(order_data.get("refund_amount", 0.0) or 0.0)
except Exception:
refund_amount = 0.0
api_spent_total = round(max(final_amount - refund_amount, 0.0), 2)
source_sum = 0.0
try:
src_order = cardinal.account.get_order(source_order_id)
source_sum = float(getattr(src_order, "sum", getattr(src_order, "price", 0.0)) or 0.0)
except Exception:
logger.debug("TRACEBACK", exc_info=True)
issued_ratio = min(max((boosted / requested) if requested > 0 else 0.0, 0.0), 1.0)
sold_gross = round(source_sum * issued_ratio, 2)
sold_net = round(sold_gross * 0.97, 2)
earned = round(sold_net - float(api_spent_total or 0.0), 2)
final_locale = extract_chat_language(cardinal, chat_id, buyer_name) or locale
upsert_orders_history_row({
"source_order_id": source_order_id,
"api_order_id": api_order_id,
"buyer_name": buyer_name,
"chat_id": chat_id,
"created_ts": time.time(),
"status": status,
"boosted": boosted,
"requested": requested,
"api_spent_total": round(float(api_spent_total or 0.0), 2),
"profit": earned,
"finalized_ts": time.time(),
})
append_stats_row(boosted, earned)
logger.info(
"NeverBoost stats recorded from final status: source_order=%s status=%s boosted=%s profit=%s",
source_order_id,
status,
boosted,
earned,
)
pending = get_pending(chat_id)
queue_orders = get_queue_orders(pending)
queued_order = queue_orders[0] if queue_orders else None
chained_order_ids = unique_order_ids([source_order_id] + ([str(queued_order.get("order_id", "") or "")] if queued_order else []))
if status == "completed":
cardinal.send_message(chat_id, buyer_text(final_locale, "status_completed"), buyer_name, watermark=False)
elif status == "partially_completed":
cardinal.send_message(
chat_id,
buyer_text(final_locale, "status_partial", boosted=boosted, requested=requested),
buyer_name,
watermark=False,
)
else:
if boosted <= 0:
try:
current_order = cardinal.account.get_order(source_order_id)
current_status = getattr(current_order, "status", None)
logger.info("NeverBoost refund pre-check: source_order=%s status=%s", source_order_id, current_status)
except Exception:
current_status = None
logger.debug("TRACEBACK", exc_info=True)
try:
cardinal.account.refund(source_order_id)
logger.info("NeverBoost FunPay refund issued: source_order=%s", source_order_id)
except Exception as exc:
logger.info("NeverBoost FunPay refund failed: source_order=%s error=%s", source_order_id, exc)
logger.debug("TRACEBACK", exc_info=True)
cardinal.send_message(chat_id, buyer_text(final_locale, "status_failed_refund"), buyer_name, watermark=False)
else:
cardinal.send_message(chat_id, buyer_text(final_locale, "status_failed_no_refund"), buyer_name, watermark=False)
if queued_order is not None:
next_qty = max(int(queued_order.get("qty", 1) or 1), 1)
next_order_id = str(queued_order.get("order_id", "") or "").strip()
if next_order_id:
next_pending = dict(pending or {})
remaining_queue = queue_orders[1:]
next_pending["order_id"] = next_order_id
next_pending["qty"] = next_qty
next_pending["chat_id"] = queued_order.get("chat_id", chat_id)
next_pending["source_orders"] = [make_source_order_payload(next_order_id, next_qty, queued_order.get("chat_id", chat_id))]
next_pending["chain_order_ids"] = chained_order_ids
if remaining_queue:
next_pending["queue_orders"] = remaining_queue
next_pending["queued_order"] = remaining_queue[0]
else:
next_pending.pop("queue_orders", None)
next_pending.pop("queued_order", None)
next_pending.pop("invite_choice_pending", None)
next_pending.pop("blocked_unconfirmed_orders", None)
set_pending(chat_id, next_pending)
cardinal.send_message(
chat_id,
delivery_text(final_locale, next_qty),
buyer_name,
watermark=False,
)
logger.info(
"NeverBoost switched to chained queued order after final status: completed_order=%s next_order=%s next_qty=%s status=%s",
source_order_id,
next_order_id,
next_qty,
status,
)
clear_processing(source_order_id)
return
if status == "completed":
Thread(
target=post_delivery_reminders,
args=(cardinal, source_order_id, chat_id, buyer_name, final_locale, chained_order_ids),
daemon=True,
).start()
elif status in {"partially_completed", "failed"}:
Thread(
target=post_delivery_reminders,
args=(cardinal, source_order_id, chat_id, buyer_name, final_locale, chained_order_ids),
daemon=True,
).start()
if pending is not None:
clear_pending(chat_id)
logger.info(
"NeverBoost API final status: source_order=%s api_order=%s status=%s boosted=%s requested=%s gross=%s net=%s cost=%s profit=%s",
source_order_id,
api_order_id,
status,
boosted,
requested,
sold_gross,
sold_net,
round(float(api_spent_total or 0.0), 2),
earned,
)
notify_authorized_users(
cardinal,
(
f"📊 NeverBoost заказ завершён\n\n"
f"Заказ FunPay: #{source_order_id}\n"
f"API заказ: {api_order_id}\n"
f"Покупатель: {buyer_name}\n"
f"Статус: {status}\n"
f"Выдано: {boosted}/{requested}\n"
f"Продано: {sold_gross:.2f}\n"
f"После комиссии FP: {sold_net:.2f}\n"
f"Себестоимость API: {float(api_spent_total or 0.0):.2f}\n"
f"Чистая прибыль: {earned:.2f}"
)
)
clear_processing(source_order_id)
return
time.sleep(5)
def watch_manual_api_order_status(cardinal: Cardinal, api_order_id: str, chat_id: int | str, requested_boosts: int) -> None:
api_key = get_api_key()
if not api_key:
return
client = NeverBoostApiClient(api_key)
final_statuses = {"completed", "partially_completed", "failed"}
for _ in range(120):
ok, data = client.get_order(api_order_id)
if ok and isinstance(data, dict):
order_data = data.get("order", data)
if isinstance(order_data, dict):
status = str(order_data.get("status", "") or "").lower()
boosted = int(order_data.get("boosted", 0) or 0)
requested = int(order_data.get("requested", requested_boosts) or requested_boosts)
if status in final_statuses:
if status == "completed":
text = f"✅ Ручная выдача завершена\n\nВыдано: {boosted}/{requested}"
elif status == "partially_completed":
text = f"⚠️ Ручная выдача завершена частично\n\nВыдано: {boosted}/{requested}"
else:
text = f"❌ Ручная выдача завершилась ошибкой\n\nВыдано: {boosted}/{requested}"
notify_authorized_users(cardinal, f"🛰 {text}\nAPI заказ: {api_order_id}")
return
time.sleep(5)
def send_menu(cardinal: Cardinal, chat_id: int) -> None:
cardinal.telegram.bot.send_message(chat_id, format_menu_text(), reply_markup=build_menu())
def open_menu_from_message(message: Message) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
if message.from_user.id not in cardinal.telegram.authorized_users:
return
send_menu(cardinal, message.chat.id)
def open_menu_from_callback(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
bot.edit_message_text(format_menu_text(), call.message.chat.id, call.message.id, reply_markup=build_menu())
bot.answer_callback_query(call.id)
def act_add_lot(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
result = bot.send_message(
call.message.chat.id,
"Отправь ссылку на лот FunPay, ID лота или точное название лота.\n\nПример:\nhttps://funpay.com/lots/offer?id=123456",
reply_markup=build_cancel_menu(),
)
cardinal.telegram.set_state(call.message.chat.id, result.id, call.from_user.id, STATE_ADD_LOT)
bot.answer_callback_query(call.id)
def act_manual_boost(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
result = bot.send_message(
call.message.chat.id,
"Отправь Discord ссылку, куда нужно выдать бусты вручную.\n\nПример:\ndiscord.gg/neverboost",
reply_markup=build_cancel_menu(),
)
cardinal.telegram.set_state(call.message.chat.id, result.id, call.from_user.id, STATE_MANUAL_INVITE)
bot.answer_callback_query(call.id)
def set_manual_invite_message(message: Message) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
cardinal.telegram.clear_state(message.chat.id, message.from_user.id, True)
invite_code = extract_invite_code(message.text or "")
if not invite_code:
bot.reply_to(message, "Не вижу нормальную Discord ссылку.")
return
invite_data = fetch_invite_data(invite_code)
if not invite_data:
bot.reply_to(message, "Ссылка не работает или истекла.")
return
if is_join_requests(invite_data):
bot.reply_to(message, "На сервере включены заявки на вступление. Нужна ссылка без этой проблемы.")
return
set_manual_draft(message.from_user.id, {"invite_code": invite_code, "chat_id": message.chat.id})
bot.send_message(message.chat.id, "Теперь выбери срок бустов.", reply_markup=build_manual_duration_menu())
def set_manual_duration(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
draft = get_manual_draft(call.from_user.id)
if draft is None:
bot.answer_callback_query(call.id, "Сессия истекла. Начни заново.", show_alert=True)
return
duration = call.data.replace("neverboost:manual_duration:", "", 1)
if duration not in {"oneMonth", "threeMonth"}:
bot.answer_callback_query(call.id)
return
draft["duration"] = duration
set_manual_draft(call.from_user.id, draft)
result = bot.send_message(
call.message.chat.id,
"Теперь отправь, сколько бустов нужно выдать.\n\nПример: 2",
reply_markup=build_cancel_menu(),
)
cardinal.telegram.set_state(call.message.chat.id, result.id, call.from_user.id, STATE_MANUAL_QTY)
bot.answer_callback_query(call.id)
def set_manual_qty_message(message: Message) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
cardinal.telegram.clear_state(message.chat.id, message.from_user.id, True)
draft = get_manual_draft(message.from_user.id)
if draft is None:
bot.reply_to(message, "Сессия истекла. Начни заново.")
return
try:
qty = max(int((message.text or "").strip()), 1)
except Exception:
bot.reply_to(message, "Введи число, например: 2")
return
api_key = get_api_key()
if not api_key:
clear_manual_draft(message.from_user.id)
bot.reply_to(message, "Сначала добавь API ключ в разделе API и наличие.")
return
duration = str(draft.get("duration") or "oneMonth")
invite_code = str(draft.get("invite_code") or "")
invite_url = f"https://discord.gg/{invite_code}"
api_order_id = uuid.uuid4().hex
client = NeverBoostApiClient(api_key)
ok, resp = client.create_boost_order(api_order_id, duration, invite_url, qty)
if not ok:
ok_check, check_resp = client.get_order(api_order_id)
order_check = check_resp.get("order") if ok_check and isinstance(check_resp, dict) else None
if isinstance(order_check, dict):
status = str(order_check.get("status", "")).lower()
if status and status != "failed":
ok = True
resp = order_check
if not ok:
clear_manual_draft(message.from_user.id)
bot.send_message(message.chat.id, "❌ Не удалось создать ручную выдачу.")
logger.info("NeverBoost manual create failed: response=%s", resp)
return
clear_manual_draft(message.from_user.id)
bot.send_message(
message.chat.id,
f"🛰 Ручная выдача создана\n\nAPI заказ: {api_order_id}\nСрок: {'1 месяц' if duration == 'oneMonth' else '3 месяца'}\nБустов: {qty}",
)
Thread(target=watch_manual_api_order_status, args=(cardinal, api_order_id, message.chat.id, qty), daemon=True).start()
def act_set_api_key(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
result = bot.send_message(call.message.chat.id, "Отправь API ключ NeverBoost одним сообщением.", reply_markup=build_cancel_menu())
cardinal.telegram.set_state(call.message.chat.id, result.id, call.from_user.id, STATE_SET_API_KEY)
bot.answer_callback_query(call.id)
def set_api_key_message(message: Message) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
cardinal.telegram.clear_state(message.chat.id, message.from_user.id, True)
api_key = (message.text or "").strip()
if not api_key:
bot.reply_to(message, "API ключ не может быть пустым.")
return
save_api_key(api_key)
logger.info("NeverBoost API key updated")
bot.send_message(message.chat.id, "API ключ сохранён.")
send_menu(cardinal, message.chat.id)
def set_boosts_per_unit_message(message: Message) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
cardinal.telegram.clear_state(message.chat.id, message.from_user.id, True)
pending_bind = get_pending_bind(message.from_user.id)
if pending_bind is None:
bot.reply_to(message, "Нет ожидающего лота для привязки.")
return
try:
boosts_per_unit = max(int((message.text or "").strip()), 1)
except Exception:
bot.reply_to(message, "Введи число, например: 1 или 2")
return
lot_name = str(pending_bind.get("lot", "") or "").strip()
lot_id = str(pending_bind.get("lot_id", "") or "").strip()
duration = str(pending_bind.get("duration", "oneMonth") or "oneMonth")
bindings = get_bindings()
if not any(i["lot"] == lot_name and i["duration"] == duration and max(int(i.get("boosts_per_unit", 1) or 1), 1) == boosts_per_unit for i in bindings):
bindings.append({"lot": lot_name, "lot_id": lot_id, "duration": duration, "boosts_per_unit": boosts_per_unit})
save_bindings(bindings)
clear_pending_bind(message.from_user.id)
logger.info("NeverBoost: linked lot %s duration=%s boosts_per_unit=%s", lot_name, duration, boosts_per_unit)
bot.send_message(
message.chat.id,
f"Готово. Лот добавлен:\n{lot_name}\nСрок бустов: {'1 месяц' if duration == 'oneMonth' else '3 месяца'}\nСколько бустов получает покупатель за 1 покупку: {boosts_per_unit}"
)
send_menu(cardinal, message.chat.id)
def sync_stock_menu(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
changed, text = sync_lots_availability(cardinal)
logger.info("NeverBoost stock sync completed: changed=%s", changed)
bot.answer_callback_query(call.id)
bot.send_message(call.message.chat.id, text)
def open_settings_menu(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
bot.edit_message_text(format_settings_text(), call.message.chat.id, call.message.id, reply_markup=build_settings_menu())
bot.answer_callback_query(call.id)
def open_stats_menu(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
logger.info("NeverBoost stats menu opened: user=%s chat=%s", getattr(call.from_user, "id", None), getattr(call.message.chat, "id", None))
bot.edit_message_text(format_stats_text(), call.message.chat.id, call.message.id, reply_markup=K().row(B("⬅️ Назад", callback_data="neverboost:menu")))
bot.answer_callback_query(call.id)
def open_orders_menu(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
bot.edit_message_text(format_orders_text(cardinal), call.message.chat.id, call.message.id, reply_markup=K().row(B("⬅️ Назад", callback_data="neverboost:menu")))
bot.answer_callback_query(call.id)
def toggle_auto_sync(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
current = get_settings()
new_value = not bool(current.get("auto_sync_enabled", True))
save_settings({"auto_sync_enabled": new_value})
logger.info("NeverBoost auto sync toggled: enabled=%s", new_value)
bot.edit_message_text(format_settings_text(), call.message.chat.id, call.message.id, reply_markup=build_settings_menu())
bot.answer_callback_query(call.id, "Авто-синк обновлен")
def toggle_welcome(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
current = get_settings()
new_value = not bool(current.get("welcome_enabled", True))
save_settings({"welcome_enabled": new_value})
logger.info("NeverBoost welcome toggled: enabled=%s", new_value)
bot.edit_message_text(format_settings_text(), call.message.chat.id, call.message.id, reply_markup=build_settings_menu())
bot.answer_callback_query(call.id, "Приветствие обновлено")
def toggle_stock_command(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
current = get_settings()
new_value = not bool(current.get("stock_command_enabled", True))
save_settings({"stock_command_enabled": new_value})
logger.info("NeverBoost stock command toggled: enabled=%s", new_value)
bot.edit_message_text(format_settings_text(), call.message.chat.id, call.message.id, reply_markup=build_settings_menu())
bot.answer_callback_query(call.id, "#stock обновлено")
def ask_price_rule(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
duration = call.data.replace("neverboost:price_rule:", "", 1)
if duration not in {"oneMonth", "threeMonth"}:
bot.answer_callback_query(call.id)
return
settings = get_settings()
rule_key = "price_rule_one_month" if duration == "oneMonth" else "price_rule_three_month"
current_rule = format_price_rule(settings.get(rule_key))
label = "1 месяц" if duration == "oneMonth" else "3 месяца"
result = bot.send_message(
call.message.chat.id,
f"Отправь правило авто-цены для {label}.\n\n"
f"Сейчас: {current_rule}\n"
f"Примеры:\n"
f"20% - цена на 20 процентов выше API\n"
f"2 - цена на 2 RUB выше API\n"
f"-5% - цена на 5 процентов ниже API\n"
f"-2 - цена на 2 RUB ниже API\n"
f"off - выключить авто-цену",
reply_markup=build_cancel_menu(),
)
state = STATE_SET_PRICE_RULE_ONE if duration == "oneMonth" else STATE_SET_PRICE_RULE_THREE
cardinal.telegram.set_state(call.message.chat.id, result.id, call.from_user.id, state)
bot.answer_callback_query(call.id)
def set_price_rule_message(message: Message, duration: str) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
cardinal.telegram.clear_state(message.chat.id, message.from_user.id, True)
try:
rule = parse_price_rule_input(message.text or "")
except Exception:
bot.reply_to(message, "Введи формат вроде 20%, 2, -5%, -2 или off")
return
key = "price_rule_one_month" if duration == "oneMonth" else "price_rule_three_month"
save_settings({key: rule})
logger.info("NeverBoost price rule updated: duration=%s rule=%s", duration, rule)
label = "1 месяц" if duration == "oneMonth" else "3 месяца"
bot.send_message(message.chat.id, f"Готово. Авто-цена для {label}: {format_price_rule(rule)}")
send_menu(cardinal, message.chat.id)
def set_price_rule_one_message(message: Message) -> None:
set_price_rule_message(message, "oneMonth")
def set_price_rule_three_message(message: Message) -> None:
set_price_rule_message(message, "threeMonth")
def open_bindings_menu(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
bot.edit_message_text(format_bindings_text(), call.message.chat.id, call.message.id, reply_markup=build_bindings_menu())
bot.answer_callback_query(call.id)
def open_binding_actions(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
try:
index = int(call.data.replace("neverboost:binding:", "", 1))
item = get_bindings()[index]
except Exception:
bot.answer_callback_query(call.id, "Привязка не найдена.", show_alert=True)
return
duration = "1 месяц" if item["duration"] == "oneMonth" else "3 месяца"
bpu = max(int(item.get("boosts_per_unit", 1) or 1), 1)
text = (
f"Настройка лота\n\n"
f"Лот: {item['lot']}\n"
f"ID лота: {item.get('lot_id', '') or '-'}\n"
f"Срок бустов: {duration}\n"
f"Сколько бустов получает покупатель за 1 покупку: {bpu}"
)
bot.edit_message_text(text, call.message.chat.id, call.message.id, reply_markup=build_binding_actions_menu(index, item))
bot.answer_callback_query(call.id)
def set_binding_duration_direct(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
try:
_, _, _, index_raw, duration = call.data.split(":", 4)
index = int(index_raw)
bindings = get_bindings()
bindings[index]["duration"] = duration
save_bindings(bindings)
item = bindings[index]
except Exception:
bot.answer_callback_query(call.id, "Не удалось изменить длительность.", show_alert=True)
return
logger.info("NeverBoost binding updated duration: lot=%s duration=%s", item["lot"], duration)
bot.answer_callback_query(call.id, "Длительность обновлена")
open_binding_actions(call)
def ask_binding_bpu_edit(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
try:
index = int(call.data.replace("neverboost:binding_bpu:", "", 1))
item = get_bindings()[index]
except Exception:
bot.answer_callback_query(call.id, "Привязка не найдена.", show_alert=True)
return
set_edit_bind(call.from_user.id, {"index": index})
result = bot.send_message(call.message.chat.id, f"Отправь новое число бустов за 1 покупку для лота:\n{item['lot']}", reply_markup=build_cancel_menu())
cardinal.telegram.set_state(call.message.chat.id, result.id, call.from_user.id, STATE_EDIT_BPU)
bot.answer_callback_query(call.id)
def cancel_bind_flow(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
cardinal.telegram.clear_state(call.message.chat.id, call.from_user.id, True)
clear_pending_bind(call.from_user.id)
clear_edit_bind(call.from_user.id)
clear_manual_draft(call.from_user.id)
bot.answer_callback_query(call.id, "Отменено")
bot.edit_message_text(format_menu_text(), call.message.chat.id, call.message.id, reply_markup=build_menu())
def set_binding_bpu_message(message: Message) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
cardinal.telegram.clear_state(message.chat.id, message.from_user.id, True)
edit = get_edit_bind(message.from_user.id)
if edit is None:
bot.reply_to(message, "Нет привязки для редактирования.")
return
try:
boosts_per_unit = max(int((message.text or "").strip()), 1)
index = int(edit.get("index"))
bindings = get_bindings()
bindings[index]["boosts_per_unit"] = boosts_per_unit
save_bindings(bindings)
item = bindings[index]
except Exception:
bot.reply_to(message, "Введи число, например: 1 или 2")
return
clear_edit_bind(message.from_user.id)
logger.info("NeverBoost binding updated boosts_per_unit: lot=%s bpu=%s", item["lot"], boosts_per_unit)
bot.send_message(message.chat.id, f"Готово. Теперь покупатель будет получать за 1 покупку: {boosts_per_unit} буст(а/ов)")
send_menu(cardinal, message.chat.id)
def check_api_key(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
api_key = get_api_key()
if not api_key:
bot.answer_callback_query(call.id, "API ключ не установлен.", show_alert=True)
return
client = NeverBoostApiClient(api_key)
ok, data = client.balance()
if not ok:
bot.answer_callback_query(call.id, "Не удалось проверить API ключ.", show_alert=True)
logger.info("NeverBoost API check failed: %s", data)
return
balance = data.get("balance", "?") if isinstance(data, dict) else "?"
bot.answer_callback_query(call.id)
bot.send_message(call.message.chat.id, f"API ключ рабочий.\nБаланс: {balance}")
def add_lot(message: Message) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
cardinal.telegram.clear_state(message.chat.id, message.from_user.id, True)
lot_name = get_lot_name(cardinal, message.text or "")
if not lot_name:
bot.reply_to(message, "Название лота не может быть пустым.")
return
lot_id_match = LOT_ID_RE.search(message.text or "")
set_pending_bind(message.from_user.id, {"lot": lot_name, "lot_id": lot_id_match.group(1) if lot_id_match else "", "chat_id": message.chat.id})
bot.send_message(
message.chat.id,
f"Выбери срок бустов для этого лота:\n{lot_name}",
reply_markup=build_duration_menu(),
)
def set_binding_duration(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
pending_bind = get_pending_bind(call.from_user.id)
if pending_bind is None:
bot.answer_callback_query(call.id, "Нет ожидающего лота для привязки.", show_alert=True)
return
duration = call.data.replace("neverboost:duration:", "", 1)
if duration not in {"oneMonth", "threeMonth"}:
bot.answer_callback_query(call.id)
return
lot_name = str(pending_bind.get("lot", "") or "").strip()
lot_id = str(pending_bind.get("lot_id", "") or "").strip()
if not lot_name:
clear_pending_bind(call.from_user.id)
bot.answer_callback_query(call.id, "Лот не найден.", show_alert=True)
return
pending_bind["duration"] = duration
pending_bind["lot_id"] = lot_id
set_pending_bind(call.from_user.id, pending_bind)
bot.answer_callback_query(call.id)
bot.edit_message_text(
f"Лот: {lot_name}\nДлительность: {'1 месяц' if duration == 'oneMonth' else '3 месяца'}\n\nТеперь отправь, сколько бустов выдаётся за 1 купленную штуку.\nНапример: 1 или 2",
call.message.chat.id,
call.message.id,
)
cardinal.telegram.set_state(call.message.chat.id, call.message.id, call.from_user.id, STATE_SET_BPU)
def open_remove_menu(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
if not get_bindings():
bot.answer_callback_query(call.id, "Нет привязанных лотов.", show_alert=True)
return
bot.edit_message_text(
"Удаление лота\n\nВыбери лот, который нужно отвязать.",
call.message.chat.id,
call.message.id,
reply_markup=build_remove_menu(),
)
bot.answer_callback_query(call.id)
def remove_selected_lot(call: CallbackQuery) -> None:
cardinal = __import__("cardinal").get_cardinal()
if cardinal is None or cardinal.telegram is None:
return
bot = cardinal.telegram.bot
bindings = get_bindings()
try:
idx = int(call.data.replace("neverboost:remove:", "", 1))
binding = bindings[idx]
except (ValueError, IndexError):
bot.answer_callback_query(call.id, "Лот уже отвязан или не найден.", show_alert=True)
bot.edit_message_text(format_menu_text(), call.message.chat.id, call.message.id, reply_markup=build_menu())
return
bindings.pop(idx)
save_bindings(bindings)
logger.info("NeverBoost: removed lot %s duration=%s boosts_per_unit=%s", binding["lot"], binding["duration"], binding.get("boosts_per_unit", 1))
bot.answer_callback_query(call.id, f"Удалён: {binding['lot']}")
bot.edit_message_text(format_menu_text(), call.message.chat.id, call.message.id, reply_markup=build_menu())
def log_new_order(cardinal: Cardinal, event) -> None:
lot_name = str(getattr(event.order, "description", "") or "")
duration = detect_duration(event.order)
binding = find_binding_for_text(lot_name, duration)
if binding is None:
logger.info(
"NeverBoost binding not matched for NEW_ORDER: id=%s duration=%s lot=%s",
getattr(event.order, "id", None),
duration,
lot_name,
)
return
remember_order_id(event.order.id)
boosts_per_unit = max(int(binding.get("boosts_per_unit", 1) or 1), 1)
requested_boosts = max(int(event.order.amount or 1), 1) * boosts_per_unit
locale = detect_order_locale(cardinal, event.order)
existing_pending = get_pending(event.order.chat_id)
if existing_pending is not None:
source_orders = get_pending_source_orders(existing_pending)
if source_orders:
active_source_order_id = str(existing_pending.get("order_id", "") or "").strip()
active_processing = get_processing(active_source_order_id) if active_source_order_id else None
current_qty = sum_order_qty(source_orders)
queued = make_source_order_payload(event.order.id, requested_boosts, event.order.chat_id)
queue_orders = get_queue_orders(existing_pending)
queue_orders.append(queued)
existing_pending["queue_orders"] = queue_orders
existing_pending["queued_order"] = queue_orders[0]
if active_processing is None:
existing_pending["invite_choice_pending"] = True
else:
existing_pending.pop("invite_choice_pending", None)
set_pending(event.order.chat_id, existing_pending)
chat = cardinal.account.get_chat_by_name(event.order.buyer_username)
chat_id = chat.id if chat else event.order.chat_id
if active_processing is None:
cardinal.send_message(
chat_id,
queue_invite_choice_text(locale, current_qty, current_qty + requested_boosts),
event.order.buyer_username,
watermark=False,
)
logger.info(
"NeverBoost queued second pending order before invite: current_order=%s queued_order=%s current_qty=%s queued_qty=%s",
existing_pending.get("order_id"),
event.order.id,
current_qty,
requested_boosts,
)
else:
logger.info(
"NeverBoost queued next order while current order is already processing: current_order=%s queued_order=%s current_qty=%s queued_qty=%s",
existing_pending.get("order_id"),
event.order.id,
current_qty,
requested_boosts,
)
return
mark_order_chat(event.order.chat_id, {"ts": time.time(), "order_id": event.order.id})
set_pending(event.order.chat_id, {
"order_id": event.order.id,
"buyer_username": event.order.buyer_username,
"locale": locale,
"qty": requested_boosts,
"duration": duration,
"lot": lot_name,
"boosts_per_unit": boosts_per_unit,
"chat_id": event.order.chat_id,
"source_orders": [make_source_order_payload(event.order.id, requested_boosts, event.order.chat_id)],
})
chat = cardinal.account.get_chat_by_name(event.order.buyer_username)
chat_id = chat.id if chat else event.order.chat_id
if check_unconfirmed_orders(
cardinal,
chat_id,
event.order.buyer_username,
locale,
current_order_ids=[str(getattr(event.order, "id", "") or "")],
repeat=False,
):
pending = get_pending(event.order.chat_id)
if pending is not None:
pending["blocked_unconfirmed_orders"] = True
set_pending(event.order.chat_id, pending)
else:
cardinal.send_message(chat_id, delivery_text(locale, requested_boosts), event.order.buyer_username, watermark=False)
logger.info(
"NeverBoost order: id=%s buyer=%s amount=%s boosts=%s lot=%s duration=%s bpu=%s price=%s %s locale=%s",
event.order.id,
event.order.buyer_username,
event.order.amount if event.order.amount else 1,
requested_boosts,
lot_name,
duration,
boosts_per_unit,
event.order.price,
event.order.currency,
locale,
)
def set_delivery_text_by_locale(cardinal: Cardinal, event) -> None:
config_obj = getattr(event, "config_section_obj", None)
if config_obj is None or config_obj.get("response") not in DELIVERY_TEXTS.values():
return
locale = detect_order_locale(cardinal, event.order)
setattr(event, "neverboost_original_response", config_obj.get("response"))
config_obj["response"] = DELIVERY_TEXTS.get(locale, DELIVERY_TEXTS["ru"])
logger.info("NeverBoost: selected locale %s for order %s", locale, event.order.id)
def restore_delivery_text(cardinal: Cardinal, event) -> None:
config_obj = getattr(event, "config_section_obj", None)
original = getattr(event, "neverboost_original_response", None)
if config_obj is None or original is None:
return
config_obj["response"] = original
def handle_new_message(cardinal: Cardinal, event) -> None:
# NeverBoost: после оплаты ждем invite-ссылку от покупателя и запускаем выдачу в API.
message = event.message
logger.info(
"NeverBoost new message event: chat=%s author=%s author_id=%s by_bot=%s text=%s",
getattr(message, "chat_id", None),
getattr(message, "author", None),
getattr(message, "author_id", None),
getattr(message, "by_bot", None),
(getattr(message, "text", None) or "")[:200],
)
settings = get_settings()
text = (getattr(message, "text", None) or "").strip()
if getattr(message, "author_id", 0) == 0 and getattr(message, "buyer_viewing", None) is not None:
sent = maybe_send_welcome_for_viewing_lot(cardinal, message, settings, buyer_viewing=message.buyer_viewing)
if sent:
return
if getattr(message, "author_id", 0) == 0:
if str(getattr(message, "author", "") or "").strip().lower() != "funpay":
return
order_match = re.search(r"оплатил заказ\s+#([A-Z0-9]+)", text, flags=re.IGNORECASE)
if order_match:
order_id = order_match.group(1)
if has_recent_order_id(order_id):
logger.info("NeverBoost fallback skipped: order %s was already processed by NEW_ORDER", order_id)
return
if has_pending_order_id(order_id):
logger.info("NeverBoost fallback skipped: pending already exists for order %s", order_id)
return
created = False
try:
order_obj = cardinal.account.get_order(order_id)
except Exception:
order_obj = None
logger.debug("TRACEBACK", exc_info=True)
if order_obj is not None:
created = create_pending_from_order(cardinal, order_obj)
if not created:
create_pending_from_payment_text(cardinal, message.chat_id, text)
return
if getattr(message, "by_bot", False):
return
if text.lower() == "#stock" and settings.get("stock_command_enabled", True):
locale = extract_chat_language(cardinal, message.chat_id, getattr(message, "author", None) or getattr(message, "chat_name", None)) or "ru"
cardinal.send_message(
message.chat_id,
stock_text(locale),
getattr(message, "author", None) or getattr(message, "chat_name", None),
watermark=False,
)
set_chat_activity(message.chat_id, {"ts": time.time(), "reason": "stock"})
return
pending_key, pending = find_pending_for_message(message)
if pending is None:
if not maybe_send_welcome_for_viewing_lot(cardinal, message, settings):
set_chat_activity(message.chat_id, {"ts": time.time(), "reason": "message", "author": getattr(message, "author", None)})
logger.info("NeverBoost new message ignored: no pending state for chat=%s", getattr(message, "chat_id", None))
return
buyer_username = pending.get("buyer_username")
if buyer_username and message.author and message.author.lower() != str(buyer_username).lower():
logger.info(
"NeverBoost new message ignored: author mismatch expected=%s got=%s",
buyer_username,
message.author,
)
return
locale = "en" if str(pending.get("locale") or "").lower() == "en" else "ru"
source_order_id = str(pending.get("order_id") or "")
locale = extract_chat_language(cardinal, message.chat_id, buyer_username or message.author) or locale
current_order_ids = unique_order_ids(
[source_order_id]
+ [item.get("order_id") for item in get_pending_source_orders(pending)]
+ [item.get("order_id") for item in get_queue_orders(pending)]
+ list(pending.get("chain_order_ids") or [])
)
if pending.get("invite_choice_pending"):
choice = text.strip()
queue_orders = get_queue_orders(pending)
queued = queue_orders[0] if queue_orders else None
current_source_orders = get_pending_source_orders(pending)
if queued is None or not current_source_orders:
pending.pop("invite_choice_pending", None)
pending.pop("queued_order", None)
pending.pop("queue_orders", None)
if pending_key is not None:
set_pending(pending_key, pending)
elif choice == "2":
current_source_orders.append(make_source_order_payload(queued.get("order_id"), int(queued.get("qty", 1) or 1), queued.get("chat_id")))
pending["source_orders"] = current_source_orders
pending["qty"] = sum_order_qty(current_source_orders)
queue_orders = queue_orders[1:]
if queue_orders:
pending["queue_orders"] = queue_orders
pending["queued_order"] = queue_orders[0]
else:
pending.pop("queue_orders", None)
pending.pop("queued_order", None)
pending.pop("invite_choice_pending", None)
if pending_key is not None:
set_pending(pending_key, pending)
cardinal.send_message(
message.chat_id,
delivery_text(locale, int(pending.get("qty", 1) or 1)),
buyer_username or message.chat_name,
watermark=False,
)
logger.info(
"NeverBoost merged queued pending order into current invite flow: base_order=%s merged_qty=%s total_qty=%s",
source_order_id,
queued.get("qty"),
pending.get("qty"),
)
return
elif choice == "1":
pending.pop("invite_choice_pending", None)
pending["queue_orders"] = queue_orders
pending["queued_order"] = queue_orders[0]
if pending_key is not None:
set_pending(pending_key, pending)
cardinal.send_message(
message.chat_id,
delivery_text(locale, int(pending.get("qty", 1) or 1)),
buyer_username or message.chat_name,
watermark=False,
)
logger.info(
"NeverBoost kept queued order separate until first invite is processed: base_order=%s queued_order=%s",
source_order_id,
queued.get("order_id"),
)
return
else:
cardinal.send_message(message.chat_id, buyer_text(locale, "pending_invite_choice_invalid"), buyer_username or message.chat_name, watermark=False)
return
blocked_once = bool(pending.get("blocked_unconfirmed_orders"))
if check_unconfirmed_orders(
cardinal,
message.chat_id,
buyer_username or message.chat_name,
locale,
current_order_ids=current_order_ids,
repeat=blocked_once,
):
pending["blocked_unconfirmed_orders"] = True
if pending_key is not None:
set_pending(pending_key, pending)
return
if pending.get("blocked_unconfirmed_orders"):
pending.pop("blocked_unconfirmed_orders", None)
if pending_key is not None:
set_pending(pending_key, pending)
cardinal.send_message(
message.chat_id,
delivery_text(locale, int(pending.get("qty", 1) or 1)),
buyer_username or message.chat_name,
watermark=False,
)
logger.info(
"NeverBoost resumed invite request after old orders confirmed: order=%s chat=%s",
source_order_id,
message.chat_id,
)
return
processing = get_processing(source_order_id)
if processing is not None:
logger.info(
"NeverBoost new message ignored: order already processing source_order=%s api_order=%s",
source_order_id,
processing.get("api_order_id"),
)
return
invite_code = extract_invite_code(message.text or "")
if not invite_code:
cardinal.send_message(message.chat_id, buyer_text(locale, "invalid"), buyer_username or message.chat_name, watermark=False)
logger.info("NeverBoost invite invalid format: order=%s chat=%s", pending.get("order_id"), message.chat_id)
return
invite_data = fetch_invite_data(invite_code)
if not invite_data:
cardinal.send_message(message.chat_id, buyer_text(locale, "bad"), buyer_username or message.chat_name, watermark=False)
logger.info("NeverBoost invite bad: order=%s code=%s", pending.get("order_id"), invite_code)
return
if is_join_requests(invite_data):
cardinal.send_message(message.chat_id, buyer_text(locale, "requests_on"), buyer_username or message.chat_name, watermark=False)
logger.info("NeverBoost invite join requests enabled: order=%s code=%s", pending.get("order_id"), invite_code)
return
api_key = get_api_key()
if not api_key:
cardinal.send_message(message.chat_id, buyer_text(locale, "api_fail"), buyer_username or message.chat_name, watermark=False)
logger.info("NeverBoost API key missing: order=%s", pending.get("order_id"))
return
qty = max(int(pending.get("qty", 1) or 1), 1)
duration = str(pending.get("duration") or "oneMonth")
api_order_id = uuid.uuid4().hex
set_processing(source_order_id, {
"api_order_id": api_order_id,
"buyer_username": buyer_username or message.chat_name,
"chat_id": message.chat_id,
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
})
upsert_orders_history_row({
"source_order_id": source_order_id,
"api_order_id": api_order_id,
"buyer_name": buyer_username or message.chat_name,
"chat_id": message.chat_id,
"created_ts": time.time(),
"duration": duration,
"status": "processing",
"boosted": 0,
"requested": qty,
})
invite_url = f"https://discord.gg/{invite_code}"
client = NeverBoostApiClient(api_key)
ok, resp = client.create_boost_order(api_order_id, duration, invite_url, qty)
if not ok:
ok_check, check_resp = client.get_order(api_order_id)
order_check = check_resp.get("order") if ok_check and isinstance(check_resp, dict) else None
if isinstance(order_check, dict):
status = str(order_check.get("status", "")).lower()
if status and status != "failed":
ok = True
resp = order_check
if not ok:
clear_processing(source_order_id)
cardinal.send_message(message.chat_id, buyer_text(locale, "api_fail"), buyer_username or message.chat_name, watermark=False)
logger.info("NeverBoost API create failed: order=%s response=%s", pending.get("order_id"), resp)
return
api_spent_total = 0.0
if isinstance(resp, dict):
try:
final_amount = float(resp.get("final_amount", resp.get("amount", 0.0)) or 0.0)
refund_amount = float(resp.get("refund_amount", 0.0) or 0.0)
api_spent_total = round(max(final_amount - refund_amount, 0.0), 2)
except Exception:
logger.debug("TRACEBACK", exc_info=True)
source_orders = get_pending_source_orders(pending)
queue_orders = get_queue_orders(pending)
queued_order = queue_orders[0] if queue_orders else None
chain_order_ids = unique_order_ids([item.get("order_id") for item in source_orders] + list(pending.get("chain_order_ids") or []))
if pending_key is not None:
if queued_order is not None and source_orders:
next_pending = dict(pending)
next_pending["chain_order_ids"] = unique_order_ids(chain_order_ids + [queued_order.get("order_id")])
next_pending["queue_orders"] = queue_orders
next_pending["queued_order"] = queued_order
set_pending(pending_key, next_pending)
else:
clear_pending_by_key(pending_key)
cardinal.send_message(
message.chat_id,
buyer_text(locale, "order_created", api_order_id=api_order_id),
buyer_username or message.chat_name,
watermark=False,
)
Thread(
target=watch_api_order_status,
args=(cardinal, source_order_id, api_order_id, message.chat_id, buyer_username or message.chat_name, locale, qty, api_spent_total),
daemon=True,
).start()
logger.info(
"NeverBoost invite valid and boost order created: source_order=%s api_order=%s code=%s qty=%s duration=%s",
pending.get("order_id"),
api_order_id,
invite_code,
qty,
duration,
)
def stop_wait_on_refund(cardinal: Cardinal, event) -> None:
try:
if event.order.status not in {OrderStatuses.REFUNDED, OrderStatuses.PARTIALLY_REFUNDED}:
return
clear_processing(str(getattr(event.order, "id", "") or ""))
pending = pop_pending_by_order_id(str(getattr(event.order, "id", "") or ""))
if not pending:
return
locale = "en" if str(pending.get("locale") or "").lower() == "en" else "ru"
buyer_username = str(pending.get("buyer_username") or "").strip() or str(getattr(event.order, "buyer_username", "") or "").strip()
chat_id = getattr(event.order, "chat_id", None)
if chat_id is None:
chat_id = pending.get("chat_id")
if chat_id is not None:
cardinal.send_message(chat_id, buyer_text(locale, "status_wait_cancelled_refund"), buyer_username or None, watermark=False)
logger.info(
"NeverBoost pending wait cancelled by refund: order=%s status=%s buyer=%s chat=%s",
getattr(event.order, "id", None),
getattr(event.order, "status", None),
buyer_username,
chat_id,
)
except Exception:
logger.debug("TRACEBACK", exc_info=True)
def bind_tg_handlers(cardinal: Cardinal) -> None:
if cardinal.telegram is None:
return
tg = cardinal.telegram
tg.msg_handler(open_menu_from_message, commands=COMMANDS)
tg.cbq_handler(open_menu_from_callback, lambda c: c.data == "neverboost:menu")
tg.cbq_handler(act_add_lot, lambda c: c.data == "neverboost:add")
tg.cbq_handler(act_manual_boost, lambda c: c.data == "neverboost:manual")
tg.cbq_handler(open_settings_menu, lambda c: c.data == "neverboost:settings")
tg.cbq_handler(open_orders_menu, lambda c: c.data == "neverboost:orders")
tg.cbq_handler(open_stats_menu, lambda c: c.data == "neverboost:stats")
tg.cbq_handler(toggle_auto_sync, lambda c: c.data == "neverboost:toggle_auto_sync")
tg.cbq_handler(toggle_welcome, lambda c: c.data == "neverboost:toggle_welcome")
tg.cbq_handler(toggle_stock_command, lambda c: c.data == "neverboost:toggle_stock")
tg.cbq_handler(ask_price_rule, lambda c: c.data.startswith("neverboost:price_rule:"))
tg.cbq_handler(cancel_bind_flow, lambda c: c.data == "neverboost:cancel_bind")
tg.cbq_handler(set_manual_duration, lambda c: c.data.startswith("neverboost:manual_duration:"))
tg.cbq_handler(set_binding_duration_direct, lambda c: c.data.startswith("neverboost:binding_duration:"))
tg.cbq_handler(ask_binding_bpu_edit, lambda c: c.data.startswith("neverboost:binding_bpu:"))
tg.cbq_handler(open_bindings_menu, lambda c: c.data == "neverboost:bindings")
tg.cbq_handler(open_binding_actions, lambda c: c.data.startswith("neverboost:binding:"))
tg.cbq_handler(set_binding_duration, lambda c: c.data.startswith("neverboost:duration:"))
tg.cbq_handler(act_set_api_key, lambda c: c.data == "neverboost:api_key")
tg.cbq_handler(check_api_key, lambda c: c.data == "neverboost:api_check")
tg.cbq_handler(sync_stock_menu, lambda c: c.data == "neverboost:sync_stock")
tg.cbq_handler(open_remove_menu, lambda c: c.data == "neverboost:remove_menu")
tg.cbq_handler(remove_selected_lot, lambda c: c.data.startswith("neverboost:remove:"))
tg.msg_handler(add_lot, func=lambda m: tg.check_state(m.chat.id, m.from_user.id, STATE_ADD_LOT))
tg.msg_handler(set_api_key_message, func=lambda m: tg.check_state(m.chat.id, m.from_user.id, STATE_SET_API_KEY))
tg.msg_handler(set_boosts_per_unit_message, func=lambda m: tg.check_state(m.chat.id, m.from_user.id, STATE_SET_BPU))
tg.msg_handler(set_binding_bpu_message, func=lambda m: tg.check_state(m.chat.id, m.from_user.id, STATE_EDIT_BPU))
tg.msg_handler(set_price_rule_one_message, func=lambda m: tg.check_state(m.chat.id, m.from_user.id, STATE_SET_PRICE_RULE_ONE))
tg.msg_handler(set_price_rule_three_message, func=lambda m: tg.check_state(m.chat.id, m.from_user.id, STATE_SET_PRICE_RULE_THREE))
tg.msg_handler(set_manual_invite_message, func=lambda m: tg.check_state(m.chat.id, m.from_user.id, STATE_MANUAL_INVITE))
tg.msg_handler(set_manual_qty_message, func=lambda m: tg.check_state(m.chat.id, m.from_user.id, STATE_MANUAL_QTY))
def pre_init_handler(cardinal: Cardinal) -> None:
cardinal.add_telegram_commands(UUID, [(cmd, "Открыть меню NeverBoost", True) for cmd in COMMANDS])
bind_tg_handlers(cardinal)
sync_lots_availability_safe(cardinal, silent=True)
Thread(target=auto_sync_loop, args=(cardinal,), daemon=True).start()
Thread(target=version_check_loop, args=(cardinal,), daemon=True).start()
logger.info("NeverBoost plugin initialized")
def post_init_handler(cardinal: Cardinal) -> None:
normalized = normalize_bindings_with_profile(cardinal)
removed = purge_legacy_delivery_bindings(cardinal)
if normalized:
logger.info("NeverBoost normalized bindings using profile data: %s", normalized)
if removed:
logger.info("NeverBoost removed legacy Cardinal auto-delivery bindings: %s", removed)
BIND_TO_PRE_INIT = [pre_init_handler]
BIND_TO_POST_INIT = [post_init_handler]
BIND_TO_NEW_MESSAGE = [handle_new_message]
BIND_TO_NEW_ORDER = [log_new_order]
BIND_TO_ORDER_STATUS_CHANGED = [stop_wait_on_refund]