From e51437f08e9369582cf42382c3e38960a7a8905e Mon Sep 17 00:00:00 2001 From: Yuriy Yuriev Date: Thu, 21 May 2026 22:51:09 +0700 Subject: [PATCH] fix some --- handlers/commands.py | 14 ++++++-- main.py | 3 ++ managers/storage.py | 3 ++ services/browser_service.py | 57 ++++++++++++++++---------------- services/socks5_to_http_proxy.py | 31 +++++++++++------ 5 files changed, 66 insertions(+), 42 deletions(-) diff --git a/handlers/commands.py b/handlers/commands.py index 11d56a0..cdda664 100644 --- a/handlers/commands.py +++ b/handlers/commands.py @@ -1124,8 +1124,9 @@ class BotInterface: ) else: from types import SimpleNamespace + chat_id = params.chat_id or callback.message.chat.id msg_mock = SimpleNamespace( - chat=SimpleNamespace(id=params.chat_id), + chat=SimpleNamespace(id=chat_id), bot=callback.message.bot, ) await interface.background_tasks.cancel_task(tid) @@ -1133,7 +1134,7 @@ class BotInterface: task_id=tid, coro=interface._run_twitch(tid, params, msg_mock), task_type="twitch_irc", - metadata={"type": "twitch_irc", "channel": params.channel, "chat_id": params.chat_id}, + metadata={"type": "twitch_irc", "channel": params.channel, "chat_id": chat_id}, ) if success: @@ -2062,11 +2063,14 @@ class BotInterface: from services.irc_service import TwitchIRCClient irc = TwitchIRCClient(params.channel, params.target_username) + url_tasks: set = set() async def on_url(url, username): if params.stream_offline: - return # не обрабатываем ссылки пока стрим оффлайн + return task = asyncio.create_task(self._process_url(url, username, params, message)) + url_tasks.add(task) + task.add_done_callback(url_tasks.discard) task.add_done_callback( lambda t: logger.error(f"URL processing failed: {t.exception()}") if not t.cancelled() and t.exception() else None ) @@ -2098,6 +2102,10 @@ class BotInterface: logger.error(f"IRC error [{params.channel}]: {e}") finally: watcher.cancel() + for t in list(url_tasks): + t.cancel() + if url_tasks: + await asyncio.gather(*url_tasks, return_exceptions=True) await irc.disconnect() if self.storage: await self.storage.save_task(task_id, params) diff --git a/main.py b/main.py index ad95026..6e4bb49 100644 --- a/main.py +++ b/main.py @@ -106,7 +106,10 @@ class BotApplication: continue # Активные задачи ставим на паузу и перезапускаем + # Если стрим был оффлайн — сохраняем этот статус params.paused = True + if params.stream_offline: + logger.info(f"Task {task_id}: stream was offline, keeping paused") await self.interface.task_manager.add_task(task_id, params) if params.task_type == "twitch_irc" and params.chat_id: diff --git a/managers/storage.py b/managers/storage.py index d0ff51e..630e41b 100644 --- a/managers/storage.py +++ b/managers/storage.py @@ -53,6 +53,8 @@ class TaskStorage: "min_ctr": params.min_ctr, "max_ctr": params.max_ctr, "started_at": params.started_at.isoformat() if params.started_at else None, + "stream_offline": params.stream_offline, + # pending_visits и total_planned_visits — runtime, не сохраняем } def _dict_to_params(self, data: dict) -> TaskParams: @@ -85,6 +87,7 @@ class TaskStorage: max_series=data.get("max_series", data.get("visits_per_link", 1)), min_ctr=data.get("min_ctr", 0.8), max_ctr=data.get("max_ctr", 1.0), + stream_offline=data.get("stream_offline", False), ) if data.get("started_at"): try: diff --git a/services/browser_service.py b/services/browser_service.py index 0d1b39d..52ef2d6 100644 --- a/services/browser_service.py +++ b/services/browser_service.py @@ -21,16 +21,9 @@ from services.socks5_to_http_proxy import Socks5ProxyPool logger = logging.getLogger(__name__) -_current_proxy_ip: Optional[str] = None - - -def _get_real_ip() -> Optional[str]: - return _current_proxy_ip - - def _set_real_ip(ip: Optional[str]) -> None: - global _current_proxy_ip - _current_proxy_ip = ip + import threading + threading.current_thread()._proxy_ip = ip def _patch_camoufox(): @@ -63,7 +56,9 @@ def _patch_camoufox(): _orig_ip = _ci.public_ip def _patched_ip(proxy_string: str = None) -> str: - ip = _get_real_ip() + # Вызывается из thread pool — используем thread-local атрибут + import threading + ip = getattr(threading.current_thread(), "_proxy_ip", None) if ip: logger.debug(f"public_ip() intercepted → returning {ip}") return ip @@ -140,24 +135,30 @@ class BrowserService: proxy_locale = None - if proxy and proxy.proxy_type in [ProxyType.SOCKS5, ProxyType.SOCKS4]: - proxy_config = await self.socks5_pool.get_proxy_config(proxy) - use_socks5 = True - real_ip = proxy.id.split("://")[-1].split(":")[0] - _set_real_ip(real_ip) - proxy_locale = _locale_for_ip(real_ip) - has_auth = bool(proxy.login and proxy.password) - logger.info(f"Using SOCKS5 tunnel for: {proxy.id} auth={'yes' if has_auth else 'NO'} locale={proxy_locale}") - elif proxy: - proxy_config = proxy.proxy_config - real_ip = proxy.id.split("://")[-1].split(":")[0] - _set_real_ip(real_ip) - proxy_locale = _locale_for_ip(real_ip) - logger.info(f"Using HTTP proxy: {proxy.id} locale={proxy_locale}") - else: - proxy_config = None - _set_real_ip(None) - logger.info("Direct connection") + try: + if proxy and proxy.proxy_type in [ProxyType.SOCKS5, ProxyType.SOCKS4]: + proxy_config = await self.socks5_pool.get_proxy_config(proxy) + use_socks5 = True + real_ip = proxy.id.split("://")[-1].split(":")[0] + _set_real_ip(real_ip) + proxy_locale = _locale_for_ip(real_ip) + has_auth = bool(proxy.login and proxy.password) + logger.info(f"Using SOCKS5 tunnel for: {proxy.id} auth={'yes' if has_auth else 'NO'} locale={proxy_locale}") + elif proxy: + proxy_config = proxy.proxy_config + real_ip = proxy.id.split("://")[-1].split(":")[0] + _set_real_ip(real_ip) + proxy_locale = _locale_for_ip(real_ip) + logger.info(f"Using HTTP proxy: {proxy.id} locale={proxy_locale}") + else: + proxy_config = None + _set_real_ip(None) + logger.info("Direct connection") + except asyncio.CancelledError: + # Гарантируем освобождение прокси даже при отмене задачи + if proxy: + await self.proxy_manager.release_proxy(proxy, success=False) + raise if proxy_config: camoufox_kwargs = dict( diff --git a/services/socks5_to_http_proxy.py b/services/socks5_to_http_proxy.py index 021e62e..2170f35 100644 --- a/services/socks5_to_http_proxy.py +++ b/services/socks5_to_http_proxy.py @@ -372,20 +372,29 @@ class Socks5ProxyPool: async def get_proxy_config(self, socks5_proxy) -> dict: """Получает или создает конфигурацию HTTP прокси для SOCKS5.""" key = socks5_proxy.id - + + # Быстрая проверка под lock'ом + async with self._lock: + if key in self._proxies: + self._last_used[key] = asyncio.get_running_loop().time() + return self._proxies[key].proxy_config_for_browser + + # proxy.start() — медленная операция, запускаем вне lock'а + proxy = Socks5ToHttpProxy( + socks5_host=socks5_proxy.ip, + socks5_port=int(socks5_proxy.port), + username=socks5_proxy.login, + password=socks5_proxy.password + ) + await proxy.start() + + # Добавляем под lock'ом (двойная проверка — другая задача могла успеть) async with self._lock: if key not in self._proxies: - proxy = Socks5ToHttpProxy( - socks5_host=socks5_proxy.ip, - socks5_port=int(socks5_proxy.port), - username=socks5_proxy.login, - password=socks5_proxy.password - ) - await proxy.start() self._proxies[key] = proxy - - loop = asyncio.get_running_loop() - self._last_used[key] = loop.time() + else: + await proxy.stop() # уже создан другой задачей + self._last_used[key] = asyncio.get_running_loop().time() return self._proxies[key].proxy_config_for_browser async def release(self, socks5_proxy):