a.gruzdev/bot_vk_ikp_prodagi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Соцсети: добавлена кнопка-ссылка на сайт infocyber.pro

Browse Source 
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Kenik
2026-06-25 14:21:52 +03:00
parent fc3361c7ea
commit 0ec849d80f
7 changed files with 160 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot.py
+114 -55
View File
@@ -40,6 +40,8 @@ NOTIFY_COOLDOWN_SEC = 120
BAN_STRIKES = 3
BAN_WINDOW_SEC = 60
BAN_SEC = 600
# Сколько секунд кэшировать список рантайм-админов из БД
ADMINS_CACHE_TTL = 30
from validators import (
RU_PHONE_REGEX, INTL_PHONE_REGEX, NAME_REGEX,
@@ -48,6 +50,37 @@ from validators import (
from keyboards import * # все get_*_keyboard
# Хэндл lock-файла держим в глобали на весь процесс — пока он открыт, блокировка жива.
_INSTANCE_LOCK_HANDLE = None
def _acquire_single_instance_lock() -> bool:
"""Не даёт запустить второй экземпляр бота (два параллельных longpoll ломают UX).
Берём неблокирующую эксклюзивную блокировку на bot.lock в каталоге скрипта.
Кросс-платформенно: msvcrt на Windows, fcntl на POSIX. Если уже заблокирован
другим процессом — возвращаем False (выживает уже запущенный инстанс).
"""
global _INSTANCE_LOCK_HANDLE
import os
lock_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bot.lock")
handle = None
try:
handle = open(lock_path, "w")
if os.name == "nt":
import msvcrt
msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
else:
import fcntl
fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
if handle is not None:
handle.close() # лок занят (или сбой) — не оставляем хэндл открытым
return False
_INSTANCE_LOCK_HANDLE = handle # держим открытым до конца процесса
return True
# ==========================================
# КЛАСС БОТА И ЛОГИКА ОБРАБОТКИ
# ==========================================
@@ -69,6 +102,10 @@ class VKBotApplication:
self._flood_strikes: Dict[int, list] = defaultdict(list) # user_id -> метки срабатываний флуда
self._banned_until: Dict[int, float] = {} # user_id -> до какого времени забанен
self._last_notify: Dict[int, float] = {} # user_id -> время последнего уведомления оператора
self._last_antiflood_cleanup: float = 0.0 # когда последний раз чистили словари анти-флуда
# Кэш рантайм-админов (из БД), чтобы не бить в SQLite на каждое сообщение
self._admins_cache: Optional[set] = None
self._admins_cache_ts: float = 0.0
# Инициализируем динамические услуги из базы данных
config.SERVICES = db.get_services()
@@ -122,16 +159,29 @@ class VKBotApplication:
"""Супер-админ (определён только в config.SUPER_ADMIN_IDS, меняется в коде)."""
return user_id in getattr(config, "SUPER_ADMIN_IDS", [])
def _runtime_admins(self) -> set:
"""Кэшированный набor админов, добавленных в рантайме (из БД).
Перечитываем из SQLite не чаще раза в ADMINS_CACHE_TTL — _is_admin вызывается
многократно на каждое сообщение, и прямой запрос каждый раз был бы расточителен.
Кэш инвалидируется при добавлении/удалении админа.
"""
now = time.time()
if self._admins_cache is None or now - self._admins_cache_ts > ADMINS_CACHE_TTL:
try:
self._admins_cache = set(db.get_admins())
except Exception:
self._admins_cache = set()
self._admins_cache_ts = now
return self._admins_cache
def _is_admin(self, user_id: int) -> bool:
"""Админ: супер-админ, либо из config.ADMIN_IDS, либо добавленный в рантайме."""
if self._is_super_admin(user_id):
return True
if hasattr(config, "ADMIN_IDS") and user_id in config.ADMIN_IDS:
return True
try:
return user_id in db.get_admins()
except Exception:
return False
return user_id in self._runtime_admins()
def _operator_ids(self) -> list:
"""Список ID операторов (несколько). Фолбэк — один OPERATOR_PEER_ID."""
@@ -223,6 +273,7 @@ class VKBotApplication:
return
new_id = int(arg)
if db.add_admin(new_id):
self._admins_cache = None # инвалидируем кэш админов
self.send_message(user_id, f"✅ Пользователь vk.com/id{new_id} теперь администратор.")
try:
self.send_message(new_id, "🛡 Вам выданы права администратора бота ИКП. "
@@ -243,12 +294,33 @@ class VKBotApplication:
return
rem_id = int(arg)
if db.remove_admin(rem_id):
self._admins_cache = None # инвалидируем кэш админов
self.send_message(user_id, f"✅ Пользователь vk.com/id{rem_id} больше не администратор.")
else:
self.send_message(user_id, f"⚠️ vk.com/id{rem_id} нет среди добавленных из бота "
"(админов из .env и супер-админов отсюда снять нельзя).")
self._send_admins_panel(user_id)
def _cleanup_antiflood(self):
"""Периодически выметает протухшие записи анти-флуда, чтобы словари не росли бесконечно."""
now = time.time()
if now - self._last_antiflood_cleanup < BAN_SEC:
return
self._last_antiflood_cleanup = now
# Сообщения/страйки старше своих окон бесполезны — удаляем пустые ключи
for uid in [u for u, ts in self._msg_times.items()
if not ts or now - ts[-1] > FLOOD_WINDOW_SEC]:
del self._msg_times[uid]
for uid in [u for u, ts in self._flood_strikes.items()
if not ts or now - ts[-1] > BAN_WINDOW_SEC]:
del self._flood_strikes[uid]
for uid in [u for u, t in self._banned_until.items() if t <= now]:
del self._banned_until[uid]
for uid in [u for u, t in self._flood_warned.items() if now - t > FLOOD_WINDOW_SEC]:
del self._flood_warned[uid]
for uid in [u for u, t in self._last_notify.items() if now - t > NOTIFY_COOLDOWN_SEC]:
del self._last_notify[uid]
def _notify_allowed(self, user_id: int) -> bool:
"""Антиспам: не чаще раза в NOTIFY_COOLDOWN_SEC уведомляем оператора об одном пользователе."""
now = time.time()
@@ -263,6 +335,9 @@ class VKBotApplication:
text_lower = text_clean.lower()
attachments = attachments or []
# Периодическая чистка in-memory словарей анти-флуда (не на каждое сообщение)
self._cleanup_antiflood()
# Анти-флуд + временный бан (админы — исключение)
if not self._is_admin(user_id):
now = time.time()
@@ -788,12 +863,19 @@ class VKBotApplication:
# Показываем услуги каруселью (с фолбэком на текст)
self._send_services(user_id, f"✅ Телефон принят!\n\n{self._progress_bar(3)}\nКакое направление вас интересует?\n\n")
def _handle_form_question(self, user_id: int, text: str, user_data: Dict[str, Any]):
"""Обработка выбора услуги с прямым переходом к проверке данных."""
selected_service = None
@staticmethod
def _match_service(text: str) -> Optional[Dict[str, str]]:
"""Сопоставляет введённый текст с услугой: сначала по синонимам дисциплин,
затем по ключу/названию (с очисткой эмодзи). Возвращает svc или None."""
text_lower = text.lower().strip()
# 1. Прямое сопоставление синонимов дисциплин для надежности
# 0. Точное совпадение по номеру/ключу услуги — высший приоритет.
# Иначе нечёткий поиск ниже ловит «2» внутри «counter-strike 2» и ломает выбор по номеру.
for key, svc in config.SERVICES.items():
if text_lower in (key, f"{key}.", f"{key}️⃣"):
return svc
# 1. Прямое сопоставление синонимов дисциплин для надёжности
synonyms = {
"1": ["кс", "cs", "ксго", "кс2", "csgo", "cs2", "контра", "кондра", "counter", "counter-strike", "counter strike"],
"2": ["дота", "dota", "дотку", "дотка", "dota2", "дота2", "доту"],
@@ -801,25 +883,26 @@ class VKBotApplication:
}
for key_opt, words in synonyms.items():
if any(w in text_lower for w in words):
selected_service = config.SERVICES.get(key_opt)
break
# 2. Поиск по названию или ключу (если синонимы не сработали)
if not selected_service:
for key, svc in config.SERVICES.items():
name_clean = svc["name"].lower().strip()
# Убираем эмодзи и спецсимволы для гибкого сравнения
name_no_emoji = re.sub(r"[^\w\s\d]", "", name_clean).strip()
text_no_emoji = re.sub(r"[^\w\s\d]", "", text_lower).strip()
if (text_lower == key or
text_lower == f"{key}." or
text_lower == f"{key}️⃣" or
name_clean in text_lower or
svc = config.SERVICES.get(key_opt)
if svc:
return svc
# 2. Нечёткий поиск по названию (если точный ключ и синонимы не сработали)
for key, svc in config.SERVICES.items():
name_clean = svc["name"].lower().strip()
# Убираем эмодзи и спецсимволы для гибкого сравнения
name_no_emoji = re.sub(r"[^\w\s\d]", "", name_clean).strip()
text_no_emoji = re.sub(r"[^\w\s\d]", "", text_lower).strip()
if (name_clean in text_lower or
text_lower in name_clean or
(name_no_emoji and text_no_emoji and (text_no_emoji in name_no_emoji or name_no_emoji in text_no_emoji))):
selected_service = svc
break
return svc
return None
def _handle_form_question(self, user_id: int, text: str, user_data: Dict[str, Any]):
"""Обработка выбора услуги с прямым переходом к проверке данных."""
selected_service = self._match_service(text)
if selected_service:
# Сохраняем временный выбор в поле question
@@ -1039,35 +1122,7 @@ class VKBotApplication:
def _handle_form_edit_question(self, user_id: int, text: str):
"""Сохранение новой услуги и возврат к проверке."""
selected_service = None
text_lower = text.lower().strip()
# 1. Прямое сопоставление синонимов дисциплин для надежности
synonyms = {
"1": ["кс", "cs", "ксго", "кс2", "csgo", "cs2", "контра", "кондра", "counter", "counter-strike", "counter strike"],
"2": ["дота", "dota", "дотку", "дотка", "dota2", "дота2", "доту"],
"3": ["танки", "танк", "wot", "world of tanks", "ворлд оф танкс", "вот", "tanks"]
}
for key_opt, words in synonyms.items():
if any(w in text_lower for w in words):
selected_service = config.SERVICES.get(key_opt)
break
# 2. Поиск по названию или ключу (если синонимы не сработали)
if not selected_service:
for key, svc in config.SERVICES.items():
name_clean = svc["name"].lower().strip()
name_no_emoji = re.sub(r"[^\w\s\d]", "", name_clean).strip()
text_no_emoji = re.sub(r"[^\w\s\d]", "", text_lower).strip()
if (text_lower == key or
text_lower == f"{key}." or
text_lower == f"{key}️⃣" or
name_clean in text_lower or
text_lower in name_clean or
(name_no_emoji and text_no_emoji and (text_no_emoji in name_no_emoji or name_no_emoji in text_no_emoji))):
selected_service = svc
break
selected_service = self._match_service(text)
if selected_service:
db.update_user_field(user_id, "question", f"{selected_service['name']} ({selected_service['price']})")
@@ -2474,5 +2529,9 @@ class VKBotApplication:
if __name__ == "__main__":
import sys
if not _acquire_single_instance_lock():
logger.error("Уже запущен другой экземпляр бота — выходим.")
sys.exit(0)
bot = VKBotApplication()
bot.run()