#!/usr/bin/env python3 """ Эмулятор пользователей для EventHub. Скрипт устойчив к отсутствию API при старте и во время работы – никогда не падает, переходит в режим ожидания и автоматически восстанавливает работу после появления сервера. """ import os, time, random, requests, logging, json # ------------------------------------------------------------------- # Настройки (можно переопределять через переменные окружения) # ------------------------------------------------------------------- DEBUG = os.getenv("DEBUG", "true").lower() == "true" VERIFY_SSL = os.getenv("VERIFY_SSL", "false").lower() == "true" ADMIN_API_HOST = os.getenv("ADMIN_API_HOST", "http://localhost:8445") CLIENT_API_HOST = os.getenv("CLIENT_API_HOST", "http://localhost:8080") ADMIN_EMAIL = os.getenv("ADMIN_EMAIL", "superadmin@eventhub.local") ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "123456") BOT_PASSWORD = os.getenv("BOT_PASSWORD", "botpass123") MIN_DELAY = float(os.getenv("MIN_DELAY", "0.5")) MAX_DELAY = float(os.getenv("MAX_DELAY", "3.0")) LOOP_FOREVER = os.getenv("LOOP_FOREVER", "true").lower() == "true" BOT_REFRESH_INTERVAL = int(os.getenv("BOT_REFRESH_INTERVAL", "300")) TICKET_CHANCE = float(os.getenv("TICKET_CHANCE", "0.07")) STARTUP_DELAY = int(os.getenv("STARTUP_DELAY", "30")) # ------------------------------------------------------------------- # Логирование # ------------------------------------------------------------------- if not DEBUG: logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s') else: logging.basicConfig(level=logging.DEBUG, format='%(asctime)s [%(levelname)s] %(message)s') logger = logging.getLogger("emulator") # ------------------------------------------------------------------- # Состояние # ------------------------------------------------------------------- bots_cache = [] # [{"email": ..., "token": ...}] admin_token = None last_bot_refresh = 0 # ------------------------------------------------------------------- # Безопасный HTTP-запрос # ------------------------------------------------------------------- def request(method, url, **kwargs): """ Абсолютно безопасный HTTP-запрос. Возвращает объект ответа или None при любой ошибке. """ try: if not VERIFY_SSL: kwargs["verify"] = False if DEBUG: logger.debug(f"--> {method} {url}") resp = requests.request(method, url, **kwargs) if DEBUG: logger.debug(f"<-- {resp.status_code} {resp.url}") try: body = resp.json() body_str = json.dumps(body, indent=2) except: body_str = resp.text[:200] logger.debug(f" Response: {body_str}") return resp except Exception as e: logger.warning(f"Request failed ({method} {url}): {e}") return None # ------------------------------------------------------------------- # Получение административного токена # ------------------------------------------------------------------- def get_admin_token(): global admin_token if admin_token: return admin_token resp = request("POST", f"{ADMIN_API_HOST}/v1/admin/login", json={"email": ADMIN_EMAIL, "password": ADMIN_PASSWORD}, headers={"Content-Type": "application/json"}) if resp is None or resp.status_code != 200: logger.warning("Failed to obtain admin token") return None admin_token = resp.json().get("token") if admin_token: logger.info("Admin token obtained") return admin_token # ------------------------------------------------------------------- # Получение списка email'ов ботов (с фильтром role=bot) # ------------------------------------------------------------------- def fetch_bot_emails(): token = get_admin_token() if not token: return [] resp = request("GET", f"{ADMIN_API_HOST}/v1/admin/users?role=bot&limit=10000", headers={"Authorization": f"Bearer {token}"}) if resp is None or resp.status_code != 200: logger.warning("Failed to fetch bots") return [] users = resp.json() return [u["email"] for u in users if u.get("role") == "bot"] # ------------------------------------------------------------------- # Логин одного бота # ------------------------------------------------------------------- def login_bot(email): resp = request("POST", f"{CLIENT_API_HOST}/v1/login", json={"email": email, "password": BOT_PASSWORD}, headers={"Content-Type": "application/json"}) if resp is None or resp.status_code != 200: raise RuntimeError(f"Bot login failed for {email}") return resp.json()["token"] # ------------------------------------------------------------------- # Обновление кеша ботов # ------------------------------------------------------------------- def refresh_bot_cache(): global bots_cache, last_bot_refresh emails = fetch_bot_emails() if not emails: logger.warning("No bot emails fetched, keeping existing cache") return new_cache = [] for email in emails: try: token = login_bot(email) new_cache.append({"email": email, "token": token}) except Exception as e: logger.warning(f"Could not login bot {email}: {e}") if new_cache: bots_cache = new_cache last_bot_refresh = time.time() logger.info(f"Bot cache refreshed, {len(bots_cache)} bots ready") else: logger.warning("No bots could be logged in, keeping existing cache") # ------------------------------------------------------------------- # Получение случайного бота (с автоматическим обновлением кеша) # ------------------------------------------------------------------- def random_bot(): global bots_cache, last_bot_refresh while True: if not bots_cache or (time.time() - last_bot_refresh > BOT_REFRESH_INTERVAL): refresh_bot_cache() if bots_cache: return random.choice(bots_cache) logger.warning("No bots available, retrying in 10 seconds...") time.sleep(10) # ------------------------------------------------------------------- # Случайная задержка между действиями # ------------------------------------------------------------------- def random_sleep(): time.sleep(random.uniform(MIN_DELAY, MAX_DELAY)) # ------------------------------------------------------------------- # Выполнение случайного действия от имени бота # ------------------------------------------------------------------- def do_random_action(bot): action = random.randint(1, 14) headers = {"Authorization": f"Bearer {bot['token']}", "Content-Type": "application/json"} base = CLIENT_API_HOST try: if action == 1: resp = request("POST", f"{base}/v1/calendars", json={"title": f"Cal-{random.randint(1,1000)}", "confirmation": "auto"}, headers=headers) if resp and resp.status_code == 201: logger.debug(f"Bot {bot['email']} created calendar {resp.json()['id']}") elif action == 2: request("GET", f"{base}/v1/calendars", headers=headers) elif action == 3: resp_cal = request("GET", f"{base}/v1/calendars", headers=headers) if resp_cal and resp_cal.status_code == 200 and resp_cal.json(): cal = random.choice(resp_cal.json()) request("POST", f"{base}/v1/calendars/{cal['id']}/events", json={"title": f"Event-{random.randint(1,1000)}", "start_time": "2027-01-01T10:00:00Z", "duration": 60}, headers=headers) elif action == 4: request("GET", f"{base}/v1/search?q=test&limit=5", headers=headers) elif action == 5: resp_ev = request("GET", f"{base}/v1/search?type=event&limit=20", headers=headers) if resp_ev and resp_ev.status_code == 200: results = resp_ev.json().get("results", {}) events = results.get("events", []) if events: ev = random.choice(events) request("POST", f"{base}/v1/events/{ev['id']}/bookings", json={}, headers=headers) elif action == 6: resp_book = request("GET", f"{base}/v1/user/bookings", headers=headers) if resp_book and resp_book.status_code == 200 and resp_book.json(): booking = random.choice(resp_book.json()) request("POST", f"{base}/v1/reviews", json={"target_type": "event", "target_id": booking["event_id"], "rating": random.randint(1,5), "comment": "Nice!"}, headers=headers) elif action == 7: resp_ev = request("GET", f"{base}/v1/search?type=event&limit=20", headers=headers) if resp_ev and resp_ev.status_code == 200: results = resp_ev.json().get("results", {}) events = results.get("events", []) if events: ev = random.choice(events) request("POST", f"{base}/v1/reports", json={"target_type": "event", "target_id": ev["id"], "reason": "Test"}, headers=headers) elif action == 8: if random.random() < TICKET_CHANCE: request("POST", f"{base}/v1/tickets", json={"error_message": "Emulated error", "stacktrace": "line 1"}, headers=headers) elif action == 9: request("POST", f"{base}/v1/subscription", json={"action": "start_trial"}, headers=headers) elif action == 10: request("GET", f"{base}/v1/user/me", headers=headers) elif action == 11: request("GET", f"{base}/v1/user/bookings", headers=headers) elif action == 12: request("GET", f"{base}/v1/user/reviews", headers=headers) elif action == 13: resp_cal = request("GET", f"{base}/v1/calendars", headers=headers) if resp_cal and resp_cal.status_code == 200 and resp_cal.json(): cal = random.choice(resp_cal.json()) request("GET", f"{base}/v1/calendars/{cal['id']}/view?month=2026-06", headers=headers) elif action == 14: request("POST", f"{base}/v1/refresh", json={"refresh_token": "dummy"}, headers=headers) except Exception as e: logger.error(f"Action {action} failed for {bot['email']}: {e}") # ------------------------------------------------------------------- # Главный цикл # ------------------------------------------------------------------- def main(): logger.info(f"Starting user emulation (startup delay: {STARTUP_DELAY}s)") time.sleep(STARTUP_DELAY) # Бесконечно ждём, пока не появятся боты while True: refresh_bot_cache() if bots_cache: break logger.info("Waiting 10 seconds before retry...") time.sleep(10) while LOOP_FOREVER: try: bot = random_bot() do_random_action(bot) except Exception as e: logger.error(f"Unexpected error in main loop: {e}") random_sleep() logger.info("Emulation finished") if __name__ == "__main__": main()