This commit is contained in:
Yuriy Yuriev
2026-05-21 22:51:09 +07:00
parent 26038c6027
commit e51437f08e
5 changed files with 66 additions and 42 deletions
+11 -3
View File
@@ -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)
+3
View File
@@ -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:
+3
View File
@@ -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:
+29 -28
View File
@@ -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(
+19 -10
View File
@@ -373,19 +373,28 @@ class Socks5ProxyPool:
"""Получает или создает конфигурацию 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):